[python tests] Add to flake8 in workflow and fix python files (part #…

…25193) (#25312)

* [python tests] Add to flake8 in workflow and fix python files

* Restyled by autopep8

* Add testing files from controller

---------

Co-authored-by: Restyled.io <[email protected]>

.flake8

@@ -33,17 +33,14 @@ exclude = third_party
           scripts/build/builders/imx.py
           scripts/build/builders/infineon.py
           scripts/build/builders/nrf.py
-          scripts/build/test.py
           scripts/codegen.py
           scripts/codepregen.py
           scripts/error_table.py
           scripts/examples/gn_to_cmakelists.py
-          scripts/examples/tests/test.py
           scripts/flashing/bouffalolab_firmware_utils.py
           scripts/flashing/cyw30739_firmware_utils.py
           scripts/flashing/nrfconnect_firmware_utils.py
           scripts/gen_chip_version.py
-          scripts/gen_test_driver.py
           scripts/helpers/bloat_check.py
           scripts/pregenerate/using_codegen.py
           scripts/pregenerate/using_zap.py
@@ -60,16 +57,7 @@ exclude = third_party
           scripts/py_matter_yamltests/test_yaml_parser.py
           scripts/run-clang-tidy-on-compile-commands.py
           scripts/setup/nrfconnect/update_ncs.py
-          scripts/tests/chiptest/__init__.py
-          scripts/tests/chiptest/runner.py
-          scripts/tests/chiptest/test_definition.py
           scripts/tests/chiptest/yamltest_with_chip_repl_tester.py
-          scripts/tests/java/base.py
-          scripts/tests/java/commissioning_test.py
-          scripts/tests/java/discover_test.py
-          scripts/tests/run_java_test.py
-          scripts/tests/run_python_test.py
-          scripts/tests/run_test_suite.py
           scripts/tools/check_zcl_file_sync.py
           scripts/tools/convert_ini.py
           scripts/tools/generate_esp32_chip_factory_bin.py
@@ -82,12 +70,10 @@ exclude = third_party
           scripts/tools/telink/mfg_tool.py
           scripts/tools/zap/generate.py
           scripts/tools/zap/prune_outputs.py
-          scripts/tools/zap/test_generate.py
           scripts/tools/zap/version_update.py
           scripts/tools/zap/zap_download.py
           scripts/tools/zap_convert_all.py
           src/app/ota_image_tool.py
-          src/app/tests/suites/certification/information.py
           src/app/zap_cluster_list.py
           src/controller/python/build-chip-wheel.py
           src/controller/python/chip-device-ctrl.py
@@ -117,23 +103,7 @@ exclude = third_party
           src/controller/python/chip/yaml/__init__.py
           src/controller/python/chip/yaml/format_converter.py
           src/controller/python/chip/yaml/runner.py
-          src/controller/python/test/test_scripts/base.py
-          src/controller/python/test/test_scripts/cluster_objects.py
-          src/controller/python/test/test_scripts/mobile-device-test.py
-          src/controller/python/test/test_scripts/network_commissioning.py
-          src/controller/python/test/unit_tests/test_cluster_objects.py
-          src/controller/python/test/unit_tests/test_tlv.py
           src/lib/asn1/gen_asn1oid.py
           src/pybindings/pycontroller/build-chip-wheel.py
           src/pybindings/pycontroller/pychip/__init__.py
-          src/python_testing/TC_ACE_1_3.py
-          src/python_testing/TC_ACE_1_4.py
-          src/python_testing/TC_CGEN_2_4.py
-          src/python_testing/TC_DA_1_7.py
-          src/python_testing/TC_RR_1_1.py
-          src/python_testing/TC_SC_3_6.py
-          src/python_testing/TC_TestEventTrigger.py
-          src/python_testing/hello_test.py
-          src/python_testing/matter_testing_support.py
           src/setup_payload/python/generate_setup_payload.py
-          src/setup_payload/tests/run_python_setup_payload_gen_test.py

build/chip/java/tests/test.py

@@ -18,7 +18,6 @@
 
 import json
 import os
-import subprocess
 import unittest
 from os import path
 

scripts/build/test.py

@@ -26,8 +26,8 @@
 
 def build_expected_output(source: str, root: str, out: str) -> List[str]:
     with open(os.path.join(SCRIPT_ROOT, source), 'rt') as f:
-        for l in f.readlines():
-            yield l.replace("{root}", root).replace("{out}", out)
+        for line in f.readlines():
+            yield line.replace("{root}", root).replace("{out}", out)
 
 
 def build_actual_output(root: str, out: str, args: List[str]) -> List[str]:
@@ -57,7 +57,7 @@ def build_actual_output(root: str, out: str, args: List[str]) -> List[str]:
         '--out-prefix', out,
     ] + args, stdout=subprocess.PIPE, check=True, encoding='UTF-8', env=runenv)
 
-    result = [l + '\n' for l in retval.stdout.split('\n')]
+    result = [line + '\n' for line in retval.stdout.split('\n')]
 
     # ensure a single terminating newline: easier to edit since autoformat
     # often strips ending double newlines on text files
@@ -73,22 +73,22 @@ def assertCommandOutput(self, expected_file: str, args: List[str]):
         ROOT = '/TEST/BUILD/ROOT'
         OUT = '/OUTPUT/DIR'
 
-        expected = [l for l in build_expected_output(expected_file, ROOT, OUT)]
-        actual = [l for l in build_actual_output(ROOT, OUT, args)]
+        expected = [line for line in build_expected_output(expected_file, ROOT, OUT)]
+        actual = [line for line in build_actual_output(ROOT, OUT, args)]
 
         diffs = [line for line in difflib.unified_diff(expected, actual)]
 
         if diffs:
             reference = os.path.basename(expected_file) + '.actual'
             with open(reference, 'wt') as fo:
-                for l in build_actual_output(ROOT, OUT, args):
-                    fo.write(l.replace(ROOT, '{root}').replace(OUT, '{out}'))
+                for line in build_actual_output(ROOT, OUT, args):
+                    fo.write(line.replace(ROOT, '{root}').replace(OUT, '{out}'))
 
             msg = "DIFFERENCE between expected and generated output in %s\n" % expected_file
             msg += "Expected file can be found in %s" % reference
-            for l in diffs:
-                msg += ("\n   " + l.replace(ROOT,
-                                            '{root}').replace(OUT, '{out}').strip())
+            for line in diffs:
+                msg += ("\n   " + line.replace(ROOT,
+                                               '{root}').replace(OUT, '{out}').strip())
             self.fail(msg)
 
     @unittest.skipUnless(sys.platform == 'linux', 'Build on linux test')

scripts/examples/tests/test.py

@@ -27,24 +27,24 @@
 
 
 def build_expected_output(root: str, out: str) -> List[str]:
-    with open(os.path.join(SCRIPT_ROOT, 'expected_test_cmakelists.txt'), 'rt') as f:
-        for l in f.readlines():
-            yield l.replace("{root}", root).replace("{out}", out)
+    with open(os.path.join(SCRIPT_ROOT, 'expected_test_cmakelists.txt'), 'rt') as file:
+        for line in file.readlines():
+            yield line.replace("{root}", root).replace("{out}", out)
 
 
 def build_actual_output(root: str, out: str) -> List[str]:
     # Fake out that we have a project root
     binary = os.path.join(SCRIPT_ROOT, '../gn_to_cmakelists.py')
     project = os.path.join(SCRIPT_ROOT, "test_project.json")
     cmake = os.path.join(SCRIPT_ROOT, "../../../out/CMakeLists.txt")
-    retval = subprocess.run([
+    subprocess.run([
         binary,
         project,
     ], stdout=subprocess.PIPE, check=True, encoding='UTF-8', )
 
     with open(cmake, 'rt') as f:
-        for l in f.readlines():
-            yield l
+        for line in f.readlines():
+            yield line
 
 
 def main():
@@ -54,15 +54,15 @@ def main():
     ROOT = '/TEST/BUILD/ROOT'
     OUT = '/OUTPUT/DIR'
 
-    expected = [l for l in build_expected_output(ROOT, OUT)]
-    actual = [l for l in build_actual_output(ROOT, OUT)]
+    expected = [line for line in build_expected_output(ROOT, OUT)]
+    actual = [line for line in build_actual_output(ROOT, OUT)]
 
     diffs = [line for line in difflib.unified_diff(expected, actual)]
 
     if diffs:
         logging.error("DIFFERENCE between expected and generated output")
-        for l in diffs:
-            logging.warning("  " + l.strip())
+        for line in diffs:
+            logging.warning("  " + line.strip())
         sys.exit(1)
 
 

scripts/gen_test_driver.py

@@ -1,4 +1,3 @@
-
 #!/usr/bin/env python
 
 # Copyright (c) 2020 Project CHIP Authors
@@ -74,8 +73,8 @@ def main(argv):
     TEST_SUITE_RE = re.compile(r'\s*CHIP_REGISTER_TEST_SUITE\(([^)]*)\)')
 
     with open(options.input_file, 'r') as input_file:
-        for l in input_file.readlines():
-            match = TEST_SUITE_RE.match(l)
+        for line in input_file.readlines():
+            match = TEST_SUITE_RE.match(line)
             if not match:
                 continue
 

scripts/tests/chiptest/__init__.py

@@ -22,7 +22,7 @@
 from typing import Iterator, Set
 
 from . import linux, runner
-from .test_definition import ApplicationPaths, TestDefinition, TestRunTime, TestTag, TestTarget
+from .test_definition import ApplicationPaths, TestDefinition, TestTag, TestTarget
 
 _DEFAULT_CHIP_ROOT = os.path.abspath(
     os.path.join(os.path.dirname(__file__), "..", "..", ".."))
@@ -149,7 +149,13 @@ def _AllYamlTests():
 
 
 def target_for_name(name: str):
-    if name.startswith("TV_") or name.startswith("Test_TC_MC_") or name.startswith("Test_TC_LOWPOWER_") or name.startswith("Test_TC_KEYPADINPUT_") or name.startswith("Test_TC_APPLAUNCHER_") or name.startswith("Test_TC_MEDIAINPUT_") or name.startswith("Test_TC_WAKEONLAN_") or name.startswith("Test_TC_CHANNEL_") or name.startswith("Test_TC_MEDIAPLAYBACK_") or name.startswith("Test_TC_AUDIOOUTPUT_") or name.startswith("Test_TC_TGTNAV_") or name.startswith("Test_TC_APBSC_") or name.startswith("Test_TC_CONTENTLAUNCHER_") or name.startswith("Test_TC_ALOGIN_"):
+    if (name.startswith("TV_") or name.startswith("Test_TC_MC_") or
+            name.startswith("Test_TC_LOWPOWER_") or name.startswith("Test_TC_KEYPADINPUT_") or
+            name.startswith("Test_TC_APPLAUNCHER_") or name.startswith("Test_TC_MEDIAINPUT_") or
+            name.startswith("Test_TC_WAKEONLAN_") or name.startswith("Test_TC_CHANNEL_") or
+            name.startswith("Test_TC_MEDIAPLAYBACK_") or name.startswith("Test_TC_AUDIOOUTPUT_") or
+            name.startswith("Test_TC_TGTNAV_") or name.startswith("Test_TC_APBSC_") or
+            name.startswith("Test_TC_CONTENTLAUNCHER_") or name.startswith("Test_TC_ALOGIN_")):
         return TestTarget.TV
     if name.startswith("DL_") or name.startswith("Test_TC_DRLK_"):
         return TestTarget.LOCK

scripts/tests/chiptest/runner.py

@@ -56,8 +56,8 @@ def CapturedLogContains(self, txt: str, index=0):
         return False, len(self.captured_logs)
 
     def FindLastMatchingLine(self, matcher):
-        for l in reversed(self.captured_logs):
-            match = re.match(matcher, l)
+        for line in reversed(self.captured_logs):
+            match = re.match(matcher, line)
             if match:
                 return match
         return None

scripts/tests/chiptest/test_definition.py

@@ -15,14 +15,12 @@
 
 import logging
 import os
-import sys
 import threading
 import time
 import typing
 from dataclasses import dataclass, field
 from datetime import datetime
 from enum import Enum, auto
-from random import randrange
 
 TEST_NODE_ID = '0x12344321'
 
@@ -94,7 +92,7 @@ def wait(self, timeout=None):
             if self.killed:
                 return 0
             # If the App was never started, wait cannot be called on the process
-            if self.process == None:
+            if self.process is None:
                 time.sleep(0.1)
                 continue
             code = self.process.wait(timeout)
@@ -169,7 +167,8 @@ class ApplicationPaths:
     chip_tool_with_python_cmd: typing.List[str]
 
     def items(self):
-        return [self.chip_tool, self.all_clusters_app, self.lock_app, self.ota_provider_app, self.ota_requestor_app, self.tv_app, self.bridge_app, self.chip_repl_yaml_tester_cmd, self.chip_tool_with_python_cmd]
+        return [self.chip_tool, self.all_clusters_app, self.lock_app, self.ota_provider_app, self.ota_requestor_app,
+                self.tv_app, self.bridge_app, self.chip_repl_yaml_tester_cmd, self.chip_tool_with_python_cmd]
 
 
 @dataclass
@@ -253,7 +252,8 @@ def tags_str(self) -> str:
         """Get a human readable list of tags applied to this test"""
         return ", ".join([t.to_s() for t in self.tags])
 
-    def Run(self, runner, apps_register, paths: ApplicationPaths, pics_file: str, timeout_seconds: typing.Optional[int], dry_run=False, test_runtime: TestRunTime = TestRunTime.CHIP_TOOL_BUILTIN):
+    def Run(self, runner, apps_register, paths: ApplicationPaths, pics_file: str,
+            timeout_seconds: typing.Optional[int], dry_run=False, test_runtime: TestRunTime = TestRunTime.CHIP_TOOL_BUILTIN):
         """
         Executes the given test case using the provided runner for execution.
         """

scripts/tests/java/base.py

@@ -17,11 +17,8 @@
 #    limitations under the License.
 #
 
-import asyncio
 import datetime
 # Commissioning test.
-import logging
-import os
 import queue
 import subprocess
 import sys
@@ -39,7 +36,7 @@ def EnqueueLogOutput(fp, tag, q):
             try:
                 timestamp = float(line[1:18].decode())
                 line = line[19:]
-            except Exception as ex:
+            except Exception:
                 pass
         sys.stdout.buffer.write(
             (f"[{datetime.datetime.fromtimestamp(timestamp).isoformat(sep=' ')}]").encode() + tag + line)

scripts/tests/java/commissioning_test.py

@@ -18,12 +18,9 @@
 #
 
 import argparse
-import asyncio
 import logging
-import os
 import queue
 import subprocess
-import sys
 import threading
 import typing
 
@@ -46,7 +43,8 @@ def __init__(self, thread_list: typing.List[threading.Thread], queue: queue.Queu
         parser.add_argument('-s', '--setup-payload', dest='setup_payload',
                             help="Setup Payload (manual pairing code or QR code content)")
         parser.add_argument('-c', '--setup-pin-code', dest='setup_pin_code',
-                            help="Setup PIN code which can be used for password-authenticated session establishment (PASE) with the Commissionee")
+                            help=("Setup PIN code which can be used for password-authenticated "
+                                  "session establishment (PASE) with the Commissionee"))
         parser.add_argument('-n', '--nodeid', help="The Node ID issued to the device", default='1')
         parser.add_argument('-d', '--discriminator', help="Discriminator of the device", default='3840')
         parser.add_argument('-u', '--paa-trust-store-path', dest='paa_trust_store_path',

scripts/tests/java/discover_test.py

@@ -18,12 +18,9 @@
 #
 
 import argparse
-import asyncio
 import logging
-import os
 import queue
 import subprocess
-import sys
 import threading
 import typing
 

scripts/tests/run_java_test.py

@@ -16,8 +16,6 @@
 
 import logging
 import os
-import pathlib
-import pty
 import queue
 import re
 import shlex
@@ -34,12 +32,18 @@
 
 
 @click.command()
-@click.option("--app", type=click.Path(exists=True), default=None, help='Path to local application to use, omit to use external apps.')
-@click.option("--app-args", type=str, default='', help='The extra arguments passed to the device.')
-@click.option("--tool-path", type=click.Path(exists=True), default=None, help='Path to java-matter-controller.')
-@click.option("--tool-cluster", type=str, default='pairing', help='The cluster name passed to the java-matter-controller.')
-@click.option("--tool-args", type=str, default='', help='The arguments passed to the java-matter-controller.')
-@click.option("--factoryreset", is_flag=True, help='Remove app configs (/tmp/chip*) before running the tests.')
+@click.option("--app", type=click.Path(exists=True), default=None,
+              help='Path to local application to use, omit to use external apps.')
+@click.option("--app-args", type=str, default='',
+              help='The extra arguments passed to the device.')
+@click.option("--tool-path", type=click.Path(exists=True), default=None,
+              help='Path to java-matter-controller.')
+@click.option("--tool-cluster", type=str, default='pairing',
+              help='The cluster name passed to the java-matter-controller.')
+@click.option("--tool-args", type=str, default='',
+              help='The arguments passed to the java-matter-controller.')
+@click.option("--factoryreset", is_flag=True,
+              help='Remove app configs (/tmp/chip*) before running the tests.')
 def main(app: str, app_args: str, tool_path: str, tool_cluster: str, tool_args: str, factoryreset: bool):
     logging.info("Execute: {script_command}")