python_testing: generate controller's nodeid and shutdown subscription

* 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 <[email protected]>

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,13 +168,15 @@ 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
         subscription_ace = await self.TH2.ReadEvent(nodeid=self.dut_node_id, events=[(0, Clusters.AccessControl.Events.AccessControlEntryChanged, urgent)], reportInterval=(1, 5), fabricFiltered=False, keepSubscriptions=True, autoResubscribe=False)
         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(

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),

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))

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

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),

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: