Skip to content

Commit

Permalink
Python script: start of TH integration code (#28708)
Browse files Browse the repository at this point in the history 
* Python script: start of TH integration code

- adds required command line functionality (endpoint, nodeID, timeout)
- adds the ability to pass in a hooks object for use by the TH
- adds an internal hooks object when scripts are being used on
  the command line
- adds hook calls as appropriate at test start / end / steps

To setup a test for use in the TH, define a steps_ and desc_
function with the same name as the test, change print_step() calls
to step() calls with the appropriate number, add pics_guard calls
around pics checks.

Changes the hello_test to demonstrate how to do this and also how
the test works if the steps_ function is not defined.

Added a demonstration of how the TH could call these to get the
hooks back out.

* Add pics_ handler, address review comments

* lint

* Restyled by isort

* Update src/python_testing/matter_testing_support.py

Co-authored-by: Carolina Lopes <[email protected]>

* Init stuff in setup_class in case setup_class fails

* Make default timeout configurable

---------

Co-authored-by: Restyled.io <[email protected]>
Co-authored-by: Carolina Lopes <[email protected]>
  • Loading branch information
@cecille @restyled-commits @ccruzagralopes
3 people authored Dec 11, 2023
1 parent c4dd75e commit 696975f
Show file tree
Hide file tree
Showing 5 changed files with 633 additions and 73 deletions.
184 changes: 125 additions & 59 deletions src/python_testing/TC_ACE_1_3.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@

import chip.clusters as Clusters
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, TestStep, async_test_body, default_matter_test_main
from mobly import asserts


Expand All @@ -46,6 +46,72 @@ async def read_descriptor_expect_unsupported_access(self, th):
await self.read_single_attribute_expect_error(
dev_ctrl=th, endpoint=0, cluster=cluster, attribute=attribute, error=Status.UnsupportedAccess)

def desc_TC_ACE_1_3(self) -> str:
return "[TC-ACE-1.3] Subjects"

def steps_TC_ACE_1_3(self) -> list[TestStep]:
steps = [
TestStep(1, "Commissioning, already done", is_commissioning=True),
TestStep(2, "TH0 writes ACL all view on PIXIT.ACE.TESTENDPOINT"),
TestStep(3, "TH1 reads EP0 descriptor - expect SUCCESS"),
TestStep(4, "TH2 reads EP0 descriptor - expect SUCCESS"),
TestStep(5, "TH3 reads EP0 descriptor - expect SUCCESS"),
TestStep(6, "TH0 writes ACL TH1 view on EP0"),
TestStep(7, "TH1 reads EP0 descriptor - expect SUCCESS"),
TestStep(8, "TH2 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(9, "TH3 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(10, "TH0 writes ACL TH2 view on EP0"),
TestStep(11, "TH1 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(12, "TH2 reads EP0 descriptor - expect SUCCESS"),
TestStep(13, "TH3 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(14, "TH0 writes ACL TH3 view on EP0"),
TestStep(15, "TH1 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(16, "TH2 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(17, "TH3 reads EP0 descriptor - expect SUCCESS"),
TestStep(18, "TH0 writes ACL TH1 TH2 view on EP0"),
TestStep(19, "TH1 reads EP0 descriptor - expect SUCCESS"),
TestStep(20, "TH2 reads EP0 descriptor - expect SUCCESS"),
TestStep(21, "TH3 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(22, "TH0 writes ACL TH1 TH3 view on EP0"),
TestStep(23, "TH1 reads EP0 descriptor - expect SUCCESS"),
TestStep(24, "TH2 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(25, "TH3 reads EP0 descriptor - expect SUCCESS"),
TestStep(26, "TH0 writes ACL TH2 TH3 view on EP0"),
TestStep(27, "TH1 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(28, "TH2 reads EP0 descriptor - expect SUCCESS"),
TestStep(29, "TH3 reads EP0 descriptor - expect SUCCESS"),
TestStep(30, "TH0 writes ACL TH1 TH2 TH3 view on EP0"),
TestStep(31, "TH1 reads EP0 descriptor - expect SUCCESS"),
TestStep(32, "TH2 reads EP0 descriptor - expect SUCCESS"),
TestStep(33, "TH3 reads EP0 descriptor - expect SUCCESS"),
TestStep(34, "TH0 writes ACL cat1v1 view on EP0"),
TestStep(35, "TH1 reads EP0 descriptor - expect SUCCESS"),
TestStep(36, "TH2 reads EP0 descriptor - expect SUCCESS"),
TestStep(37, "TH3 reads EP0 descriptor - expect SUCCESS"),
TestStep(38, "TH0 writes ACL cat1v2 view on EP0"),
TestStep(39, "TH1 reads EP0 descriptor - expect SUCCESS"),
TestStep(40, "TH2 reads EP0 descriptor - expect SUCCESS"),
TestStep(41, "TH3 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(42, "TH0 writes ACL cat1v3 view on EP0"),
TestStep(43, "TH1 reads EP0 descriptor - expect SUCCESS"),
TestStep(44, "TH2 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(45, "TH3 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(46, "TH0 writes ACL cat2v1 view on EP0"),
TestStep(47, "TH1 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(48, "TH2 reads EP0 descriptor - expect SUCCESS"),
TestStep(49, "TH3 reads EP0 descriptor - expect SUCCESS"),
TestStep(50, "TH0 writes ACL cat2v2 view on EP0"),
TestStep(51, "TH1 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(52, "TH2 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(53, "TH3 reads EP0 descriptor - expect SUCCESS"),
TestStep(54, "TH0 writes ACL cat2v3 view on EP0"),
TestStep(55, "TH1 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(56, "TH2 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(57, "TH3 reads EP0 descriptor - expect UNSUPPORTED_ACCESS"),
TestStep(58, "TH0 writes ACL back to default")
]
return steps

@async_test_body
async def test_TC_ACE_1_3(self):
cat1_id = 0x11110000
Expand All @@ -59,7 +125,7 @@ async def test_TC_ACE_1_3(self):
cat2v3 = cat2_id | 0x0003
logging.info('cat1v1 0x%x', cat1v1)

self.print_step(1, "Commissioning, already done")
self.step(1)

fabric_admin = self.certificate_authority_manager.activeCaList[0].adminList[0]

Expand All @@ -78,7 +144,7 @@ async def test_TC_ACE_1_3(self):
paaTrustStorePath=str(self.matter_test_config.paa_trust_store_path),
catTags=[cat1v1, cat2v2])

self.print_step(2, "TH0 writes ACL all view on PIXIT.ACE.TESTENDPOINT")
self.step(2)
TH0_admin_acl = Clusters.AccessControl.Structs.AccessControlEntryStruct(
privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kAdminister,
authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
Expand All @@ -92,33 +158,33 @@ async def test_TC_ACE_1_3(self):
acl = [TH0_admin_acl, all_view]
await self.write_acl(acl)

self.print_step(3, "TH1 reads EP0 descriptor - expect SUCCESS")
self.step(3)
await self.read_descriptor_expect_success(TH1)

self.print_step(4, "TH2 reads EP0 descriptor - expect SUCCESS")
self.step(4)
await self.read_descriptor_expect_success(TH2)

self.print_step(5, "TH3 reads EP0 descriptor - expect SUCCESS")
self.step(5)
await self.read_descriptor_expect_success(TH3)

self.print_step(6, "TH0 writes ACL TH1 view on EP0")
self.step(6)
th1_view = Clusters.AccessControl.Structs.AccessControlEntryStruct(
privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView,
authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
subjects=[TH1_nodeid],
targets=[Clusters.AccessControl.Structs.AccessControlTargetStruct(endpoint=0)])
acl = [TH0_admin_acl, th1_view]
await self.write_acl(acl)
self.print_step(7, "TH1 reads EP0 descriptor - expect SUCCESS")
self.step(7)
await self.read_descriptor_expect_success(TH1)

self.print_step(8, "TH2 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(8)
await self.read_descriptor_expect_unsupported_access(TH2)

self.print_step(9, "TH3 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(9)
await self.read_descriptor_expect_unsupported_access(TH3)

self.print_step(10, "TH0 writes ACL TH2 view on EP0")
self.step(10)
th2_view = Clusters.AccessControl.Structs.AccessControlEntryStruct(
privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView,
authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
Expand All @@ -127,16 +193,16 @@ async def test_TC_ACE_1_3(self):

acl = [TH0_admin_acl, th2_view]
await self.write_acl(acl)
self.print_step(11, "TH1 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(11)
await self.read_descriptor_expect_unsupported_access(TH1)

self.print_step(12, "TH2 reads EP0 descriptor - expect SUCCESS")
self.step(12)
await self.read_descriptor_expect_success(TH2)

self.print_step(13, "TH3 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(13)
await self.read_descriptor_expect_unsupported_access(TH3)

self.print_step(14, "TH0 writes ACL TH3 view on EP0")
self.step(14)
th3_view = Clusters.AccessControl.Structs.AccessControlEntryStruct(
privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView,
authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
Expand All @@ -145,16 +211,16 @@ async def test_TC_ACE_1_3(self):

acl = [TH0_admin_acl, th3_view]
await self.write_acl(acl)
self.print_step(15, "TH1 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(15)
await self.read_descriptor_expect_unsupported_access(TH1)

self.print_step(16, "TH2 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(16)
await self.read_descriptor_expect_unsupported_access(TH2)

self.print_step(17, "TH3 reads EP0 descriptor - expect SUCCESS")
self.step(17)
await self.read_descriptor_expect_success(TH3)

self.print_step(18, "TH0 writes ACL TH1 TH2 view on EP0")
self.step(18)
th12_view = Clusters.AccessControl.Structs.AccessControlEntryStruct(
privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView,
authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
Expand All @@ -163,16 +229,16 @@ async def test_TC_ACE_1_3(self):

acl = [TH0_admin_acl, th12_view]
await self.write_acl(acl)
self.print_step(19, "TH1 reads EP0 descriptor - expect SUCCESS")
self.step(19)
await self.read_descriptor_expect_success(TH1)

self.print_step(20, "TH2 reads EP0 descriptor - expect SUCCESS")
self.step(20)
await self.read_descriptor_expect_success(TH2)

self.print_step(21, "TH3 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(21)
await self.read_descriptor_expect_unsupported_access(TH3)

self.print_step(22, "TH0 writes ACL TH1 TH3 view on EP0")
self.step(22)
th13_view = Clusters.AccessControl.Structs.AccessControlEntryStruct(
privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView,
authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
Expand All @@ -181,16 +247,16 @@ async def test_TC_ACE_1_3(self):

acl = [TH0_admin_acl, th13_view]
await self.write_acl(acl)
self.print_step(23, "TH1 reads EP0 descriptor - expect SUCCESS")
self.step(23)
await self.read_descriptor_expect_success(TH1)

self.print_step(24, "TH2 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(24)
await self.read_descriptor_expect_unsupported_access(TH2)

self.print_step(25, "TH3 reads EP0 descriptor - expect SUCCESS")
self.step(25)
await self.read_descriptor_expect_success(TH3)

self.print_step(26, "TH0 writes ACL TH2 TH3 view on EP0")
self.step(26)
th23_view = Clusters.AccessControl.Structs.AccessControlEntryStruct(
privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView,
authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
Expand All @@ -199,16 +265,16 @@ async def test_TC_ACE_1_3(self):

acl = [TH0_admin_acl, th23_view]
await self.write_acl(acl)
self.print_step(27, "TH1 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(27)
await self.read_descriptor_expect_unsupported_access(TH1)

self.print_step(28, "TH2 reads EP0 descriptor - expect SUCCESS")
self.step(28)
await self.read_descriptor_expect_success(TH2)

self.print_step(29, "TH3 reads EP0 descriptor - expect SUCCESS")
self.step(29)
await self.read_descriptor_expect_success(TH3)

self.print_step(30, "TH0 writes ACL TH1 TH2 TH3 view on EP0")
self.step(30)
th123_view = Clusters.AccessControl.Structs.AccessControlEntryStruct(
privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView,
authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
Expand All @@ -217,16 +283,16 @@ async def test_TC_ACE_1_3(self):

acl = [TH0_admin_acl, th123_view]
await self.write_acl(acl)
self.print_step(31, "TH1 reads EP0 descriptor - expect SUCCESS")
self.step(31)
await self.read_descriptor_expect_success(TH1)

self.print_step(32, "TH2 reads EP0 descriptor - expect SUCCESS")
self.step(32)
await self.read_descriptor_expect_success(TH2)

self.print_step(33, "TH3 reads EP0 descriptor - expect SUCCESS")
self.step(33)
await self.read_descriptor_expect_success(TH3)

self.print_step(34, "TH0 writes ACL cat1v1 view on EP0")
self.step(34)
cat1v1_view = Clusters.AccessControl.Structs.AccessControlEntryStruct(
privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView,
authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
Expand All @@ -235,16 +301,16 @@ async def test_TC_ACE_1_3(self):
acl = [TH0_admin_acl, cat1v1_view]
await self.write_acl(acl)

self.print_step(35, "TH1 reads EP0 descriptor - expect SUCCESS")
self.step(35)
await self.read_descriptor_expect_success(TH1)

self.print_step(36, "TH2 reads EP0 descriptor - expect SUCCESS")
self.step(36)
await self.read_descriptor_expect_success(TH2)

self.print_step(37, "TH3 reads EP0 descriptor - expect SUCCESS")
self.step(37)
await self.read_descriptor_expect_success(TH3)

self.print_step(38, "TH0 writes ACL cat1v2 view on EP0")
self.step(38)
cat1v2_view = Clusters.AccessControl.Structs.AccessControlEntryStruct(
privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView,
authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
Expand All @@ -254,16 +320,16 @@ async def test_TC_ACE_1_3(self):
acl = [TH0_admin_acl, cat1v2_view]
await self.write_acl(acl)

self.print_step(39, "TH1 reads EP0 descriptor - expect SUCCESS")
self.step(39)
await self.read_descriptor_expect_success(TH1)

self.print_step(40, "TH2 reads EP0 descriptor - expect SUCCESS")
self.step(40)
await self.read_descriptor_expect_success(TH2)

self.print_step(41, "TH3 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(41)
await self.read_descriptor_expect_unsupported_access(TH3)

self.print_step(42, "TH0 writes ACL cat1v3 view on EP0")
self.step(42)
cat1v3_view = Clusters.AccessControl.Structs.AccessControlEntryStruct(
privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView,
authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
Expand All @@ -273,16 +339,16 @@ async def test_TC_ACE_1_3(self):
acl = [TH0_admin_acl, cat1v3_view]
await self.write_acl(acl)

self.print_step(43, "TH1 reads EP0 descriptor - expect SUCCESS")
self.step(43)
await self.read_descriptor_expect_success(TH1)

self.print_step(44, "TH2 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(44)
await self.read_descriptor_expect_unsupported_access(TH2)

self.print_step(45, "TH3 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(45)
await self.read_descriptor_expect_unsupported_access(TH3)

self.print_step(46, "TH0 writes ACL cat2v1 view on EP0")
self.step(46)
cat2v1_view = Clusters.AccessControl.Structs.AccessControlEntryStruct(
privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView,
authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
Expand All @@ -292,16 +358,16 @@ async def test_TC_ACE_1_3(self):
acl = [TH0_admin_acl, cat2v1_view]
await self.write_acl(acl)

self.print_step(47, "TH1 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(47)
await self.read_descriptor_expect_unsupported_access(TH1)

self.print_step(48, "TH2 reads EP0 descriptor - expect SUCCESS")
self.step(48)
await self.read_descriptor_expect_success(TH2)

self.print_step(49, "TH3 reads EP0 descriptor - expect SUCCESS")
self.step(49)
await self.read_descriptor_expect_success(TH3)

self.print_step(50, "TH0 writes ACL cat2v2 view on EP0")
self.step(50)
cat2v2_view = Clusters.AccessControl.Structs.AccessControlEntryStruct(
privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView,
authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
Expand All @@ -311,16 +377,16 @@ async def test_TC_ACE_1_3(self):
acl = [TH0_admin_acl, cat2v2_view]
await self.write_acl(acl)

self.print_step(51, "TH1 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(51)
await self.read_descriptor_expect_unsupported_access(TH1)

self.print_step(52, "TH2 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(52)
await self.read_descriptor_expect_unsupported_access(TH2)

self.print_step(53, "TH3 reads EP0 descriptor - expect SUCCESS")
self.step(53)
await self.read_descriptor_expect_success(TH3)

self.print_step(54, "TH0 writes ACL cat2v3 view on EP0")
self.step(54)
cat2v3_view = Clusters.AccessControl.Structs.AccessControlEntryStruct(
privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView,
authMode=Clusters.AccessControl.Enums.AccessControlEntryAuthModeEnum.kCase,
Expand All @@ -330,16 +396,16 @@ async def test_TC_ACE_1_3(self):
acl = [TH0_admin_acl, cat2v3_view]
await self.write_acl(acl)

self.print_step(55, "TH1 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(55)
await self.read_descriptor_expect_unsupported_access(TH1)

self.print_step(56, "TH2 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(56)
await self.read_descriptor_expect_unsupported_access(TH2)

self.print_step(57, "TH3 reads EP0 descriptor - expect UNSUPPORTED_ACCESS")
self.step(57)
await self.read_descriptor_expect_unsupported_access(TH3)

self.print_step(58, "TH0 writes ACL back to default")
self.step(58)

full_acl = Clusters.AccessControl.Structs.AccessControlEntryStruct(
privilege=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kAdminister,
Expand Down
Loading

0 comments on commit 696975f

Please sign in to comment.