[Tests] Properly clean up test suite tests on KeyboardInterrupt

scripts/tests/chiptest/accessories.py

@@ -43,6 +43,9 @@ def init(self):
     def uninit(self):
         self.__stopXMLRPCServer()
 
+    def terminate(self):
+        self.uninit()
+
     @property
     def accessories(self):
         """List of registered accessory applications."""

scripts/tests/chiptest/linux.py

@@ -29,7 +29,7 @@
 
 import sdbus
 
-from .runner import Executor, SubprocessInfo
+from .runner import Executor, SubprocessInfo, SubprocessKind
 
 log = logging.getLogger(__name__)
 
@@ -136,9 +136,8 @@ class IsolatedNetworkNamespace:
         "ip netns del app-{index}",
     ]
 
-    def __init__(self, index=0, setup_app_link_up=True, setup_tool_link_up=True,
-                 app_link_name='eth-app', tool_link_name='eth-tool',
-                 unshared=False):
+    def __init__(self, index: int = 0, setup_app_link_up: bool = True, setup_tool_link_up: bool = True,
+                 wait_for_dad: bool = True, app_link_name: str = 'eth-app', tool_link_name: str = 'eth-tool', unshared: bool = False):
 
         if not unshared:
             # If not running in an unshared network namespace yet, try
@@ -151,17 +150,24 @@ def __init__(self, index=0, setup_app_link_up=True, setup_tool_link_up=True,
         self.app_link_name = app_link_name
         self.tool_link_name = tool_link_name
 
-        self._setup()
-        if setup_app_link_up:
-            self._setup_app_link_up(wait_for_dad=False)
-        if setup_tool_link_up:
-            self._setup_tool_link_up(wait_for_dad=False)
-        self._wait_for_duplicate_address_detection()
+        try:
+            self._setup()
+            if setup_app_link_up:
+                self._setup_app_link_up(wait_for_dad=False)
+            if setup_tool_link_up:
+                self._setup_tool_link_up(wait_for_dad=False)
+        except Exception:
+            # Ensure that we leave a clean state on any exception.
+            self.terminate()
+            raise
 
-    def netns_for_subprocess(self, subproc: SubprocessInfo):
-        return "{}-{}".format(subproc.kind, self.index)
+        if wait_for_dad:
+            self.wait_for_duplicate_address_detection()
+
+    def netns_for_subprocess_kind(self, kind: SubprocessKind):
+        return "{}-{}".format(kind.name.lower(), self.index)
 
-    def _wait_for_duplicate_address_detection(self):
+    def wait_for_duplicate_address_detection(self):
         # IPv6 does Duplicate Address Detection even though
         # we know ULAs provided are isolated. Wait for 'tentative'
         # address to be gone.
@@ -182,35 +188,36 @@ def _setup_app_link_up(self, wait_for_dad=True):
         for command in self.COMMANDS_APP_LINK_UP:
             self._run(command)
         if wait_for_dad:
-            self._wait_for_duplicate_address_detection()
+            self.wait_for_duplicate_address_detection()
 
     def _setup_tool_link_up(self, wait_for_dad=True):
         for command in self.COMMANDS_TOOL_LINK_UP:
             self._run(command)
         if wait_for_dad:
-            self._wait_for_duplicate_address_detection()
-
-    def _run(self, command: str):
-        command = command.format(app_link_name=self.app_link_name,
-                                 tool_link_name=self.tool_link_name,
-                                 index=self.index)
-        log.debug("Executing: '%s'", command)
-        if subprocess.run(command.split()).returncode != 0:
-            log.error("Failed to execute '%s'", command)
-            log.error("Are you using --privileged if running in docker?")
-            sys.exit(1)
+            self.wait_for_duplicate_address_detection()
+
+    def _run(self, *command: str):
+        for c in command:
+            c = c.format(app_link_name=self.app_link_name, tool_link_name=self.tool_link_name, index=self.index)
+            log.debug("Executing: '%s'", c)
+            if subprocess.run(c.split()).returncode != 0:
+                raise RuntimeError(f"Failed to execute '{c}'. Are you using --privileged if running in docker?")
 
     def terminate(self):
-        for command in self.COMMANDS_TERMINATE:
-            self._run(command)
+        """Execute all down commands gracefully omitting errors."""
+        for cmd in self.COMMANDS_TERMINATE:
+            try:
+                self._run(cmd)
+            except Exception as e:
+                log.warning("Encountered error during namespace termination: %s", e)
 
 
 class LinuxNamespacedExecutor(Executor):
     def __init__(self, ns: IsolatedNetworkNamespace):
         self.ns = ns
 
     def run(self, subproc: SubprocessInfo, stdin=None, stdout=None, stderr=None):
-        wrapped = subproc.wrap_with("ip", "netns", "exec", self.ns.netns_for_subprocess(subproc))
+        wrapped = subproc.wrap_with("ip", "netns", "exec", self.ns.netns_for_subprocess_kind(subproc.kind))
         return subprocess.Popen(wrapped.to_cmd(), stdin=stdin, stdout=stdout, stderr=stderr)
 
 

scripts/tests/chiptest/runner.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import enum
 import logging
 import os
 import pathlib
@@ -21,8 +22,7 @@
 import subprocess
 import threading
 import typing
-from dataclasses import dataclass
-from typing import Literal
+from dataclasses import dataclass, replace
 
 log = logging.getLogger(__name__)
 
@@ -124,11 +124,15 @@ def get(self):
         return self.queue.get()
 
 
+class SubprocessKind(enum.Enum):
+    APP = enum.auto()
+    TOOL = enum.auto()
+    RPC = enum.auto()
+
+
 @dataclass
 class SubprocessInfo:
-    # Restricted as this identifies the name of the network namespace in an executor implementing
-    # test case isolation.
-    kind: Literal['app', 'tool']
+    kind: SubprocessKind
     path: pathlib.Path | str
     wrapper: tuple[str, ...] = ()
     args: tuple[str, ...] = ()
@@ -137,10 +141,10 @@ def __post_init__(self):
         self.path = pathlib.Path(self.path)
 
     def with_args(self, *args: str):
-        return SubprocessInfo(kind=self.kind, path=self.path, wrapper=self.wrapper, args=self.args + tuple(args))
+        return replace(self, args=self.args + tuple(args))
 
     def wrap_with(self, *args: str):
-        return SubprocessInfo(kind=self.kind, path=self.path, wrapper=tuple(args) + self.wrapper, args=self.args)
+        return replace(self, wrapper=tuple(args) + self.wrapper)
 
     def to_cmd(self) -> typing.List[str]:
         return list(self.wrapper) + [str(self.path)] + list(self.args)

scripts/tests/run_darwin_framework_ota_test.py

@@ -9,7 +9,7 @@
 import click
 from chiptest.accessories import AppsRegister
 from chiptest.darwin import DarwinExecutor
-from chiptest.runner import Runner, SubprocessInfo
+from chiptest.runner import Runner, SubprocessInfo, SubprocessKind
 from chiptest.test_definition import App, ExecutionCapture
 from chipyaml.paths_finder import PathsFinder
 
@@ -115,10 +115,10 @@ def cmd_run(context, darwin_framework_tool, ota_requestor_app, ota_data_file, ot
         ota_requestor_app = paths_finder.get('chip-ota-requestor-app')
 
     if darwin_framework_tool is not None:
-        darwin_framework_tool = SubprocessInfo(kind='tool', path=darwin_framework_tool)
+        darwin_framework_tool = SubprocessInfo(kind=SubprocessKind.TOOL, path=darwin_framework_tool)
 
     if ota_requestor_app is not None:
-        ota_requestor_app = SubprocessInfo(kind='app', path=ota_requestor_app,
+        ota_requestor_app = SubprocessInfo(kind=SubprocessKind.APP, path=ota_requestor_app,
                                            args=('--otaDownloadPath', ota_destination_file))
 
     runner = Runner(executor=DarwinExecutor())