From fa4d1cf3e41fefbec2546a5001e2307ea0170ad8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Szablowski?= Date: Tue, 9 Jul 2024 09:24:30 +0200 Subject: [PATCH] python_testing: generate controller's nodeid and shutdown subscription MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * implementation of generate_random_node_id with exculdes values, the method add possibilty to run multiple test cases within one test session (the controllers' nodeid is the same for new controller in different test cases) * ACE-1.2 script update: shutdown scubscriptions Signed-off-by: Michał Szablowski --- src/python_testing/TC_ACE_1_2.py | 17 +++++++++++++-- src/python_testing/TC_ACE_1_3.py | 23 ++++++++++++++++---- src/python_testing/TC_ACE_1_5.py | 4 ++-- src/python_testing/TC_AccessChecker.py | 4 ++-- src/python_testing/TC_IDM_4_2.py | 8 +++++-- src/python_testing/matter_testing_support.py | 9 +++++++- 6 files changed, 52 insertions(+), 13 deletions(-) diff --git a/src/python_testing/TC_ACE_1_2.py b/src/python_testing/TC_ACE_1_2.py index 2c6f07a217183a..421a318fd66f66 100644 --- a/src/python_testing/TC_ACE_1_2.py +++ b/src/python_testing/TC_ACE_1_2.py @@ -30,7 +30,7 @@ from chip.clusters.Attribute import EventReadResult, SubscriptionTransaction, TypedAttributePath from chip.exceptions import ChipStackError from chip.interaction_model import Status -from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main +from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main, generate_random_nodeid from mobly import asserts @@ -87,8 +87,16 @@ def __init__(self, *args): self.breadcrumb = 1 self.breadcrumb_queue = queue.Queue() self.subscription_breadcrumb = None + self.subscriptions = [] super().__init__(*args) + + def teardown_class(self): + for subscription in self.subscriptions: + subscription.Shutdown() + if self.subscription_breadcrumb is not None: + self.subscription_breadcrumb.Shutdown() + async def write_acl(self, acl): # This returns an attribute status result = await self.default_controller.WriteAttribute(self.dut_node_id, [(0, Clusters.AccessControl.Attributes.Acl(acl))]) @@ -97,6 +105,9 @@ async def write_acl(self, acl): async def steps_subscribe_breadcrumb(self, print_steps: bool): if print_steps: self.print_step(3, "TH2 subscribes to the Breadcrumb attribute") + if isinstance(self.subscription_breadcrumb, Clusters.Attribute.SubscriptionTransaction): + self.subscription_breadcrumb.Shutdown() + self.subscription_breadcrumb = await self.TH2.ReadAttribute(nodeid=self.dut_node_id, attributes=[(0, Clusters.GeneralCommissioning.Attributes.Breadcrumb)], reportInterval=(1, 5), keepSubscriptions=False, autoResubscribe=False) breadcrumb_cb = AttributeChangeCallback(Clusters.GeneralCommissioning.Attributes.Breadcrumb, self.breadcrumb_queue) self.subscription_breadcrumb.SetAttributeUpdateCallback(breadcrumb_cb) @@ -136,7 +147,7 @@ async def test_TC_ACE_1_2(self): fabric_admin = self.certificate_authority_manager.activeCaList[0].adminList[0] TH1_nodeid = self.matter_test_config.controller_node_id - TH2_nodeid = self.matter_test_config.controller_node_id + 1 + TH2_nodeid = generate_random_nodeid(excluded_nodeid={TH1_nodeid}) self.TH2 = fabric_admin.NewController(nodeId=TH2_nodeid, paaTrustStorePath=str(self.matter_test_config.paa_trust_store_path)) @@ -157,6 +168,7 @@ async def test_TC_ACE_1_2(self): acl_queue = queue.Queue() acl_cb = AttributeChangeCallback(Clusters.AccessControl.Attributes.Acl, acl_queue) subscription_acl.SetAttributeUpdateCallback(acl_cb) + self.subscriptions.append(subscription_acl) self.print_step(5, "TH2 subscribes to the AccessControlEntryChanged event") urgent = 1 @@ -164,6 +176,7 @@ async def test_TC_ACE_1_2(self): ace_queue = queue.Queue() ace_cb = EventChangeCallback(Clusters.AccessControl.Events.AccessControlEntryChanged, ace_queue) subscription_ace.SetEventUpdateCallback(ace_cb) + self.subscriptions.append(subscription_ace) self.print_step(6, "TH1 writes ACL attribute") acl = Clusters.AccessControl.Structs.AccessControlEntryStruct( diff --git a/src/python_testing/TC_ACE_1_3.py b/src/python_testing/TC_ACE_1_3.py index 150683357e1e9c..6d9090d494fe4f 100644 --- a/src/python_testing/TC_ACE_1_3.py +++ b/src/python_testing/TC_ACE_1_3.py @@ -26,7 +26,7 @@ import chip.clusters as Clusters from chip.interaction_model import Status -from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main +from matter_testing_support import MatterBaseTest, TestStep, async_test_body, default_matter_test_main, generate_random_nodeid from mobly import asserts @@ -137,9 +137,24 @@ async def test_TC_ACE_1_3(self): fabric_admin = self.certificate_authority_manager.activeCaList[0].adminList[0] TH0_nodeid = self.matter_test_config.controller_node_id - TH1_nodeid = self.matter_test_config.controller_node_id + 1 - TH2_nodeid = self.matter_test_config.controller_node_id + 2 - TH3_nodeid = self.matter_test_config.controller_node_id + 3 + TH1_nodeid = generate_random_nodeid( + excluded_nodeid={ + TH0_nodeid + } + ) + TH2_nodeid = generate_random_nodeid( + excluded_nodeid={ + TH0_nodeid, + TH1_nodeid + } + ) + TH3_nodeid = generate_random_nodeid( + excluded_nodeid={ + TH0_nodeid, + TH1_nodeid, + TH2_nodeid + } + ) TH1 = fabric_admin.NewController(nodeId=TH1_nodeid, paaTrustStorePath=str(self.matter_test_config.paa_trust_store_path), diff --git a/src/python_testing/TC_ACE_1_5.py b/src/python_testing/TC_ACE_1_5.py index e88a040e31b983..292506b431fedb 100644 --- a/src/python_testing/TC_ACE_1_5.py +++ b/src/python_testing/TC_ACE_1_5.py @@ -27,7 +27,7 @@ import chip.clusters as Clusters from chip import ChipDeviceCtrl from chip.interaction_model import Status -from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main +from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main, generate_random_nodeid from mobly import asserts @@ -53,7 +53,7 @@ async def test_TC_ACE_1_5(self): new_fabric_admin = new_certificate_authority.NewFabricAdmin(vendorId=0xFFF1, fabricId=self.matter_test_config.fabric_id + 1) TH1_nodeid = self.matter_test_config.controller_node_id - TH2_nodeid = self.matter_test_config.controller_node_id + 2 + TH2_nodeid = generate_random_nodeid(excluded_nodeid={TH1_nodeid}) self.th2 = new_fabric_admin.NewController(nodeId=TH2_nodeid, paaTrustStorePath=str(self.matter_test_config.paa_trust_store_path)) diff --git a/src/python_testing/TC_AccessChecker.py b/src/python_testing/TC_AccessChecker.py index f2bbf36330ec6a..2819ba86432b98 100644 --- a/src/python_testing/TC_AccessChecker.py +++ b/src/python_testing/TC_AccessChecker.py @@ -16,7 +16,7 @@ from chip.tlv import uint from global_attribute_ids import GlobalAttributeIds from matter_testing_support import (AttributePathLocation, ClusterPathLocation, MatterBaseTest, TestStep, async_test_body, - default_matter_test_main) + default_matter_test_main, generate_random_nodeid) from spec_parsing_support import XmlCluster, build_xml_clusters @@ -66,7 +66,7 @@ async def setup_class(self): self._record_errors() # We need to run this test from two controllers so we can test access to the ACL cluster while retaining access to the ACL cluster fabric_admin = self.certificate_authority_manager.activeCaList[0].adminList[0] - self.TH2_nodeid = self.matter_test_config.controller_node_id + 1 + self.TH2_nodeid = generate_random_nodeid(excluded_nodeid={self.matter_test_config.controller_node_id}) self.TH2 = fabric_admin.NewController(nodeId=self.TH2_nodeid) # Both the tests in this suite are potentially long-running if there are a large number of attributes on the DUT diff --git a/src/python_testing/TC_IDM_4_2.py b/src/python_testing/TC_IDM_4_2.py index 60016ba4dcf20a..acd31423f82f56 100644 --- a/src/python_testing/TC_IDM_4_2.py +++ b/src/python_testing/TC_IDM_4_2.py @@ -24,7 +24,7 @@ from chip.clusters.Attribute import AttributePath, TypedAttributePath from chip.exceptions import ChipStackError from chip.interaction_model import Status -from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main +from matter_testing_support import MatterBaseTest, async_test_body, default_matter_test_main, generate_random_nodeid from mobly import asserts ''' @@ -134,7 +134,11 @@ async def test_TC_IDM_4_2(self): # Subscriber/client with limited access to the DUT # Will validate error status codes fabric_admin = self.certificate_authority_manager.activeCaList[0].adminList[0] - CR2_nodeid = self.matter_test_config.controller_node_id + 1 + CR2_nodeid = generate_random_nodeid( + excluded_nodeid={ + self.matter_test_config.controller_node_id + } + ) CR2: ChipDeviceController = fabric_admin.NewController( nodeId=CR2_nodeid, paaTrustStorePath=str(self.matter_test_config.paa_trust_store_path), diff --git a/src/python_testing/matter_testing_support.py b/src/python_testing/matter_testing_support.py index 796397c57e0458..4c8e9bf01d3684 100644 --- a/src/python_testing/matter_testing_support.py +++ b/src/python_testing/matter_testing_support.py @@ -1493,6 +1493,13 @@ def parse_matter_test_args(argv: Optional[List[str]] = None) -> MatterTestConfig return convert_args_to_matter_config(parser.parse_known_args(argv)[0]) +def generate_random_nodeid(excluded_nodeid: typing.Optional[typing.Set] = set()) -> int: + "Generate random complainat nodeid, excluding a set of provided nodeids." + nodeid = random.randint(1, 0xFFFF_FFEF_FFFF_FFFF) + if nodeid in excluded_nodeid: + return generate_random_nodeid(excluded_nodeid) + return nodeid + def async_test_body(body): """Decorator required to be applied whenever a `test_*` method is `async def`. @@ -1697,7 +1704,7 @@ def run_tests_no_exit(test_class: MatterBaseTest, matter_test_config: MatterTest try: runner.run() - ok = runner.results.is_all_pass and ok + ok = runner.results.is_all_pass and ok except TimeoutError: ok = False except signals.TestAbortAll: