[chiptest] Add support for door lock app tests (#15412)

* [chiptest] Fix warnings reported by PEP8 checker

* [chiptest] Add support for door lock app tests

.github/workflows/tests.yaml

@@ -81,6 +81,7 @@ jobs:
                      "./scripts/build/build_examples.py \
                         --target linux-x64-chip-tool-${BUILD_VARIANT}${CHIP_TOOL_VARIANT} \
                         --target linux-x64-all-clusters-${BUILD_VARIANT}-test-group \
+                        --target linux-x64-door-lock-${BUILD_VARIANT} \
                         --target linux-x64-tv-app-${BUILD_VARIANT} \
                         build \
                         --copy-artifacts-to objdir-clone \
@@ -94,6 +95,7 @@ jobs:
                      run \
                      --iterations 1 \
                      --all-clusters-app ./out/linux-x64-all-clusters-${BUILD_VARIANT}-test-group/chip-all-clusters-app \
+                     --door-lock-app ./out/linux-x64-door-lock-${BUILD_VARIANT}/chip-door-lock-app \
                      --tv-app ./out/linux-x64-tv-app-${BUILD_VARIANT}/chip-tv-app \
                   "
             - name: Uploading core files
@@ -183,7 +185,7 @@ jobs:
                   ./scripts/run_in_build_env.sh \
                   "./scripts/tests/run_test_suite.py \
                      --chip-tool ./out/darwin-x64-chip-tool-${BUILD_VARIANT}${CHIP_TOOL_VARIANT}/chip-tool \
-                     --target-skip-glob '{TestGroupMessaging,TV_*}' \
+                     --target-skip-glob '{TestGroupMessaging,DL_*,TV_*}' \
                      run \
                      --iterations 1 \
                      --all-clusters-app ./out/darwin-x64-all-clusters-${BUILD_VARIANT}/chip-all-clusters-app \

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,11 +138,12 @@ 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(
         chip_tool='ip netns exec tool'.split() + paths.chip_tool,
         all_clusters_app='ip netns exec app'.split() + paths.all_clusters_app,
+        door_lock_app='ip netns exec app'.split() + paths.door_lock_app,
         tv_app='ip netns exec app'.split() + paths.tv_app,
     )

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)]
         app_cmd = app_cmd + ['--interface-id', str(-1)]
         return runner.RunSubprocess(app_cmd, name='APP ', wait=False)
@@ -118,8 +117,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:
@@ -147,6 +146,7 @@ class TestTarget(Enum):
 class ApplicationPaths:
     chip_tool: typing.List[str]
     all_clusters_app: typing.List[str]
+    door_lock_app: typing.List[str]
     tv_app: typing.List[str]
 
 
@@ -196,7 +196,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,12 +207,10 @@ def Run(self, runner, apps_register, paths: ApplicationPaths):
             elif self.target == TestTarget.TV:
                 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")
-                return
+                app_cmd = paths.door_lock_app
             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
 
@@ -226,16 +226,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