Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Cherry pick PR #37397 to V1.4 branch #37443

Merged
merged 2 commits into from
Feb 7, 2025
Merged

Cherry pick PR #37397 to V1.4 branch #37443

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7515363
Add test steps logging for TCP Tests
pidarped Feb 4, 2025
128959e
Fix naming of Test steps.
pidarped Feb 5, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
140 changes: 132 additions & 8 deletions src/python_testing/TCP_Tests.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@
import chip.clusters as Clusters
from chip import ChipDeviceCtrl
from chip.interaction_model import InteractionModelError
from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main
from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main
from mobly import asserts


Expand All @@ -48,141 +48,265 @@ async def teardown_test(self):
await self.send_single_cmd(cmd=cmd, endpoint=0,
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)

def pics_SC_8_1(self):
def pics_TC_SC_8_1(self):
return ['MCORE.SC.TCP']

def steps_TC_SC_8_1(self) -> list[TestStep]:
steps = [
TestStep(1, "Commissioning, already done", is_commissioning=True),
TestStep(2, "TH initiates a CASE session establishment with DUT, requesting a session supporting large payloads."),
TestStep(3, "Verifying that a session is set up with an underlying TCP connection established with DUT."),
]
return steps

# TCP Connection Establishment
@async_test_body
async def test_TC_SC_8_1(self):

self.step(1)
try:
self.step(2)
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
except TimeoutError:
asserts.fail("Unable to establish a CASE session over TCP to the device")

self.step(3)
asserts.assert_equal(device.isSessionOverTCPConnection, True, "Session does not have associated TCP connection")
asserts.assert_equal(device.isActiveSession, True, "Large Payload Session should be active over TCP connection")

def pics_SC_8_2(self):
def pics_TC_SC_8_2(self):
return ['MCORE.SC.TCP']

def steps_TC_SC_8_2(self) -> list[TestStep]:
steps = [
TestStep(1, "Commissioning, already done", is_commissioning=True),
TestStep(2, "TH initiates a CASE session establishment with DUT, requesting a session supporting large payloads."),
TestStep(3, "Verifying that the session established with DUT allows large payloads."),
]
return steps

# Large Payload Session Establishment
@async_test_body
async def test_TC_SC_8_2(self):

self.step(1)
try:
self.step(2)
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
except TimeoutError:
asserts.fail("Unable to establish a CASE session over TCP to the device")

self.step(3)
asserts.assert_equal(device.sessionAllowsLargePayload, True, "Session does not have associated TCP connection")

def pics_SC_8_3(self):
def pics_TC_SC_8_3(self):
return ['MCORE.SC.TCP']

def steps_TC_SC_8_3(self) -> list[TestStep]:
steps = [
TestStep(1, "Commissioning, already done", is_commissioning=True),
TestStep(2, "TH initiates a CASE session establishment with DUT, requesting a session supporting large payloads."),
TestStep(3, "Verifying that a session is set up with an underlying TCP connection established with DUT."),
TestStep(4, "TH closes the TCP connection with DUT"),
TestStep(5, "Verifying that the secure session with DUT is inactive."),
]
return steps

# Session Inactive After TCP Disconnect
@async_test_body
async def test_TC_SC_8_3(self):

self.step(1)
try:
self.step(2)
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
except TimeoutError:
asserts.fail("Unable to establish a CASE session over TCP to the device")

self.step(3)
asserts.assert_equal(device.isSessionOverTCPConnection, True, "Session does not have associated TCP connection")

self.step(4)
device.closeTCPConnectionWithPeer()

self.step(5)
asserts.assert_equal(device.isActiveSession, False,
"Large Payload Session should not be active after TCP connection closure")

def pics_SC_8_4(self):
def pics_TC_SC_8_4(self):
return ['MCORE.SC.TCP']

def steps_TC_SC_8_4(self) -> list[TestStep]:
steps = [
TestStep(1, "Commissioning, already done", is_commissioning=True),
TestStep(2, "TH initiates a CASE session establishment with DUT, requesting a session supporting large payloads."),
TestStep(3, "Verifying that a session is set up with an underlying TCP connection established with DUT."),
TestStep(4, "TH closes the TCP connection with DUT"),
TestStep(5, "Verifyng that the secure session with DUT is inactive."),
TestStep(6, "TH re-initiates CASE session establishment over TCP with DUT"),
TestStep(7, "Verifying that a session is set up with an underlying TCP connection established with DUT."),
TestStep(8, "Verifying that the large-payload secure session with DUT is active."),
]
return steps

# TCP Connect, Disconnect, Then Connect Again
@async_test_body
async def test_TC_SC_8_4(self):

self.step(1)
try:
self.step(2)
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
except TimeoutError:
asserts.fail("Unable to establish a CASE session over TCP to the device")

self.step(3)
asserts.assert_equal(device.isSessionOverTCPConnection, True, "Session does not have associated TCP connection")

self.step(4)
device.closeTCPConnectionWithPeer()

self.step(5)
asserts.assert_equal(device.isActiveSession, False,
"Large Payload Session should not be active after TCP connection closure")

# Connect again
try:
self.step(6)
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
except TimeoutError:
asserts.fail("Unable to establish a CASE session over TCP to the device")

self.step(7)
asserts.assert_equal(device.isSessionOverTCPConnection, True, "Session does not have associated TCP connection")

self.step(8)
asserts.assert_equal(device.isActiveSession, True, "Large Payload Session should be active over TCP connection")

def pics_SC_8_5(self):
def pics_TC_SC_8_5(self):
return ['MCORE.SC.TCP']

def steps_TC_SC_8_5(self) -> list[TestStep]:
steps = [
TestStep(1, "Commissioning, already done", is_commissioning=True),
TestStep(2, "TH initiates a CASE session establishment with DUT, requesting a session supporting large payloads."),
TestStep(3, "Verifying that a session is set up with an underlying TCP connection established with DUT."),
TestStep(4, "Verifying that the large-payload secure session with DUT is active."),
TestStep(5, "TH initiates an InvokeCommandRequest with DUT over the established session."),
TestStep(6, "Verifying successful invocation with DUT over the established session without any error."),
]
return steps

@async_test_body
async def test_TC_SC_8_5(self):

self.step(1)
try:
self.step(2)
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
except TimeoutError:
asserts.fail("Unable to establish a CASE session over TCP to the device")

self.step(3)
asserts.assert_equal(device.isSessionOverTCPConnection, True, "Session does not have associated TCP connection")

self.step(4)
asserts.assert_equal(device.isActiveSession, True, "Large Payload Session should be active over TCP connection")
asserts.assert_equal(device.sessionAllowsLargePayload, True, "Session does not have associated TCP connection")

try:
self.step(5)
await self.send_arm_cmd(ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
except InteractionModelError:
asserts.fail("Unexpected error returned by DUT")
self.step(6)

def pics_SC_8_6(self):
def pics_TC_SC_8_6(self):
return ['MCORE.SC.TCP']

def steps_TC_SC_8_6(self) -> list[TestStep]:
steps = [
TestStep(1, "Commissioning, already done", is_commissioning=True),
TestStep(2, "TH initiates a CASE session establishment with DUT, requesting a session supporting large payloads."),
TestStep(3, "Verifying that a session is set up with an underlying TCP connection established with DUT."),
TestStep(4, "Verifying that the large-payload secure session with DUT is active."),
TestStep(5, "TH initiates a Read of all attributes of all clusters of DUT."),
TestStep(6, "Verifying wildcard read was successful with DUT over the established session without any error."),
]
return steps

# WildCard Read Over TCP Session
@async_test_body
async def test_TC_SC_8_6(self):

self.step(1)
try:
self.step(2)
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
except TimeoutError:
asserts.fail("Unable to establish a CASE session over TCP to the device")

self.step(3)
asserts.assert_equal(device.isSessionOverTCPConnection, True, "Session does not have associated TCP connection")

self.step(4)
asserts.assert_equal(device.isActiveSession, True, "Large Payload Session should be active over TCP connection")
asserts.assert_equal(device.sessionAllowsLargePayload, True, "Session does not have associated TCP connection")

try:
self.step(5)
await self.default_controller.Read(self.dut_node_id, [()], payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
except InteractionModelError:
asserts.fail("Unexpected error returned by DUT")
self.step(6)

def pics_SC_8_7(self):
def pics_TC_SC_8_7(self):
return ['MCORE.SC.TCP']

def steps_TC_SC_8_7(self) -> list[TestStep]:
steps = [
TestStep(1, "Commissioning, already done", is_commissioning=True),
TestStep(2, "TH initiates a CASE session establishment with DUT, requesting a session supporting large payloads."),
TestStep(3, "Verifying that a session is set up with an underlying TCP connection established with DUT."),
TestStep(4, "Verifying that the large-payload secure session with DUT is active."),
TestStep(5, "TH initiates a regularly-sized InvokeCommandRequest with DUT, specifying that either a MRP or TCP-based session is usable."),
TestStep(6, "Verifying successful invocation with DUT over the established TCP-based session without any error."),
]
return steps

# Use TCP Session If Available For MRP Interaction
@async_test_body
async def test_TC_SC_8_7(self):

self.step(1)

try:
self.step(2)
device = await self.default_controller.GetConnectedDevice(nodeid=self.dut_node_id, allowPASE=False, timeoutMs=1000,
payloadCapability=ChipDeviceCtrl.TransportPayloadCapability.LARGE_PAYLOAD)
except TimeoutError:
asserts.fail("Unable to establish a CASE session over TCP to the device")

self.step(3)
asserts.assert_equal(device.isSessionOverTCPConnection, True, "Session does not have associated TCP connection")

self.step(4)
asserts.assert_equal(device.isActiveSession, True, "Large Payload Session should be active over TCP connection")
asserts.assert_equal(device.sessionAllowsLargePayload, True, "Session does not have associated TCP connection")

try:
self.step(5)
self.send_arm_cmd(ChipDeviceCtrl.TransportPayloadCapability.MRP_OR_TCP_PAYLOAD)
except InteractionModelError:
asserts.fail("Unexpected error returned by DUT")
self.step(6)


if __name__ == "__main__":
Expand Down
Loading