[chiptest] Fix warnings reported by PEP8 checker

scripts/tests/chiptest/__init__.py

@@ -14,16 +14,10 @@
 #    limitations under the License.
 #
 
-from pathlib import Path
-import os
-import logging
 import subprocess
-import re
 
-import chiptest.linux
-import chiptest.runner
-
-from .test_definition import TestTarget, TestDefinition, ApplicationPaths
+from . import linux, runner
+from .test_definition import ApplicationPaths, TestDefinition, TestTarget
 
 
 def AllTests(chip_tool: str):
@@ -46,4 +40,7 @@ def AllTests(chip_tool: str):
         yield TestDefinition(run_name=name, name=name, target=target)
 
 
-__all__ = ['TestTarget', 'TestDefinition', 'AllTests', 'ApplicationPaths']
+__all__ = [
+    'TestTarget', 'TestDefinition', 'AllTests', 'ApplicationPaths',
+    'linux', 'runner',
+]

scripts/tests/chiptest/accessories.py

@@ -13,13 +13,10 @@
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 
-import logging
-import time
-import threading
 import sys
-from random import randrange
-from xmlrpc.server import SimpleXMLRPCServer
+import threading
 from xmlrpc.client import ServerProxy
+from xmlrpc.server import SimpleXMLRPCServer
 
 IP = '127.0.0.1'
 PORT = 9000
@@ -114,9 +111,11 @@ def __startXMLRPCServer(self):
         self.server.register_function(self.reboot, 'reboot')
         self.server.register_function(self.factoryReset, 'factoryReset')
         self.server.register_function(
-            self.waitForCommissionableAdvertisement, 'waitForCommissionableAdvertisement')
+            self.waitForCommissionableAdvertisement,
+            'waitForCommissionableAdvertisement')
         self.server.register_function(
-            self.waitForOperationalAdvertisement, 'waitForOperationalAdvertisement')
+            self.waitForOperationalAdvertisement,
+            'waitForOperationalAdvertisement')
         self.server.register_function(self.ping, 'ping')
 
         self.server_thread = threading.Thread(target=self.__handle_request)
@@ -129,7 +128,8 @@ def __handle_request(self):
 
     def __stopXMLRPCServer(self):
         self.__should_handle_requests = False
-        # handle_request will wait until it receives a message, so let's send a ping to the server
+        # handle_request will wait until it receives a message,
+        # so let's send a ping to the server
         client = ServerProxy('http://' + IP + ':' +
                              str(PORT) + '/', allow_none=True)
         client.ping()

scripts/tests/chiptest/linux.py

@@ -35,8 +35,10 @@ def EnsureNetworkNamespaceAvailability():
         logging.warn("Running as root and this will change global namespaces.")
         return
 
-    os.execvpe("unshare", ["unshare", "--map-root-user", "-n", "-m", "python3",
-                           sys.argv[0], '--internal-inside-unshare'] + sys.argv[1:], test_environ)
+    os.execvpe(
+        "unshare", ["unshare", "--map-root-user", "-n", "-m", "python3",
+                    sys.argv[0], '--internal-inside-unshare'] + sys.argv[1:],
+        test_environ)
 
 
 def EnsurePrivateState():
@@ -110,8 +112,9 @@ def CreateNamespacesForAppTest():
             logging.error("Are you using --privileged if running in docker?")
             sys.exit(1)
 
-    # IPv6 does Duplicate Address  Detection even though
-    # we know ULAs provided are isolated. Wait for 'tenative' address to be gone
+    # IPv6 does Duplicate Address Detection even though
+    # we know ULAs provided are isolated. Wait for 'tenative'
+    # address to be gone.
 
     logging.info('Waiting for IPv6 DaD to complete (no tentative addresses)')
     for i in range(100):  # wait at most 10 seconds
@@ -135,7 +138,7 @@ def PrepareNamespacesForTestExecution(in_unshare: bool):
 
 def PathsWithNetworkNamespaces(paths: ApplicationPaths) -> ApplicationPaths:
     """
-    Returns a copy of paths with updated command arrays to invoke the 
+    Returns a copy of paths with updated command arrays to invoke the
     commands in an appropriate network namespace.
     """
     return ApplicationPaths(

scripts/tests/chiptest/runner.py

@@ -14,23 +14,19 @@
 
 import logging
 import os
+import pty
+import re
 import subprocess
 import sys
 import threading
-import time
-import pty
-import re
-
-from dataclasses import dataclass
 
 
 class LogPipe(threading.Thread):
 
     def __init__(self, level, capture_delegate=None, name=None):
-        """Setup the object with a logger and a loglevel
-
-            and start the thread
-            """
+        """
+        Setup the object with a logger and a loglevel and start the thread.
+        """
         threading.Thread.__init__(self)
 
         self.daemon = False
@@ -61,7 +57,7 @@ def FindLastMatchingLine(self, matcher):
         return None
 
     def fileno(self):
-        """Return the write file descriptor of the pipe"""
+        """Return the write file descriptor of the pipe."""
         return self.fd_write
 
     def run(self):
@@ -85,9 +81,11 @@ def __init__(self, capture_delegate=None):
 
     def RunSubprocess(self, cmd, name, wait=True, dependencies=[]):
         outpipe = LogPipe(
-            logging.DEBUG, capture_delegate=self.capture_delegate, name=name + ' OUT')
+            logging.DEBUG, capture_delegate=self.capture_delegate,
+            name=name + ' OUT')
         errpipe = LogPipe(
-            logging.INFO, capture_delegate=self.capture_delegate, name=name + ' ERR')
+            logging.INFO, capture_delegate=self.capture_delegate,
+            name=name + ' ERR')
 
         if self.capture_delegate:
             self.capture_delegate.Log(name, 'EXECUTING %r' % cmd)

scripts/tests/chiptest/test_definition.py

@@ -15,15 +15,12 @@
 
 import logging
 import os
+import threading
 import time
-from datetime import datetime
 import typing
-import threading
-from pathlib import Path
-import platform
-
-from enum import Enum, auto
 from dataclasses import dataclass
+from datetime import datetime
+from enum import Enum, auto
 from random import randrange
 
 TEST_NODE_ID = '0x12344321'
@@ -89,8 +86,9 @@ def waitForOperationalAdvertisement(self):
         return True
 
     def poll(self):
-        # When the server is manually stopped, process polling is overriden so the other
-        # processes that depends on the accessory beeing alive does not stop.
+        # When the server is manually stopped, process polling is overridden
+        # so the other processes that depends on the accessory beeing alive
+        # does not stop.
         if self.stopped:
             return None
         return self.process.poll()
@@ -105,7 +103,8 @@ def wait(self, duration):
 
     def __startServer(self, runner, command, discriminator):
         logging.debug(
-            'Executing application under test with discriminator %s.' % discriminator)
+            'Executing application under test with discriminator %s.' %
+            discriminator)
         app_cmd = command + ['--discriminator', str(discriminator)]
         return runner.RunSubprocess(app_cmd, name='APP ', wait=False)
 
@@ -117,8 +116,8 @@ def __waitFor(self, waitForString, server_process, outpipe):
             waitForString, self.lastLogIndex)
         while not ready:
             if server_process.poll() is not None:
-                died_str = 'Server died while waiting for %s, returncode %d' % (
-                    waitForString, server_process.returncode)
+                died_str = ('Server died while waiting for %s, returncode %d' %
+                            (waitForString, server_process.returncode))
                 logging.error(died_str)
                 raise Exception(died_str)
             if time.time() - start_time > 10:
@@ -195,7 +194,9 @@ class TestDefinition:
     target: TestTarget
 
     def Run(self, runner, apps_register, paths: ApplicationPaths):
-        """Executes the given test case using the provided runner for execution."""
+        """
+        Executes the given test case using the provided runner for execution.
+        """
         runner.capture_delegate = ExecutionCapture()
 
         try:
@@ -205,11 +206,12 @@ def Run(self, runner, apps_register, paths: ApplicationPaths):
                 app_cmd = paths.tv_app
             elif self.target == TestTarget.DOOR_LOCK:
                 logging.info(
-                    "Ignore test - test is made for door lock which is not supported yet")
+                    "Ignore test - test is made for door lock which"
+                    " is not supported yet")
                 return
             else:
-                raise Exception(
-                    "Unknown test target - don't know which application to run")
+                raise Exception("Unknown test target - "
+                                "don't know which application to run")
 
             tool_cmd = paths.chip_tool
 
@@ -225,16 +227,21 @@ def Run(self, runner, apps_register, paths: ApplicationPaths):
                     os.unlink(f)
 
             app = App(runner, app_cmd)
-            app.factoryReset()  # Remove server application storage, so it will be commissionable again
+            # Remove server application storage (factory reset),
+            # so it will be commissionable again.
+            app.factoryReset()
             app.start(str(randrange(1, 4096)))
             apps_register.add("default", app)
 
-            runner.RunSubprocess(tool_cmd + ['pairing', 'qrcode', TEST_NODE_ID, app.setupCode],
-                                 name='PAIR', dependencies=[apps_register])
+            runner.RunSubprocess(
+                tool_cmd + ['pairing', 'qrcode', TEST_NODE_ID, app.setupCode],
+                name='PAIR', dependencies=[apps_register])
+
+            runner.RunSubprocess(
+                tool_cmd + ['tests', self.run_name],
+                name='TEST', dependencies=[apps_register])
 
-            runner.RunSubprocess(tool_cmd + ['tests', self.run_name],
-                                 name='TEST', dependencies=[apps_register])
-        except:
+        except Exception:
             logging.error("!!!!!!!!!!!!!!!!!!!! ERROR !!!!!!!!!!!!!!!!!!!!!!")
             runner.capture_delegate.LogContents()
             raise

scripts/tests/run_test_suite.py

@@ -14,23 +14,22 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from chiptest.accessories import AppsRegister
-import coloredlogs
-import click
 import logging
 import os
 import shutil
 import sys
-import typing
 import time
-
-from pathlib import Path
+import typing
 from dataclasses import dataclass
+from pathlib import Path
+
+import click
+import coloredlogs
 
-sys.path.append(os.path.abspath(os.path.dirname(__file__)))
+import chiptest
+from chiptest.accessories import AppsRegister
+from chiptest.glob_matcher import GlobMatcher
 
-import chiptest  # noqa: E402
-from chiptest.glob_matcher import GlobMatcher  # noqa: E402
 
 DEFAULT_CHIP_ROOT = os.path.abspath(
     os.path.join(os.path.dirname(__file__), '..', '..'))
@@ -108,7 +107,8 @@ class RunContext:
     default=FindBinaryPath('chip-tool'),
     help='Binary path of chip tool app to use to run the test')
 @click.pass_context
-def main(context, log_level, target, target_glob, target_skip_glob, no_log_timestamps, root, internal_inside_unshare, chip_tool):
+def main(context, log_level, target, target_glob, target_skip_glob,
+         no_log_timestamps, root, internal_inside_unshare, chip_tool):
     # Ensures somewhat pretty logging of what is going on
     log_fmt = '%(asctime)s.%(msecs)03d %(levelname)-7s %(message)s'
     if no_log_timestamps:
@@ -152,7 +152,7 @@ def main(context, log_level, target, target_glob, target_skip_glob, no_log_times
 @main.command(
     'list', help='List available test suites')
 @click.pass_context
-def cmd_generate(context):
+def cmd_list(context):
     for test in context.obj.tests:
         print(test.name)
 
@@ -190,7 +190,8 @@ def cmd_run(context, iterations, all_clusters_app, tv_app):
     # Testing prerequisites: tv app requires a config. Copy it just in case
     shutil.copyfile(
         os.path.join(
-            context.obj.root, 'examples/tv-app/linux/include/endpoint-configuration/chip_tv_config.ini'),
+            context.obj.root, ('examples/tv-app/linux/include/'
+                               'endpoint-configuration/chip_tv_config.ini')),
         '/tmp/chip_tv_config.ini'
     )
 
@@ -208,7 +209,7 @@ def cmd_run(context, iterations, all_clusters_app, tv_app):
                 test_end = time.time()
                 logging.info('%-20s - Completed in %0.2f seconds' %
                              (test.name, (test_end - test_start)))
-            except:
+            except Exception:
                 test_end = time.time()
                 logging.exception('%s - FAILED in %0.2f seconds' %
                                   (test.name, (test_end - test_start)))
@@ -221,9 +222,11 @@ def cmd_run(context, iterations, all_clusters_app, tv_app):
 # On linux, allow an execution shell to be prepared
 if sys.platform == 'linux':
     @main.command(
-        'shell', help='Execute a bash shell in the environment (useful to test network namespaces)')
+        'shell',
+        help=('Execute a bash shell in the environment (useful to test '
+              'network namespaces)'))
     @click.pass_context
-    def cmd_run(context):
+    def cmd_shell(context):
         chiptest.linux.PrepareNamespacesForTestExecution(
             context.obj.in_unshare)
         os.execvpe("bash", ["bash"], os.environ.copy())