diff --git a/src/python_testing/TC_ACE_1_2.py b/src/python_testing/TC_ACE_1_2.py
index 2c6f07a217183a..d1496c89f6d6ba 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,15 @@ 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 +104,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 +146,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 +167,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 +175,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 dbf4e14905015a..f862ff88f7c512 100644
--- a/src/python_testing/matter_testing_support.py
+++ b/src/python_testing/matter_testing_support.py
@@ -1499,6 +1499,14 @@ 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`.