implement more of the test.

Author: cecille

src/python_testing/TC_CCTRL.py

@@ -74,7 +74,6 @@ async def setup_class(self):
         new_certificate_authority = self.certificate_authority_manager.NewCertificateAuthority()
         new_fabric_admin = new_certificate_authority.NewFabricAdmin(vendorId=0xFFF1, fabricId=2)
         paa_path = str(self.matter_test_config.paa_trust_store_path)
-        print(f"paa_path = {paa_path}  ------------------------------------------------")
         self.TH_server_controller = new_fabric_admin.NewController(nodeId=112233, paaTrustStorePath=paa_path)
         self.server_nodeid = 1111
         await self.TH_server_controller.CommissionOnNetwork(nodeId=self.server_nodeid, setupPinCode=passcode, filterType=ChipDeviceCtrl.DiscoveryFilterType.LONG_DISCRIMINATOR, filter=discriminator)
@@ -88,17 +87,23 @@ def teardown_class(self):
 
         os.remove(self.kvs)
         super().teardown_class()
-    
 
-    #@per_endpoint_test(has_cluster(Clusters.CommissionerControl))
-    @async_test_body
+    @per_endpoint_test(has_cluster(Clusters.CommissionerControl))
     async def test_TC_CCTRL_3_1(self):
+        self.is_ci = self.check_pics('PICS_SDK_CI_ONLY')
+
+        #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)
         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)
         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)
 
-        # TODO: Read event, not yet implemented in mock
+        #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)
         ipaddr = ipaddress.IPv6Address('::1')
         cmd = Clusters.CommissionerControl.Commands.CommissionNode(requestId=1, responseTimeoutSeconds=30, ipAddress=ipaddr.packed, port=self.port)
         try:
@@ -107,8 +112,8 @@ async def test_TC_CCTRL_3_1(self):
         except InteractionModelError as e:
             asserts.assert_equal(e.status, Status.Failure, "Incorrect error returned")
 
+        #self.step(6)
         params = await self.openCommissioningWindow(dev_ctrl=self.default_controller, node_id=self.dut_node_id)
-
         pase_nodeid = self.dut_node_id + 1
         await self.default_controller.FindOrEstablishPASESession(setupCode=params.commissioningParameters.setupQRCode, nodeid=pase_nodeid)
         try:
@@ -117,19 +122,112 @@ async def test_TC_CCTRL_3_1(self):
         except InteractionModelError as e:
             asserts.assert_equal(e.status, Status.UnsupportedAccess, "Incorrect error returned")
 
+        # self.step(7)
         good_request_id = 0x1234567887654321
         cmd = Clusters.CommissionerControl.Commands.RequestCommissioningApproval(requestId=good_request_id, vendorId=th_server_vid, productId=th_server_pid)
         try:
             await self.send_single_cmd(cmd=cmd, node_id=pase_nodeid)
-            asserts.fail("Unexpected success on CommissionNode")
+            asserts.fail("Unexpected success on RequestCommissioningApproval over PASE")
         except InteractionModelError as e:
             asserts.assert_equal(e.status, Status.UnsupportedAccess, "Incorrect error returned")
 
+        #self.step(8)
+        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")
 
-        # TODO: read event - need to implement in the mock
 
+        #self.step(9)
+        bad_vid = 0x6006
+        cmd = Clusters.CommissionerControl.Commands.RequestCommissioningApproval(requestId=good_request_id, vendorId=bad_vid, productId=th_server_pid)
         # If no exception is raised, this is success
         await self.send_single_cmd(cmd)
 
+        #self.step(10)
+        if not self.is_ci:
+            self.wait_for_use_input("Approve Commissioning approval request using manufacturer specified mechanism")
+
+        #self.step(11)
+        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(len(new_event), 1, "Unexpected event list len")
+        asserts.assert_equal(new_event[0].Data.statusCode, 0, "Unexpected status code")
+        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)
+        bad_request_id = 0x1234567887654322
+        cmd = Clusters.CommissionerControl.Commands.CommissionNode(requestId=bad_request_id, responseTimeoutSeconds=30)
+        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(13)
+        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)
+        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)
+        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(16)
+        #TODO: we do not have a direct line to OpenEnhancedCommissioningWindow. Need to add plumbing
+        # For now, open with anything so the below command is ok
+        await self.openCommissioningWindow(dev_ctrl=self.TH_server_controller, node_id=self.server_nodeid)
+
+        #self.step(17)
+        logging.info("Test now waits for 30 seconds")
+        #time.sleep(30)
+
+        #self.step(18)
+        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)
+        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)
+        good_request_id = 0x1234567812345678
+        cmd = Clusters.CommissionerControl.Commands.RequestCommissioningApproval(requestId=good_request_id, vendorId=th_server_vid, productId=th_server_pid, label="Test Ecosystem")
+        self.send_single_cmd(cmd)
+
+        #self.step(21)
+        if not self.is_ci:
+            self.wait_for_use_input("Approve Commissioning approval request using manufacturer specified mechanism")
+
+        #self.step(22)
+        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)
+        asserts.assert_equal(len(new_event), 1, "Unexpected event list len")
+        asserts.assert_equal(new_event[0].Data.statusCode, 0, "Unexpected status code")
+        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")
+
+
+
 if __name__ == "__main__":
     default_matter_test_main()

src/python_testing/test_testing/test_TC_CCNTL.py

@@ -35,21 +35,50 @@
         os.path.join(os.path.dirname(__file__), '..')))
     from matter_testing_support import get_default_paa_trust_store, run_tests_no_exit
 
-call_count = 0
+invoke_call_count = 0
+event_call_count = 0
 
-def dynamic_return(*args, **argv):
-    print("using Mock invoke")
-    global call_count
-    call_count += 1
+def dynamic_invoke_return(*args, **argv):
+    global invoke_call_count
+    invoke_call_count += 1
 
-    if call_count == 1: # Commission node with no prior request, return failure
+    if invoke_call_count == 1: # Commission node with no prior request, return failure
         raise InteractionModelError(status=Status.Failure)
-    elif call_count == 2: # Commission node over pase - return unsupported access
+    elif invoke_call_count == 2: # Commission node over pase - return unsupported access
         raise InteractionModelError(status=Status.UnsupportedAccess)
-    elif call_count == 3: # request commissioning approval over pase - return unsupported access
+    elif invoke_call_count == 3: # request commissioning approval over pase - return unsupported access
         raise InteractionModelError(status=Status.UnsupportedAccess)
-    elif call_count == 4: # good RequestCommissioningApproval over CASE
+    elif invoke_call_count == 4: # good RequestCommissioningApproval over CASE with bad vid
         return None
+    elif invoke_call_count == 5: # CommissionNode with bad request id
+        raise InteractionModelError(status=Status.Failure)
+    elif invoke_call_count == 6: # CommissionNode with bad timeout (low)
+        raise InteractionModelError(status=Status.Failure)
+    elif invoke_call_count == 7: # CommissionNode with bad timeout (high)
+        raise InteractionModelError(status=Status.Failure)
+    elif invoke_call_count == 8: #CommissionNode
+        return Clusters.CommissionerControl.Commands.ReverseOpenCommissioningWindow(commissioningTimeout=30, PAKEPasscodeVerifier=b'', discriminator=2222, iterations=10000, salt=bytes('SaltyMcSaltersons', 'utf-8'))
+    else:
+        raise InteractionModelError(Status.Failure)
+
+def dynamic_event_return(*args, **argv):
+    global event_call_count
+    event_call_count += 1
+
+    if event_call_count == 1: # reading events, start empty - no events
+        return []
+    elif event_call_count == 2: # read event with filter - expect empty
+        return []
+    elif event_call_count == 3: # returned event
+        header = Attribute.EventHeader(EndpointId=0, ClusterId=Clusters.CommissionerControl.id, EventId=Clusters.CommissionerControl.Events.CommissioningRequestResult.event_id, EventNumber=1)
+        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 id
+        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)
+        return [result]
     else:
         raise InteractionModelError(Status.Failure)
 
@@ -79,14 +108,15 @@ def wildcard() -> Attribute.AsyncReadTransaction.ReadResponse:
 
 class MyMock(MockTestRunner):
     # TODO consolidate with above
-    def run_test_with_mock(self, dynamic_invoke_return: typing.Callable, read_cache: Attribute.AsyncReadTransaction.ReadResponse, hooks=None):
+    def run_test_with_mock(self, dynamic_invoke_return: typing.Callable, dynamic_event_return: typing.Callable, read_cache: Attribute.AsyncReadTransaction.ReadResponse, hooks=None):
         ''' Effects is a list of callable functions with *args, **kwargs parameters. It can either throw an InteractionModelException or return the command response.'''
         self.default_controller.Read = AsyncMock(return_value=read_cache)
         self.default_controller.SendCommand = AsyncMock(return_value=None, side_effect=dynamic_invoke_return)
         # It doesn't actually matter what we return here because I'm going to catch the next pase session connection anyway
         params = ChipDeviceCtrl.CommissioningParameters(setupPinCode=0, setupManualCode='', setupQRCode='')
         self.default_controller.OpenCommissioningWindow = AsyncMock(return_value=params)
         self.default_controller.FindOrEstablishPASESession = AsyncMock(return_value=None)
+        self.default_controller.ReadEvent = AsyncMock(return_value=[], side_effect=dynamic_event_return)
 
         return run_tests_no_exit(self.test_class, self.config, hooks, self.default_controller, self.stack)
 
@@ -96,9 +126,10 @@ def main():
     paa_path = get_default_paa_trust_store(root)
     print(f'paa = {paa_path}')
 
-    test_runner = MyMock('TC_CCTRL', 'TC_CCTRL', 'test_TC_CCTRL_3_1', 1, paa_trust_store_path=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.run_test_with_mock(dynamic_return, wildcard())
+    test_runner.run_test_with_mock(dynamic_invoke_return, dynamic_event_return, wildcard())
     test_runner.Shutdown()
 
 if __name__ == "__main__":