Restyled by autopep8

Author: restyled-commits

src/python_testing/TC_CCTRL_2_2.py

@@ -52,7 +52,8 @@ async def setup_class(self):
         super().setup_class()
         # TODO: This needs to come from an arg and needs to be something available on the TH
         # TODO: confirm whether we can open processes like this on the TH
-        app = os.path.join(pathlib.Path(__file__).resolve().parent, '..','..','out', 'linux-x64-all-clusters-no-ble', 'chip-all-clusters-app')
+        app = os.path.join(pathlib.Path(__file__).resolve().parent, '..', '..', 'out',
+                           'linux-x64-all-clusters-no-ble', 'chip-all-clusters-app')
 
         self.kvs = f'kvs_{str(uuid.uuid4())}'
         self.port = 5543
@@ -69,7 +70,6 @@ async def setup_class(self):
 
         logging.info("Commissioning from separate fabric")
 
-
         # Create a second controller on a new fabric to communicate to the server
         new_certificate_authority = self.certificate_authority_manager.NewCertificateAuthority()
         new_fabric_admin = new_certificate_authority.NewFabricAdmin(vendorId=0xFFF1, fabricId=2)
@@ -88,7 +88,6 @@ 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"),
@@ -138,7 +137,8 @@ async def test_TC_CCTRL_2_2(self):
 
         self.step(5)
         ipaddr = ipaddress.IPv6Address('::1')
-        cmd = Clusters.CommissionerControl.Commands.CommissionNode(requestId=1, responseTimeoutSeconds=30, ipAddress=ipaddr.packed, port=self.port)
+        cmd = Clusters.CommissionerControl.Commands.CommissionNode(
+            requestId=1, responseTimeoutSeconds=30, ipAddress=ipaddr.packed, port=self.port)
         try:
             await self.send_single_cmd(cmd)
             asserts.fail("Unexpected success on CommissionNode")
@@ -158,7 +158,8 @@ async def test_TC_CCTRL_2_2(self):
 
         self.step(8)
         good_request_id = 0x1234567887654321
-        cmd = Clusters.CommissionerControl.Commands.RequestCommissioningApproval(requestId=good_request_id, vendorId=th_server_vid, productId=th_server_pid)
+        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 RequestCommissioningApproval over PASE")
@@ -182,7 +183,8 @@ async def test_TC_CCTRL_2_2(self):
         # 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)
+        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)
 
@@ -198,7 +200,8 @@ async def test_TC_CCTRL_2_2(self):
             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.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(14)
@@ -229,7 +232,8 @@ async def test_TC_CCTRL_2_2(self):
         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")
+        asserts.assert_equal(type(resp), Clusters.CommissionerControl.Commands.ReverseOpenCommissioningWindow,
+                             "Incorrect response type")
 
         self.step(18)
         # min commissioning timeout is 3*60 seconds, so use that even though the command said 30.
@@ -255,7 +259,8 @@ async def test_TC_CCTRL_2_2(self):
 
         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")
+        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(23)
@@ -268,13 +273,15 @@ async def test_TC_CCTRL_2_2(self):
         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.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(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")
+        asserts.assert_equal(type(resp), Clusters.CommissionerControl.Commands.ReverseOpenCommissioningWindow,
+                             "Incorrect response type")
 
         self.step(26)
         # min commissioning timeout is 3*60 seconds, so use that even though the command said 30.
@@ -290,11 +297,11 @@ async def test_TC_CCTRL_2_2(self):
 
         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.
+        # TODO: this should be mocked too.
         if not self.is_ci:
-            asserts.assert_equal(len(th_server_fabrics) + 1, len(th_server_fabrics_new), "Unexpected number of fabrics on TH_SERVER")
-
+            asserts.assert_equal(len(th_server_fabrics) + 1, len(th_server_fabrics_new),
+                                 "Unexpected number of fabrics on TH_SERVER")
 
 
 if __name__ == "__main__":
-    default_matter_test_main()
+    default_matter_test_main()

src/python_testing/test_testing/MockTestRunner.py

@@ -34,9 +34,11 @@
 class AsyncMock(MagicMock):
     async def __call__(self, *args, **kwargs):
         return super(AsyncMock, self).__call__(*args, **kwargs)
+
+
 class MockTestRunner():
 
-    def __init__(self, filename: str, classname: str, test: str, endpoint: int = 0, pics: dict[str, bool] = None, paa_trust_store_path = None):
+    def __init__(self, filename: str, classname: str, test: str, endpoint: int = 0, pics: dict[str, bool] = None, paa_trust_store_path=None):
         self.test = test
         self.endpoint = endpoint
         self.pics = pics

src/python_testing/test_testing/test_TC_CCNTL_2_2.py

@@ -39,6 +39,7 @@
 invoke_call_count = 0
 event_call_count = 0
 
+
 def dynamic_invoke_return(*args, **argv):
     global invoke_call_count
     invoke_call_count += 1
@@ -49,54 +50,60 @@ def dynamic_invoke_return(*args, **argv):
                                                                                         discriminator=2222, iterations=10000, salt=base64.b64encode(bytes('SaltyMcSalterson', 'utf-8')))
 
     print(f'invoke call {invoke_call_count}')
-    if invoke_call_count == 1: # Commission node with no prior request, return failure - step 5
+    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 7
+    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 8
+    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 RevokeCommissioning 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: # good RequestCommissioningApproval over CASE with bad vid - step 10
+    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
+    elif invoke_call_count == 6:  # CommissionNode with bad request id - step 14
         raise InteractionModelError(status=Status.Failure)
-    elif invoke_call_count == 7: # CommissionNode with bad timeout (low) - step 15
+    elif invoke_call_count == 7:  # CommissionNode with bad timeout (low) - step 15
         raise InteractionModelError(status=Status.Failure)
-    elif invoke_call_count == 8: # CommissionNode with bad timeout (high) - step 16
+    elif invoke_call_count == 8:  # CommissionNode with bad timeout (high) - step 16
         raise InteractionModelError(status=Status.Failure)
-    elif invoke_call_count == 9: # CommissionNode - step 17
+    elif invoke_call_count == 9:  # CommissionNode - step 17
         # passcode 20202024
         return reverse_open
-    elif invoke_call_count == 10: # RequestCommissioningApproval with good vid - step 22
+    elif invoke_call_count == 10:  # RequestCommissioningApproval with good vid - step 22
         return None
-    elif invoke_call_count == 11: # CommissionNode - step 25
+    elif invoke_call_count == 11:  # CommissionNode - step 25
         # passcode 20202024
         return reverse_open
     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
+    if event_call_count == 1:  # reading events, start empty - no events
         return []
-    elif event_call_count == 2: # read event with filter - expect empty
+    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)
+    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: # 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)
+    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)
         return [result]
     else:
         raise InteractionModelError(Status.Failure)
 
+
 def wildcard() -> Attribute.AsyncReadTransaction.ReadResponse:
     cc = Clusters.CommissionerControl
     ei = Clusters.EcosystemInformation
@@ -105,22 +112,27 @@ def wildcard() -> Attribute.AsyncReadTransaction.ReadResponse:
 
     # EP1 is aggregator device type with a commissioner control cluster
     # children - EP2 type bridged node endpoint, ecosystem information, bridged device basic information. Should also have and admin commissioning, but I don't need it for this test.
-    desc_ep1 = {desc.Attributes.PartsList: [2], desc.Attributes.ServerList: [cc.id], desc.Attributes.DeviceTypeList: [desc.Structs.DeviceTypeStruct(deviceType=0x000E, revision=2)]}
-    desc_ep2 = {desc.Attributes.ServerList: [bdbi.id, ei.id], desc.Attributes.DeviceTypeList: [desc.Structs.DeviceTypeStruct(deviceType=0x0013, revision=3)]}
+    desc_ep1 = {desc.Attributes.PartsList: [2], desc.Attributes.ServerList: [
+        cc.id], desc.Attributes.DeviceTypeList: [desc.Structs.DeviceTypeStruct(deviceType=0x000E, revision=2)]}
+    desc_ep2 = {desc.Attributes.ServerList: [bdbi.id, ei.id], desc.Attributes.DeviceTypeList: [
+        desc.Structs.DeviceTypeStruct(deviceType=0x0013, revision=3)]}
 
     # I'm not filling anything in here, because I don't care. I just care that the cluster exists.
-    ei_attrs = {ei.Attributes.AttributeList:[ei.Attributes.DeviceDirectory.attribute_id, ei.Attributes.LocationDirectory.attribute_id], ei.Attributes.DeviceDirectory:[], ei.Attributes.LocationDirectory:[]}
+    ei_attrs = {ei.Attributes.AttributeList: [ei.Attributes.DeviceDirectory.attribute_id,
+                                              ei.Attributes.LocationDirectory.attribute_id], ei.Attributes.DeviceDirectory: [], ei.Attributes.LocationDirectory: []}
 
     # This cluster just needs to exist, so I'm just going to throw on the mandatory items for now.
-    bdbi_attrs = {bdbi.Attributes.AttributeList:[bdbi.Attributes.Reachable.attribute_id, bdbi.Attributes.UniqueID.attribute_id], bdbi.Attributes.Reachable:True, bdbi.Attributes.UniqueID:'something'}
+    bdbi_attrs = {bdbi.Attributes.AttributeList: [bdbi.Attributes.Reachable.attribute_id,
+                                                  bdbi.Attributes.UniqueID.attribute_id], bdbi.Attributes.Reachable: True, bdbi.Attributes.UniqueID: 'something'}
 
-    cc_attrs = {cc.Attributes.AttributeList:[cc.Attributes.SupportedDeviceCategories], cc.Attributes.AcceptedCommandList:[cc.Commands.RequestCommissioningApproval, cc.Commands.CommissionNode],
-                cc.Attributes.GeneratedCommandList:[cc.Commands.RequestCommissioningApproval], cc.Attributes.SupportedDeviceCategories:1}
+    cc_attrs = {cc.Attributes.AttributeList: [cc.Attributes.SupportedDeviceCategories], cc.Attributes.AcceptedCommandList: [cc.Commands.RequestCommissioningApproval, cc.Commands.CommissionNode],
+                cc.Attributes.GeneratedCommandList: [cc.Commands.RequestCommissioningApproval], cc.Attributes.SupportedDeviceCategories: 1}
 
     resp = Attribute.AsyncReadTransaction.ReadResponse({}, [], {})
-    resp.attributes = {1: {desc: desc_ep1, cc:cc_attrs}, 2:{desc:desc_ep2, ei:ei_attrs, bdbi:bdbi_attrs}}
+    resp.attributes = {1: {desc: desc_ep1, cc: cc_attrs}, 2: {desc: desc_ep2, ei: ei_attrs, bdbi: bdbi_attrs}}
     return resp
 
+
 class MyMock(MockTestRunner):
     # TODO consolidate with above
     def run_test_with_mock(self, dynamic_invoke_return: typing.Callable, dynamic_event_return: typing.Callable, read_cache: Attribute.AsyncReadTransaction.ReadResponse, hooks=None):
@@ -135,17 +147,19 @@ def run_test_with_mock(self, dynamic_invoke_return: typing.Callable, dynamic_eve
 
         return run_tests_no_exit(self.test_class, self.config, hooks, self.default_controller, self.stack)
 
+
 def main():
-    root = os.path.abspath(os.path.join(pathlib.Path(__file__).resolve().parent, '..','..','..'))
+    root = os.path.abspath(os.path.join(pathlib.Path(__file__).resolve().parent, '..', '..', '..'))
     print(f'root = {root}')
     paa_path = get_default_paa_trust_store(root)
     print(f'paa = {paa_path}')
 
-    pics = {"PICS_SDK_CI_ONLY": True }
+    pics = {"PICS_SDK_CI_ONLY": True}
     test_runner = MyMock('TC_CCTRL_2_2', 'TC_CCTRL_2_2', '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()
 
+
 if __name__ == "__main__":
     sys.exit(main())