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.

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

python_testing: generate controller's nodeid and shutdown subscription #34231

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open
Changes from all commits
Commits
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
16 changes: 14 additions & 2 deletions src/python_testing/TC_ACE_1_2.py
Original file line number Diff line number Diff line change
@@ -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,13 +167,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(
23 changes: 19 additions & 4 deletions src/python_testing/TC_ACE_1_3.py
Original file line number Diff line number Diff line change
@@ -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),
4 changes: 2 additions & 2 deletions src/python_testing/TC_ACE_1_5.py
Original file line number Diff line number Diff line change
@@ -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))
4 changes: 2 additions & 2 deletions src/python_testing/TC_AccessChecker.py
Original file line number Diff line number Diff line change
@@ -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
8 changes: 6 additions & 2 deletions src/python_testing/TC_IDM_4_2.py
Original file line number Diff line number Diff line change
@@ -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),
8 changes: 8 additions & 0 deletions src/python_testing/matter_testing_support.py
Original file line number Diff line number Diff line change
@@ -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`.

Loading