From 5151ad37ff9c6c1b6d6fd13f0352224c09d7d24b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Micha=C5=82=20Szablowski?= <michal.szablowski@nordicsemi.no>
Date: Tue, 9 Jul 2024 09:24:30 +0200
Subject: [PATCH 1/2] 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 <michal.szablowski@nordicsemi.no>
---
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 dbf4e14905015a..4bf5b5a4d08599 100644
--- a/src/python_testing/matter_testing_support.py
+++ b/src/python_testing/matter_testing_support.py
@@ -1499,6 +1499,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`.
@@ -1703,7 +1710,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:
From 26e8cfb47992274fe48491229b491144bfe1262b Mon Sep 17 00:00:00 2001
From: "Restyled.io" <commits@restyled.io>
Date: Tue, 9 Jul 2024 07:28:59 +0000
Subject: [PATCH 2/2] Restyled by autopep8
---
src/python_testing/TC_ACE_1_2.py | 1 -
src/python_testing/matter_testing_support.py | 3 ++-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/src/python_testing/TC_ACE_1_2.py b/src/python_testing/TC_ACE_1_2.py
index 421a318fd66f66..d1496c89f6d6ba 100644
--- a/src/python_testing/TC_ACE_1_2.py
+++ b/src/python_testing/TC_ACE_1_2.py
@@ -90,7 +90,6 @@ def __init__(self, *args):
self.subscriptions = []
super().__init__(*args)
-
def teardown_class(self):
for subscription in self.subscriptions:
subscription.Shutdown()
diff --git a/src/python_testing/matter_testing_support.py b/src/python_testing/matter_testing_support.py
index 4bf5b5a4d08599..f862ff88f7c512 100644
--- a/src/python_testing/matter_testing_support.py
+++ b/src/python_testing/matter_testing_support.py
@@ -1506,6 +1506,7 @@ def generate_random_nodeid(excluded_nodeid: typing.Optional[typing.Set] = set())
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`.
@@ -1710,7 +1711,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: