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

extract runner.py from matter_testing #37949

Draft
wants to merge 9 commits into
base: master
Choose a base branch
from
+335 −266
Draft

extract runner.py from matter_testing #37949

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
a9d6d85
initial runner decouple
asirko-soft Mar 10, 2025
347c5ab
lint code
asirko-soft Mar 10, 2025
13d6fbb
Merge branch 'master' into osirko_extract_runner
asirko-soft Mar 11, 2025
d544079
Merge branch 'master' into osirko_extract_runner
asirko-soft Mar 11, 2025
91dd73f
expose test class to module globals
asirko-soft Mar 12, 2025
694e0cd
Merge branch 'master' into osirko_extract_runner
asirko-soft Mar 12, 2025
6931d02
restyled
asirko-soft Mar 12, 2025
c9e063d
merge master
asirko-soft Mar 12, 2025
4a35fc2
move assignment inside __main__
asirko-soft Mar 12, 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
59 changes: 40 additions & 19 deletions src/python_testing/TC_CADMIN_1_3_4.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,13 +77,15 @@ async def get_txt_record(self):

async def write_nl_attr(self, th: ChipDeviceCtrl, attr_val: object):
result = await th.WriteAttribute(nodeid=self.dut_node_id, attributes=[(0, attr_val)])
asserts.assert_equal(result[0].Status, Status.Success, f"{th} node label write failed")
asserts.assert_equal(
result[0].Status, Status.Success, f"{th} node label write failed")

async def read_nl_attr(self, th: ChipDeviceCtrl, attr_val: object):
try:
await th.ReadAttribute(nodeid=self.dut_node_id, attributes=[(0, attr_val)])
except Exception as e:
asserts.assert_equal(e.err, "Received error message from read attribute attempt")
asserts.assert_equal(
e.err, "Received error message from read attribute attempt")
self.print_step(0, e)

async def read_currentfabricindex(self, th: ChipDeviceCtrl) -> int:
Expand Down Expand Up @@ -124,7 +126,8 @@ async def combined_commission_val_steps(self, commission_type: str):
)

elif commission_type == "BCM":
obcCmd = Clusters.AdministratorCommissioning.Commands.OpenBasicCommissioningWindow(180)
obcCmd = Clusters.AdministratorCommissioning.Commands.OpenBasicCommissioningWindow(
180)
await self.th1.SendCommand(nodeid=self.dut_node_id, endpoint=0, payload=obcCmd, timedRequestTimeoutMs=6000)

else:
Expand All @@ -134,7 +137,8 @@ async def combined_commission_val_steps(self, commission_type: str):
services = await self.get_txt_record()
expected_cm_value = "2" if commission_type == "ECM" else "1"
if services.txt_record['CM'] != expected_cm_value:
asserts.fail(f"Expected cm record value {expected_cm_value}, but found {services.txt_record['CM']}")
asserts.fail(
f"Expected cm record value {expected_cm_value}, but found {services.txt_record['CM']}")

self.step("3c")
BI_cluster = Clusters.BasicInformation
Expand All @@ -145,8 +149,10 @@ async def combined_commission_val_steps(self, commission_type: str):
self.step(4)
# Establishing TH2
th2_certificate_authority = self.certificate_authority_manager.NewCertificateAuthority()
th2_fabric_admin = th2_certificate_authority.NewFabricAdmin(vendorId=0xFFF1, fabricId=self.th1.fabricId + 1)
self.th2 = th2_fabric_admin.NewController(nodeId=2, useTestCommissioner=True)
th2_fabric_admin = th2_certificate_authority.NewFabricAdmin(
vendorId=0xFFF1, fabricId=self.th1.fabricId + 1)
self.th2 = th2_fabric_admin.NewController(
nodeId=2, useTestCommissioner=True)

if commission_type == "ECM":
await self.th2.CommissionOnNetwork(
Expand All @@ -158,7 +164,8 @@ async def combined_commission_val_steps(self, commission_type: str):
else:
setupPayloadInfo = self.get_setup_payload_info()
if not setupPayloadInfo:
asserts.fail("Setup payload info is required for basic commissioning.")
asserts.fail(
"Setup payload info is required for basic commissioning.")
await self.th2.CommissionOnNetwork(
nodeId=self.dut_node_id,
setupPinCode=setupPayloadInfo[0].passcode,
Expand All @@ -174,17 +181,21 @@ async def combined_commission_val_steps(self, commission_type: str):
th1_cam_rcac = TLVReader(base64.b64decode(
self.certificate_authority_manager.activeCaList[0]._persistentStorage._jsonData["sdk-config"]["f/1/r"])).get()["Any"][9]
if th1_fabric_info[0].rootPublicKey != th1_cam_rcac:
asserts.fail("Public keys from fabric and certs for TH1 are not the same.")
asserts.fail(
"Public keys from fabric and certs for TH1 are not the same.")
if th1_fabric_info[0].nodeID != self.dut_node_id:
asserts.fail("DUT node ID from fabric does not equal DUT node ID for TH1 during commissioning.")
asserts.fail(
"DUT node ID from fabric does not equal DUT node ID for TH1 during commissioning.")

self.step(6)
# TH_CR2 reads the Fabrics attribute
th2_fabric_info = await self.get_fabrics(th=self.th2)
if th2_fabric_info[0].rootPublicKey != th2_rcac_decoded:
asserts.fail("Public keys from fabric and certs for TH2 are not the same.")
asserts.fail(
"Public keys from fabric and certs for TH2 are not the same.")
if th2_fabric_info[0].nodeID != self.dut_node_id:
asserts.fail("DUT node ID from fabric does not equal DUT node ID for TH2 during commissioning.")
asserts.fail(
"DUT node ID from fabric does not equal DUT node ID for TH2 during commissioning.")

if commission_type == "ECM":
self.step(7)
Expand All @@ -195,7 +206,8 @@ async def combined_commission_val_steps(self, commission_type: str):
self.step(8)
# TH_CR2 writes and reads the Basic Information Cluster’s NodeLabel mandatory attribute of DUT_CE
val = await self.read_nl_attr(th=self.th2, attr_val=self.nl_attribute)
self.print_step("basic information cluster node label attr value", val)
self.print_step(
"basic information cluster node label attr value", val)
await self.write_nl_attr(th=self.th2, attr_val=self.nl_attribute)
await self.read_nl_attr(th=self.th2, attr_val=self.nl_attribute)

Expand All @@ -219,7 +231,8 @@ async def combined_commission_val_steps(self, commission_type: str):
outer_key = list(window_status.keys())[0]
inner_key = list(window_status[outer_key].keys())[1]
if window_status[outer_key][inner_key] != Clusters.AdministratorCommissioning.Enums.CommissioningWindowStatusEnum.kWindowNotOpen:
asserts.fail("Commissioning window is expected to be closed, but was found to be open")
asserts.fail(
"Commissioning window is expected to be closed, but was found to be open")

self.step(12)
# TH_CR2 opens a commissioning window on DUT_CE using ECM
Expand Down Expand Up @@ -258,19 +271,22 @@ async def combined_commission_val_steps(self, commission_type: str):
outer_key = list(th2_idx.keys())[0]
inner_key = list(th2_idx[outer_key].keys())[0]
attribute_key = list(th2_idx[outer_key][inner_key].keys())[1]
removeFabricCmd = Clusters.OperationalCredentials.Commands.RemoveFabric(th2_idx[outer_key][inner_key][attribute_key])
removeFabricCmd = Clusters.OperationalCredentials.Commands.RemoveFabric(
th2_idx[outer_key][inner_key][attribute_key])
await self.th1.SendCommand(nodeid=self.dut_node_id, endpoint=0, payload=removeFabricCmd)

def pics_TC_CADMIN_1_3(self) -> list[str]:
return ["CADMIN.S"]

def steps_TC_CADMIN_1_3(self) -> list[TestStep]:
return [
TestStep(1, "TH_CR1 starts a commissioning process with DUT_CE", is_commissioning=True),
TestStep(
1, "TH_CR1 starts a commissioning process with DUT_CE", is_commissioning=True),
TestStep(2, "TH_CR1 reads the BasicCommissioningInfo attribute from the General Commissioning cluster and saves the MaxCumulativeFailsafeSeconds field as max_window_duration."),
TestStep("3a", "TH_CR1 opens a commissioning window on DUT_CE using a commissioning timeout of max_window_duration using ECM",
"DUT_CE opens its Commissioning window to allow a second commissioning."),
TestStep("3b", "DNS-SD records shows DUT_CE advertising", "Verify that the DNS-SD advertisement shows CM=2"),
TestStep("3b", "DNS-SD records shows DUT_CE advertising",
"Verify that the DNS-SD advertisement shows CM=2"),
TestStep("3c", "TH_CR1 writes and reads the Basic Information Cluster’s NodeLabel mandatory attribute of DUT_CE",
"Verify DUT_CE responds to both write/read with a success"),
TestStep(4, "TH creates a controller (TH_CR2) on a new fabric and commissions DUT_CE using that controller. TH_CR2 should commission the device using a different NodeID than TH_CR1.",
Expand All @@ -283,7 +299,8 @@ def steps_TC_CADMIN_1_3(self) -> list[TestStep]:
"Verify DUT_CE responds to both write/read with a success"),
TestStep(8, "TH_CR2 reads, writes and then reads the Basic Information Cluster’s NodeLabel mandatory attribute of DUT_CE",
"Verify the initial read reflect the value written in the above step. Verify DUT_CE responds to both write/read with a success"),
TestStep(9, "TH_CR2 opens a commissioning window on DUT_CE for 180 seconds using ECM"),
TestStep(
9, "TH_CR2 opens a commissioning window on DUT_CE for 180 seconds using ECM"),
TestStep(10, "Wait for the commissioning window in step 9 to timeout"),
TestStep(11, "TH_CR2 reads the window status to verify the DUT_CE window is closed",
"DUT_CE windows status shows the window is closed"),
Expand All @@ -300,11 +317,13 @@ def pics_TC_CADMIN_1_4(self) -> list[str]:

def steps_TC_CADMIN_1_4(self) -> list[TestStep]:
return [
TestStep(1, "TH_CR1 starts a commissioning process with DUT_CE", is_commissioning=True),
TestStep(
1, "TH_CR1 starts a commissioning process with DUT_CE", is_commissioning=True),
TestStep(2, "TH_CR1 reads the BasicCommissioningInfo attribute from the General Commissioning cluster and saves the MaxCumulativeFailsafeSeconds field as max_window_duration."),
TestStep("3a", "TH_CR1 opens a commissioning window on DUT_CE using a commissioning timeout of max_window_duration using BCM",
"DUT_CE opens its Commissioning window to allow a second commissioning."),
TestStep("3b", "DNS-SD records shows DUT_CE advertising", "Verify that the DNS-SD advertisement shows CM=1"),
TestStep("3b", "DNS-SD records shows DUT_CE advertising",
"Verify that the DNS-SD advertisement shows CM=1"),
TestStep("3c", "TH_CR1 writes and reads the Basic Information Cluster’s NodeLabel mandatory attribute of DUT_CE",
"Verify DUT_CE responds to both write/read with a success"),
TestStep(4, "TH creates a controller (TH_CR2) on a new fabric and commissions DUT_CE using that controller. TH_CR2 should commission the device using a different NodeID than TH_CR1.",
Expand All @@ -327,4 +346,6 @@ async def test_TC_CADMIN_1_4(self):


if __name__ == "__main__":
# This makes the test class discoverable by the test runner
__test_class__ = TC_CADMIN
default_matter_test_main()
Loading
Loading