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
Show file tree
Hide file tree
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
Expand Up @@ -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


Expand Down Expand Up @@ -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))])
Expand All @@ -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)
Expand Down Expand Up @@ -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))
Expand All @@ -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(
Expand Down
23 changes: 19 additions & 4 deletions src/python_testing/TC_ACE_1_3.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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),
Expand Down
4 changes: 2 additions & 2 deletions src/python_testing/TC_ACE_1_5.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand All @@ -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))
Expand Down
4 changes: 2 additions & 2 deletions src/python_testing/TC_AccessChecker.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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
Expand Down
8 changes: 6 additions & 2 deletions src/python_testing/TC_IDM_4_2.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

'''
Expand Down Expand Up @@ -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),
Expand Down
8 changes: 8 additions & 0 deletions src/python_testing/matter_testing_support.py
Original file line number Diff line number Diff line change
Expand Up @@ -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`.
Expand Down
Loading