Fix lint

app/test_engine/models/test_case.py

@@ -62,7 +62,7 @@ def __init__(self, test_case_execution: TestCaseExecution):
         self.create_test_steps()
         self.__state = TestStateEnum.PENDING
         self.errors: List[str] = []
-        self.analytics: dict[str:str] = {}  # Move to dictionary
+        self.analytics: dict[str, str] = {}  # Move to dictionary
 
     # Make pics a class method as they are mostly needed at class level.
     @classmethod

test_collections/matter/sdk_tests/support/performance_tests/models/test_case.py

@@ -83,7 +83,7 @@ def __init__(self, test_case_execution: TestCaseExecution) -> None:
         self.__runned = 0
         self.test_stop_called = False
         self.test_socket = None
-        self.step_execution_times = []
+        self.step_execution_times = []  # type: ignore[var-annotated]
 
     # Move to the next step if the test case has additional steps apart from the 2
     # deafult ones
@@ -158,7 +158,7 @@ async def show_prompt(
             response = f"{user_response.response_str}\n".encode()
             self.test_socket._sock.sendall(response)  # type: ignore[attr-defined]
 
-    def generate_analytics_data(self) -> dict[str:str]:
+    def generate_analytics_data(self) -> dict[str, str]:
         print(self.step_execution_times)
         self.step_execution_times.sort()
         print(self.step_execution_times)
@@ -174,7 +174,7 @@ def generate_analytics_data(self) -> dict[str:str]:
                 "p99": f"{self.step_execution_times[p99_index]}",
                 "unit": "ms",
             }
-        except:
+        except:  # noqa: E722
             logger.info("Error generating analytics data for step execution times.")
         return {"p50": "0", "p95": "0", "p99": "0", "unit": "ms"}
 

test_collections/matter/sdk_tests/support/performance_tests/models/utils.py

@@ -33,7 +33,7 @@
 def generate_command_arguments(
     config: TestEnvironmentConfig, omit_commissioning_method: bool = False
 ) -> list:
-    dut_config = config.dut_config
+    dut_config = config.dut_config  # type: ignore[attr-defined]
     test_parameters = config.test_parameters
 
     pairing_mode = (

test_collections/matter/sdk_tests/support/performance_tests/scripts/sdk/TC_COMMISSIONING_1_0.py

@@ -14,19 +14,21 @@
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 #
-import time
 import logging
 import subprocess
+import time
+
 from chip import ChipDeviceCtrl
 from matter_testing_support import (
     MatterBaseTest,
     TestStep,
     async_test_body,
     default_matter_test_main,
 )
-from .accessory_manager import AccessoryManager
 from mobly import asserts
 
+from .accessory_manager import AccessoryManager
+
 # We don't have a good pipe between the c++ enums in CommissioningDelegate and python
 # so this is hardcoded.
 # I realize this is dodgy, not sure how to cross the enum from c++ to python cleanly
@@ -39,12 +41,11 @@
 
 
 class TC_COMMISSIONING_1_0(MatterBaseTest):
-
-    def __init__(self, *args):
+    def __init__(self, *args):  # type: ignore[no-untyped-def]
         super().__init__(*args)
         self.additional_steps = []
 
-    def setup_class(self):
+    def setup_class(self):  # type: ignore[no-untyped-def]
         self.commissioner = None
         self.commissioned = False
         self.discriminator = 3842
@@ -62,10 +63,10 @@ def steps_TC_COMMISSIONING_1_0(self) -> list[TestStep]:
             return steps
 
     @async_test_body
-    async def teardown_test(self):
+    async def teardown_test(self):  # type: ignore[no-untyped-def]
         return super().teardown_test()
 
-    async def commission_and_base_checks(self):
+    async def commission_and_base_checks(self):  # type: ignore[no-untyped-def]
         errcode = self.commissioner.CommissionOnNetwork(
             nodeId=self.dut_node_id,
             setupPinCode=20202021,
@@ -77,7 +78,7 @@ async def commission_and_base_checks(self):
         )
         self.commissioned = True
 
-    async def create_commissioner(self):
+    async def create_commissioner(self) -> None:
         new_certificate_authority = (
             self.certificate_authority_manager.NewCertificateAuthority()
         )
@@ -92,7 +93,7 @@ async def create_commissioner(self):
         self.commissioner.ResetTestCommissioner()
 
     @async_test_body
-    async def test_TC_COMMISSIONING_1_0(self):
+    async def test_TC_COMMISSIONING_1_0(self):  # type: ignore[no-untyped-def]
         accessory_manager = AccessoryManager()
         self.clean_chip_tool_kvs()
         await self.create_commissioner()
@@ -140,10 +141,10 @@ async def test_TC_COMMISSIONING_1_0(self):
             accessory_manager.stop()
             accessory_manager.clean()
 
-    def clean_chip_tool_kvs(self):
+    def clean_chip_tool_kvs(self):  # type: ignore[no-untyped-def]
         try:
             subprocess.check_call("rm -f /root/admin_storage.json", shell=True)
-            print(f"KVS info deleted.")
+            print("KVS info deleted.")
         except subprocess.CalledProcessError as e:
             print(f"Error deleting KVS info: {e}")
 

test_collections/matter/sdk_tests/support/performance_tests/scripts/sdk/accessory_manager.py

@@ -21,31 +21,30 @@
 # This interface must be implemented to provide basic access to accessory functionality.
 class AccessoryInterface(ABC):
     @abstractmethod
-    def start(self):
+    def start(self) -> None:
         pass
 
     @abstractmethod
-    def stop(self):
+    def stop(self) -> None:
         pass
 
     @abstractmethod
-    def clean(self):
+    def clean(self) -> None:
         pass
 
 
-from .simulated_accessory import SimulatedAccessory
+from .simulated_accessory import SimulatedAccessory  # noqa: E402
 
 
 class AccessoryManager:
-
     def __init__(self, accessory: AccessoryInterface = SimulatedAccessory()):
         self.accessory = accessory
 
-    def start(self):
+    def start(self) -> None:
         self.accessory.start()
 
-    def stop(self):
+    def stop(self) -> None:
         self.accessory.stop()
 
-    def clean(self):
+    def clean(self) -> None:
         self.accessory.clean()

test_collections/matter/sdk_tests/support/performance_tests/scripts/sdk/simulated_accessory.py

@@ -16,15 +16,15 @@
 #
 import signal
 import subprocess
+
 from .accessory_manager import AccessoryInterface
 
 
 class SimulatedAccessory(AccessoryInterface):
-
-    def __init__(self):
+    def __init__(self) -> None:
         self.process = None
 
-    def start(self):
+    def start(self) -> None:
         if self.process is None:
             # # Arguments to pass to the binary
             arguments = ["--discriminator", "3842", "--KVS", "kvs1"]
@@ -38,21 +38,21 @@ def start(self):
         else:
             print("Simulated App already running.")
 
-    def stop(self):
+    def stop(self) -> None:
         if self.process is not None:
             self.process.send_signal(signal.SIGTERM)
             self.process.wait()  # Wait for the process to exit
             self.process = None
         else:
             print("Simulated App is not running.")
 
-    def clean(self):
+    def clean(self) -> None:
         if self.process is not None:
             self.stop()  # Simulate App still running?
         try:
             subprocess.check_call("rm -rf /root/kvs1", shell=True)
             subprocess.check_call("rm -rf /tmp/chip_*", shell=True)
-            print(f"KVS info deleted.")
+            print("KVS info deleted.")
         except subprocess.CalledProcessError as e:
             print(f"Error deleting KVS info: {e}")
         try:

test_collections/matter/sdk_tests/support/performance_tests/utils.py

@@ -30,7 +30,6 @@
 def create_summary_report(
     timestamp: str, log_lines: list, commissioning_method: str
 ) -> tuple[str, str]:
-
     log_lines_list = "\n".join(log_lines)
 
     LOGS_FOLDER = "/test_collections/logs"
@@ -253,12 +252,12 @@ def generate_summary(
     summary_dict["test_summary_record"]["number_of_iterations_completed"] = len(
         durations
     )
-    summary_dict["test_summary_record"]["number_of_iterations_passed"] = (
-        compute_count_state(execution_status, True)
-    )
-    summary_dict["test_summary_record"]["number_of_iterations_failed"] = (
-        compute_count_state(execution_status, False)
-    )
+    summary_dict["test_summary_record"][
+        "number_of_iterations_passed"
+    ] = compute_count_state(execution_status, True)
+    summary_dict["test_summary_record"][
+        "number_of_iterations_failed"
+    ] = compute_count_state(execution_status, False)
     summary_dict["test_summary_record"]["platform"] = "rpi"
     summary_dict["test_summary_record"]["commissioning_method"] = commissioning_method
     summary_dict["test_summary_record"]["list_of_iterations_failed"] = []
@@ -396,7 +395,8 @@ def add_event(self, line: str) -> None:
             if not (stage in self.commissioning):
                 self.commissioning[stage] = {}
 
-            # pattern_begin = f"(?=.*{re.escape(stage)})(?=.*{re.escape(self.step_type[0])})"
+            # pattern_begin:
+            # f"(?=.*{re.escape(stage)})(?=.*{re.escape(self.step_type[0])})"
             if re.search(patterns["begin"], line) is not None:
                 match = re.findall(date_pattern, line)
                 if match[0]:
@@ -405,7 +405,8 @@ def add_event(self, line: str) -> None:
                         self.commissioning["begin"] = begin
                     self.commissioning[stage]["begin"] = begin
 
-            # pattern_end = f"(?=.*{re.escape(stage)})(?=.*{re.escape(self.step_type[1])})"
+            # pattern_end:
+            # f"(?=.*{re.escape(stage)})(?=.*{re.escape(self.step_type[1])})"
             if re.search(patterns["end"], line) is not None:
                 match = re.findall(date_pattern, line)
                 if match[0]:

test_collections/matter/sdk_tests/support/sdk_container.py

@@ -65,7 +65,7 @@
     "/root/python_testing/scripts/sdk/test_harness_client.py"
 )
 
-# Stress/Stability Test Script (For now it is injected on SDK container. Not the final solution.)
+# Stress/Stability Test Script (For now it is injected on SDK container.)
 LOCAL_STRESS_TEST_SCRIPT_PATH = Path(
     LOCAL_TEST_COLLECTIONS_PATH + "/sdk_tests/support/performance_tests/scripts/sdk/"
     "TC_COMMISSIONING_1_0.py"