Skip to content

Commit a035606

Browse files
pidarpedpidarped
committed
TCP test cases
Test script for tests defined in CHIP-Specifications/chip-test-plans#4243
1 parent 138fb4f commit a035606

File tree

1 file changed

+156
-0
lines changed

1 file changed

+156
-0
lines changed

‎src/python_testing/TCP_Tests.py

+156
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,156 @@
1+
#
2+
# Copyright (c) 2023 Project CHIP Authors
3+
# All rights reserved.
4+
#
5+
# Licensed under the Apache License, Version 2.0 (the "License");
6+
# you may not use this file except in compliance with the License.
7+
# You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing, software
12+
# distributed under the License is distributed on an "AS IS" BASIS,
13+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
# See the License for the specific language governing permissions and
15+
# limitations under the License.
16+
#
17+
18+
# test-runner-runs: run1
19+
# test-runner-run/run1/app: ${ALL_CLUSTERS_APP}
20+
# test-runner-run/run1/factoryreset: True
21+
# test-runner-run/run1/quiet: True
22+
# test-runner-run/run1/app-args: --discriminator 1234 --KVS kvs1 --trace-to json:${TRACE_APP}.json
23+
# test-runner-run/run1/script-args: --storage-path admin_storage.json --commissioning-method on-network --discriminator 1234 --passcode 20202021 --trace-to json:${TRACE_TEST_JSON}.json --trace-to perfetto:${TRACE_TEST_PERFETTO}.perfetto
24+
25+
import inspect
26+
import logging
27+
import random
28+
from dataclasses import dataclass
29+
30+
import chip.clusters as Clusters
31+
import chip.discovery as Discovery
32+
from chip import ChipUtility
33+
from chip import ChipDeviceCtrl
34+
from chip.exceptions import ChipStackError
35+
from chip.interaction_model import InteractionModelError, Status
36+
from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main, type_matches
37+
from mobly import asserts
38+
39+
40+
class TCP_Tests(MatterBaseTest):
41+
42+
@async_test_body
43+
async def test_TCPConnectionEstablishment(self):
44+
45+
try:
46+
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
47+
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
48+
except TimeoutError:
49+
asserts.fail("Unable to establish a CASE session over TCP to the device")
50+
asserts.assert_equal(device.isSessionOverTCPConnection, True, "Session does not have associated TCP connection")
51+
asserts.assert_equal(device.isActiveSession, True, "Large Payload Session should be active over TCP connection")
52+
53+
@async_test_body
54+
async def test_LargePayloadSessionEstablishment(self):
55+
56+
try:
57+
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
58+
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
59+
except TimeoutError:
60+
asserts.fail("Unable to establish a CASE session over TCP to the device")
61+
asserts.assert_equal(device.sessionAllowsLargePayload, True, "Session does not have associated TCP connection")
62+
63+
@async_test_body
64+
async def test_SessionInactiveAfterTCPDisconnect(self):
65+
66+
try:
67+
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
68+
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
69+
except TimeoutError:
70+
asserts.fail("Unable to establish a CASE session over TCP to the device")
71+
asserts.assert_equal(device.isSessionOverTCPConnection, True, "Session does not have associated TCP connection")
72+
73+
device.closeTCPConnectionWithPeer()
74+
asserts.assert_equal(device.isActiveSession, False, "Large Payload Session should not be active after TCP connection closure")
75+
76+
@async_test_body
77+
async def test_TCPConnectDisconnectThenConnectAgain(self):
78+
79+
try:
80+
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
81+
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
82+
except TimeoutError:
83+
asserts.fail("Unable to establish a CASE session over TCP to the device")
84+
asserts.assert_equal(device.isSessionOverTCPConnection, True, "Session does not have associated TCP connection")
85+
86+
device.closeTCPConnectionWithPeer()
87+
asserts.assert_equal(device.isActiveSession, False, "Large Payload Session should not be active after TCP connection closure")
88+
89+
# Connect again
90+
try:
91+
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
92+
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
93+
except TimeoutError:
94+
asserts.fail("Unable to establish a CASE session over TCP to the device")
95+
asserts.assert_equal(device.isSessionOverTCPConnection, True, "Session does not have associated TCP connection")
96+
asserts.assert_equal(device.isActiveSession, True, "Large Payload Session should be active over TCP connection")
97+
98+
@async_test_body
99+
async def test_OnOffToggleCommandOverTCPSession(self):
100+
101+
try:
102+
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
103+
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
104+
except TimeoutError:
105+
asserts.fail("Unable to establish a CASE session over TCP to the device")
106+
asserts.assert_equal(device.isSessionOverTCPConnection, True, "Session does not have associated TCP connection")
107+
asserts.assert_equal(device.isActiveSession, True, "Large Payload Session should be active over TCP connection")
108+
asserts.assert_equal(device.sessionAllowsLargePayload, True, "Session does not have associated TCP connection")
109+
110+
commands = Clusters.Objects.OnOff.Commands
111+
try:
112+
await self.send_single_cmd(cmd=commands.Toggle(), endpoint=1,
113+
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
114+
except InteractionModelError as e:
115+
asserts.assert_equal(e.status, Status.Success, "Unexpected error returned")
116+
117+
@async_test_body
118+
async def test_WildCardReadOverTCPSession(self):
119+
120+
try:
121+
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
122+
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
123+
except TimeoutError:
124+
asserts.fail("Unable to establish a CASE session over TCP to the device")
125+
asserts.assert_equal(device.isSessionOverTCPConnection, True, "Session does not have associated TCP connection")
126+
asserts.assert_equal(device.isActiveSession, True, "Large Payload Session should be active over TCP connection")
127+
asserts.assert_equal(device.sessionAllowsLargePayload, True, "Session does not have associated TCP connection")
128+
129+
try:
130+
wildcard_read = await self.default_controller.Read(self.dut_node_id, [()],
131+
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
132+
except InteractionModelError as e:
133+
asserts.assert_equal(e.status, Status.Success, "Unexpected error returned")
134+
135+
@async_test_body
136+
async def test_UseTCPSessionIfAvailableForMRPInteraction(self):
137+
138+
try:
139+
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
140+
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
141+
except TimeoutError:
142+
asserts.fail("Unable to establish a CASE session over TCP to the device")
143+
asserts.assert_equal(device.isSessionOverTCPConnection, True, "Session does not have associated TCP connection")
144+
asserts.assert_equal(device.isActiveSession, True, "Large Payload Session should be active over TCP connection")
145+
asserts.assert_equal(device.sessionAllowsLargePayload, True, "Session does not have associated TCP connection")
146+
147+
commands = Clusters.Objects.OnOff.Commands
148+
try:
149+
await self.send_single_cmd(cmd=commands.Toggle(), endpoint=1,
150+
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.MRP_OR_TCP_PAYLOAD)
151+
except InteractionModelError as e:
152+
asserts.assert_equal(e.status, Status.Success, "Unexpected error returned")
153+
154+
155+
if __name__ == "__main__":
156+
default_matter_test_main()

0 commit comments

Comments
 (0)