minor updates to latest testplan spec

Author: tehampson

src/python_testing/TC_CCTRL.py

@@ -32,7 +32,7 @@
 
 import chip.clusters as Clusters
 import chip.exceptions
-from matter_testing_support import MatterBaseTest, default_matter_test_main, per_endpoint_test, has_cluster, async_test_body
+from matter_testing_support import MatterBaseTest, TestStep, default_matter_test_main, per_endpoint_test, has_cluster, async_test_body
 from mobly import asserts
 from chip.interaction_model import InteractionModelError, Status
 
@@ -88,22 +88,55 @@ def teardown_class(self):
         os.remove(self.kvs)
         super().teardown_class()
 
+    
+    def steps_TC_CCTRL_2_2(self) -> list[TestStep]:
+        steps = [TestStep(1, "Get number of fabrics from TH_SERVER", is_commissioning=True),
+                 TestStep(2, "Reading Attribute VendorId from TH_SERVER"),
+                 TestStep(3, "Reading Attribute ProductId from TH_SERVER"),
+                 TestStep(4, "Reading Event CommissioningRequestResult from DUT"),
+                 TestStep(5, "Send CommissionNode command to DUT with CASE session"),
+                 TestStep(6, "Send OpenCommissioningWindow command on Administrator Commissioning Cluster to DUT with CASE session"),
+                 TestStep(7, "Send CommissionNode command to DUT with PASE session"),
+                 TestStep(8, "Send RequestCommissioningApproval command to DUT with PASE session"),
+                 TestStep(9, "Send RevokeCommissioning command on Administrator Commissioning Cluster to DUT with CASE session"),
+                 TestStep(10, "Reading Event CommissioningRequestResult from DUT, confirm no new events"),
+                 TestStep(11, "Send RequestCommissioningApproval command to DUT with CASE session with incorrect vendorID"),
+                 TestStep(12, "(Manual Step) Approve Commissioning Approval Request on DUT using method indicated by the manufacturer"),
+                 TestStep(13, "Reading Event CommissioningRequestResult from DUT, confirm one new event"),
+                 TestStep(14, "Send CommissionNode command to DUT with CASE session, with invalid RequestId"),
+                 TestStep(15, "Send CommissionNode command to DUT with CASE session, with invalid ResponseTimeoutSeconds too low"),
+                 TestStep(16, "Send CommissionNode command to DUT with CASE session, with invalid ResponseTimeoutSeconds too high"),
+                 TestStep(17, "Send CommissionNode command to DUT with CASE session, with valid parameters"),
+                 TestStep(18, "Send OpenCommissioningWindow command on Administrator Commissioning Cluster sent to TH_SERVER"),
+                 TestStep(19, "Wait for DUT to fail commissioning TH_SERVER, 30 seconds"),
+                 TestStep(20, "Get number of fabrics from TH_SERVER"),
+                 TestStep(21, "Send RevokeCommissioning command on Administrator Commissioning Cluster sent to TH_SERVER"),
+                 TestStep(22, "Send RequestCommissioningApproval command to DUT with CASE session with correct vendorID"),
+                 TestStep(23, "(Manual Step) Approve Commissioning Approval Request on DUT using method indicated by the manufacturer"),
+                 TestStep(24, "Reading Event CommissioningRequestResult from DUT, confirm one new event"),
+                 TestStep(25, "Send CommissionNode command to DUT with CASE session, with valid parameters"),
+                 TestStep(26, "Send OpenCommissioningWindow command on Administrator Commissioning Cluster sent to TH_SERVER"),
+                 TestStep(27, "Wait for DUT to successfully commission TH_SERVER, 30 seconds"),
+                 TestStep(28, "Get number of fabrics from TH_SERVER, verify DUT successfully commissioned TH_SERVER")]
+
+        return steps
+
     @per_endpoint_test(has_cluster(Clusters.CommissionerControl))
     async def test_TC_CCTRL_2_2(self):
         self.is_ci = self.check_pics('PICS_SDK_CI_ONLY')
 
-        #self.step(1)
+        self.step(1)
         th_server_fabrics = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.Fabrics, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0)
-        #self.step(2)
+        self.step(2)
         th_server_vid = await self.read_single_attribute_check_success(cluster=Clusters.BasicInformation, attribute=Clusters.BasicInformation.Attributes.VendorID, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0)
-        #self.step(3)
+        self.step(3)
         th_server_pid = await self.read_single_attribute_check_success(cluster=Clusters.BasicInformation, attribute=Clusters.BasicInformation.Attributes.ProductID, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0)
 
-        #self.step(4)
+        self.step(4)
         event_path = [(self.matter_test_config.endpoint, Clusters.CommissionerControl.Events.CommissioningRequestResult, 1)]
         events = await self.default_controller.ReadEvent(nodeid=self.dut_node_id, events=event_path)
 
-        #self.step(5)
+        self.step(5)
         ipaddr = ipaddress.IPv6Address('::1')
         cmd = Clusters.CommissionerControl.Commands.CommissionNode(requestId=1, responseTimeoutSeconds=30, ipAddress=ipaddr.packed, port=self.port)
         try:
@@ -112,8 +145,9 @@ async def test_TC_CCTRL_2_2(self):
         except InteractionModelError as e:
             asserts.assert_equal(e.status, Status.Failure, "Incorrect error returned")
 
-        #self.step(6)
+        self.step(6)
         params = await self.openCommissioningWindow(dev_ctrl=self.default_controller, node_id=self.dut_node_id)
+        self.step(7)
         pase_nodeid = self.dut_node_id + 1
         await self.default_controller.FindOrEstablishPASESession(setupCode=params.commissioningParameters.setupQRCode, nodeid=pase_nodeid)
         try:
@@ -122,7 +156,7 @@ async def test_TC_CCTRL_2_2(self):
         except InteractionModelError as e:
             asserts.assert_equal(e.status, Status.UnsupportedAccess, "Incorrect error returned")
 
-        # self.step(7)
+        self.step(8)
         good_request_id = 0x1234567887654321
         cmd = Clusters.CommissionerControl.Commands.RequestCommissioningApproval(requestId=good_request_id, vendorId=th_server_vid, productId=th_server_pid)
         try:
@@ -131,26 +165,32 @@ async def test_TC_CCTRL_2_2(self):
         except InteractionModelError as e:
             asserts.assert_equal(e.status, Status.UnsupportedAccess, "Incorrect error returned")
 
-        #self.step(8)
+        self.step(9)
+        cmd = Clusters.AdministratorCommissioning.Commands.RevokeCommissioning()
+        # If no exception is raised, this is success
+        await self.send_single_cmd(cmd)
+
+        self.step(10)
         if not events:
             new_event = await self.default_controller.ReadEvent(nodeid=self.dut_node_id, events=event_path)
         else:
             event_nums = [e.Header.EventNumber for e in events]
             new_event = await self.default_controller.ReadEvent(nodeid=self.dut_node_id, events=event_path, eventNumberFilter=max(event_nums)+1)
         asserts.assert_equal(new_event, [], "Unexpected event")
 
-
-        #self.step(9)
-        bad_vid = 0x6006
-        cmd = Clusters.CommissionerControl.Commands.RequestCommissioningApproval(requestId=good_request_id, vendorId=bad_vid, productId=th_server_pid)
+        self.step(11)
+        # There should be nothing on the TH that uses VID 0x6006 as this is a real vendor ID.
+        not_th_server_vid = 0x6006
+        asserts.assert_not_equal(not_th_server_vid, th_server_vid, "Test implementation assumption incorrect")
+        cmd = Clusters.CommissionerControl.Commands.RequestCommissioningApproval(requestId=good_request_id, vendorId=not_th_server_vid, productId=th_server_pid)
         # If no exception is raised, this is success
         await self.send_single_cmd(cmd)
 
-        #self.step(10)
+        self.step(12)
         if not self.is_ci:
             self.wait_for_use_input("Approve Commissioning approval request using manufacturer specified mechanism")
 
-        #self.step(11)
+        self.step(13)
         if not events:
             new_event = await self.default_controller.ReadEvent(nodeid=self.dut_node_id, events=event_path)
         else:
@@ -161,7 +201,7 @@ async def test_TC_CCTRL_2_2(self):
         asserts.assert_equal(new_event[0].Data.clientNodeId, self.matter_test_config.controller_node_id, "Unexpected client node id")
         asserts.assert_equal(new_event[0].Data.requestId, good_request_id, "Unexpected request ID")
 
-        #self.step(12)
+        self.step(14)
         bad_request_id = 0x1234567887654322
         cmd = Clusters.CommissionerControl.Commands.CommissionNode(requestId=bad_request_id, responseTimeoutSeconds=30)
         try:
@@ -170,59 +210,59 @@ async def test_TC_CCTRL_2_2(self):
         except InteractionModelError as e:
             asserts.assert_equal(e.status, Status.Failure, "Incorrect error returned")
 
-        #self.step(13)
+        self.step(15)
         cmd = Clusters.CommissionerControl.Commands.CommissionNode(requestId=good_request_id, responseTimeoutSeconds=29)
         try:
             await self.send_single_cmd(cmd=cmd)
             asserts.fail("Unexpected success on CommissionNode")
         except InteractionModelError as e:
             asserts.assert_equal(e.status, Status.Failure, "Incorrect error returned")
 
-        #self.step(14)
+        self.step(16)
         cmd = Clusters.CommissionerControl.Commands.CommissionNode(requestId=good_request_id, responseTimeoutSeconds=121)
         try:
             await self.send_single_cmd(cmd=cmd)
             asserts.fail("Unexpected success on CommissionNode")
         except InteractionModelError as e:
             asserts.assert_equal(e.status, Status.Failure, "Incorrect error returned")
 
-        #self.step(15)
+        self.step(17)
         cmd = Clusters.CommissionerControl.Commands.CommissionNode(requestId=good_request_id, responseTimeoutSeconds=30)
         resp: Clusters.CommissionerControl.Commands.ReverseOpenCommissioningWindow = await self.send_single_cmd(cmd)
         asserts.assert_equal(type(resp), Clusters.CommissionerControl.Commands.ReverseOpenCommissioningWindow, "Incorrect response type")
 
-        #self.step(16)
+        self.step(18)
         # min commissioning timeout is 3*60 seconds, so use that even though the command said 30.
         cmd = Clusters.AdministratorCommissioning.Commands.OpenCommissioningWindow(commissioningTimeout=3*60,
                                                                                    PAKEPasscodeVerifier=resp.PAKEPasscodeVerifier,
                                                                                    discriminator=resp.discriminator,
                                                                                    iterations=resp.iterations, salt=resp.salt)
         await self.send_single_cmd(cmd, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, timedRequestTimeoutMs=5000)
 
-        #self.step(17)
+        self.step(19)
         logging.info("Test now waits for 30 seconds")
-        #time.sleep(30)
+        if not self.is_ci:
+            time.sleep(30)
 
-        #self.step(18)
+        self.step(20)
         print(f'server node id {self.server_nodeid}')
         th_server_fabrics_new = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.Fabrics, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0)
         asserts.assert_equal(len(th_server_fabrics), len(th_server_fabrics_new), "Unexpected number of fabrics on TH_SERVER")
 
-        #self.step(19)
+        self.step(21)
         cmd = Clusters.AdministratorCommissioning.Commands.RevokeCommissioning()
         await self.send_single_cmd(cmd, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, timedRequestTimeoutMs=5000, endpoint=0)
 
-        #self.step(20)
-        print('step 20')
+        self.step(22)
         good_request_id = 0x1234567812345678
         cmd = Clusters.CommissionerControl.Commands.RequestCommissioningApproval(requestId=good_request_id, vendorId=th_server_vid, productId=th_server_pid, label="Test Ecosystem")
         await self.send_single_cmd(cmd)
 
-        #self.step(21)
+        self.step(23)
         if not self.is_ci:
             self.wait_for_use_input("Approve Commissioning approval request using manufacturer specified mechanism")
 
-        #self.step(22)
+        self.step(24)
         events = new_event
         event_nums = [e.Header.EventNumber for e in events]
         new_event = await self.default_controller.ReadEvent(nodeid=self.dut_node_id, events=event_path, eventNumberFilter=max(event_nums)+1)
@@ -231,23 +271,24 @@ async def test_TC_CCTRL_2_2(self):
         asserts.assert_equal(new_event[0].Data.clientNodeId, self.matter_test_config.controller_node_id, "Unexpected client node id")
         asserts.assert_equal(new_event[0].Data.requestId, good_request_id, "Unexpected request ID")
 
-        #self.step(23)
+        self.step(25)
         cmd = Clusters.CommissionerControl.Commands.CommissionNode(requestId=good_request_id, responseTimeoutSeconds=30)
         resp = await self.send_single_cmd(cmd)
         asserts.assert_equal(type(resp), Clusters.CommissionerControl.Commands.ReverseOpenCommissioningWindow, "Incorrect response type")
 
-        #self.step(24)
+        self.step(26)
         # min commissioning timeout is 3*60 seconds, so use that even though the command said 30.
         cmd = Clusters.AdministratorCommissioning.Commands.OpenCommissioningWindow(commissioningTimeout=3*60,
                                                                                    PAKEPasscodeVerifier=resp.PAKEPasscodeVerifier,
                                                                                    discriminator=resp.discriminator,
                                                                                    iterations=resp.iterations, salt=resp.salt)
         await self.send_single_cmd(cmd, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0, timedRequestTimeoutMs=5000)
 
-        #self.step(25)
+        self.step(27)
         if not self.is_ci:
             time.sleep(30)
 
+        self.step(28)
         th_server_fabrics_new = await self.read_single_attribute_check_success(cluster=Clusters.OperationalCredentials, attribute=Clusters.OperationalCredentials.Attributes.Fabrics, dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid, endpoint=0)
         #TODO: this should be mocked too.
         if not self.is_ci:

src/python_testing/test_testing/test_TC_CCNTL.py

@@ -51,24 +51,26 @@ def dynamic_invoke_return(*args, **argv):
     print(f'invoke call {invoke_call_count}')
     if invoke_call_count == 1: # Commission node with no prior request, return failure - step 5
         raise InteractionModelError(status=Status.Failure)
-    elif invoke_call_count == 2: # Commission node over pase - return unsupported access - step 6
+    elif invoke_call_count == 2: # Commission node over pase - return unsupported access - step 7
         raise InteractionModelError(status=Status.UnsupportedAccess)
-    elif invoke_call_count == 3: # request commissioning approval over pase - return unsupported access - step 7
+    elif invoke_call_count == 3: # request commissioning approval over pase - return unsupported access - step 8
         raise InteractionModelError(status=Status.UnsupportedAccess)
-    elif invoke_call_count == 4: # good RequestCommissioningApproval over CASE with bad vid - step 9
+    elif invoke_call_count == 4: # good RevokeCommissioning over CASE with bad vid - step 9
         return None
-    elif invoke_call_count == 5: # CommissionNode with bad request id - step 12
+    elif invoke_call_count == 5: # good RequestCommissioningApproval over CASE with bad vid - step 10
+        return None
+    elif invoke_call_count == 6: # CommissionNode with bad request id - step 14
         raise InteractionModelError(status=Status.Failure)
-    elif invoke_call_count == 6: # CommissionNode with bad timeout (low) - step 13
+    elif invoke_call_count == 7: # CommissionNode with bad timeout (low) - step 15
         raise InteractionModelError(status=Status.Failure)
-    elif invoke_call_count == 7: # CommissionNode with bad timeout (high) - step 14
+    elif invoke_call_count == 8: # CommissionNode with bad timeout (high) - step 16
         raise InteractionModelError(status=Status.Failure)
-    elif invoke_call_count == 8: # CommissionNode - step 15
+    elif invoke_call_count == 9: # CommissionNode - step 17
         # passcode 20202024
         return reverse_open
-    elif invoke_call_count == 9: # RequestCommissioningApproval with good vid - step 20
+    elif invoke_call_count == 10: # RequestCommissioningApproval with good vid - step 22
         return None
-    elif invoke_call_count == 10: # CommissionNode - step 23
+    elif invoke_call_count == 11: # CommissionNode - step 25
         # passcode 20202024
         return reverse_open
     else:
@@ -87,7 +89,7 @@ def dynamic_event_return(*args, **argv):
         data = Clusters.CommissionerControl.Events.CommissioningRequestResult(requestId=0x1234567887654321, clientNodeId=112233, statusCode=0)
         result = Attribute.EventReadResult(Header=header, Status=Status.Success, Data=data)
         return [result]
-    elif event_call_count == 4: # returmed event with new request
+    elif event_call_count == 4: # returned event with new request
         header = Attribute.EventHeader(EndpointId=0, ClusterId=Clusters.CommissionerControl.id, EventId=Clusters.CommissionerControl.Events.CommissioningRequestResult.event_id, EventNumber=1)
         data = Clusters.CommissionerControl.Events.CommissioningRequestResult(requestId=0x1234567812345678, clientNodeId=112233, statusCode=0)
         result = Attribute.EventReadResult(Header=header, Status=Status.Success, Data=data)
@@ -140,7 +142,7 @@ def main():
     print(f'paa = {paa_path}')
 
     pics = {"PICS_SDK_CI_ONLY": True }
-    test_runner = MyMock('TC_CCTRL', 'TC_CCTRL', 'test_TC_CCTRL_3_1', 1, paa_trust_store_path=paa_path, pics=pics)
+    test_runner = MyMock('TC_CCTRL', 'TC_CCTRL', 'test_TC_CCTRL_2_2', 1, paa_trust_store_path=paa_path, pics=pics)
 
     test_runner.run_test_with_mock(dynamic_invoke_return, dynamic_event_return, wildcard())
     test_runner.Shutdown()