[Fix] Reuse of Admin Storage To Skip Commissioning

app/container_manager/docker_shell_commands.py

@@ -18,7 +18,7 @@
 Useful for debugging and understanding what Docker operations are being performed.
 """
 from pathlib import Path
-from typing import Dict, List, Union
+from typing import Dict, List, Optional, Union
 
 # Log message constants
 SHELL_CMD_LOG_PREFIX = "Docker API call equivalent shell command:\n"
@@ -42,18 +42,25 @@
 ]
 
 
-def escape_shell_arg(arg: str) -> str:
+def escape_shell_arg(arg: Optional[str]) -> str:
     """
     Escape shell argument if it contains spaces or special characters.
 
     Uses single-quote wrapping for safety. Any single quotes in the argument
     are escaped using the pattern: ' becomes '\''
     (close quote, escaped quote, open quote).
 
+    Args:
+        arg: The argument to escape. Docker container/image names can be None
+            (e.g. a Container whose attrs have no "Name" key), so this is
+            tolerated and rendered as an empty string rather than raising.
+
     Returns:
         The argument wrapped in single quotes if it contains special characters,
         otherwise returns the argument unchanged.
     """
+    if arg is None:
+        return ""
     if any(c in arg for c in SHELL_SPECIAL_CHARS):
         # Escape any single quotes: ' becomes '\''
         escaped_arg = arg.replace("'", "'\\''")
@@ -193,7 +200,7 @@ def docker_exec_command(
     return " ".join(cmd_parts)
 
 
-def docker_kill_command(container_name: str) -> str:
+def docker_kill_command(container_name: Optional[str]) -> str:
     """
     Generate docker kill command.
 
@@ -206,7 +213,7 @@ def docker_kill_command(container_name: str) -> str:
     return f"docker kill {escape_shell_arg(container_name)}"
 
 
-def docker_stop_command(container_name: str) -> str:
+def docker_stop_command(container_name: Optional[str]) -> str:
     """
     Generate docker stop command.
 
@@ -219,7 +226,7 @@ def docker_stop_command(container_name: str) -> str:
     return f"docker stop {escape_shell_arg(container_name)}"
 
 
-def docker_rm_command(container_name: str, force: bool = False) -> str:
+def docker_rm_command(container_name: Optional[str], force: bool = False) -> str:
     """
     Generate docker rm command.
 

test_collections/matter/sdk_tests/support/python_testing/models/test_suite.py

@@ -14,7 +14,7 @@
 # limitations under the License.
 #
 from enum import Enum
-from typing import Type, TypeVar
+from typing import Optional, Type, TypeVar
 
 from app.constants.shared_constants import DutPairingModeEnum
 from app.schemas.test_environment_config import ThreadAutoConfig
@@ -30,6 +30,7 @@
 from ...utils import PromptOption, prompt_for_commissioning_mode
 from .utils import (
     DUTCommissioningError,
+    capture_admin_storage_file,
     commission_device,
     should_perform_new_commissioning,
 )
@@ -57,6 +58,7 @@ class PythonTestSuite(TestSuite):
     suite_name: str
     sdk_container: SDKContainer = SDKContainer(logger)
     border_router: ThreadBorderRouter = ThreadBorderRouter()
+    matter_config: Optional[TestEnvironmentConfigMatter] = None
 
     @classmethod
     def class_factory(
@@ -117,6 +119,18 @@ async def setup(self) -> None:
     async def cleanup(self) -> None:
         logger.info("Suite Cleanup")
 
+        if self.matter_config is not None and self.sdk_container.is_running():
+            try:
+                logger.info(
+                    "Capturing latest admin_storage.json snapshot from container"
+                )
+                capture_admin_storage_file(self.matter_config, logger)
+            except Exception as e:
+                # Deliberately broad Exception.
+                # The ideia is to never block container/border-router teardown below,
+                # so don't narrow this to specific exception types.
+                logger.warning(f"Could not capture admin_storage.json snapshot: {e}")
+
         logger.info("Stopping SDK container")
         self.sdk_container.destroy()
 
@@ -127,6 +141,7 @@ async def cleanup(self) -> None:
 class CommissioningPythonTestSuite(PythonTestSuite, UserPromptSupport):
     async def setup(self) -> None:
         await super().setup()
+        assert self.matter_config is not None
 
         # If in BLE-Thread, NFC-Thread, or THREAD_MESHCOP mode and a Thread Auto-Config
         # was provided by the user, start a new OTBR container app with the according

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

@@ -207,6 +207,17 @@ def __copy_admin_storage_file(
     )
 
 
+def capture_admin_storage_file(
+    config: TestEnvironmentConfigMatter,
+    logger: loguru.Logger,
+) -> None:
+    """Re-capture admin_storage.json from the still-running container to the host
+    snapshot, so the snapshot reflects the message counters as they stood at the end
+    of this run rather than only as they stood right after the last commissioning.
+    """
+    __copy_admin_storage_file(config, logger)
+
+
 def log_test_output_file(logger: loguru.Logger) -> None:
     """Log the entire content of test_output.txt file.
 
@@ -262,7 +273,10 @@ async def commission_device(
     logger.info("---- End of commissioning test output ----")
 
     # Copy admin_storage.json file from container, in case the user wants to
-    # reuse this information in the next execution
+    # reuse this information in the next execution. This duplicates the capture
+    # PythonTestSuite.cleanup() does unconditionally at the end of the suite run,
+    # but is kept intentionally: if the container is torn down abnormally before
+    # cleanup() runs, this is the only snapshot that survives.
     __copy_admin_storage_file(config, logger)
 
 

test_collections/matter/sdk_tests/support/tests/python_tests/test_python_test_suite.py

@@ -444,6 +444,128 @@ async def test_should_perform_new_commissioning_yes() -> None:
         mock_commission_device.assert_called_once()
 
 
+@pytest.mark.asyncio
+async def test_cleanup_captures_admin_storage_before_destroy() -> None:
+    """Issue #1070: cleanup should re-capture admin_storage.json from the
+    still-running container before the container is destroyed, so the host
+    snapshot reflects this run's advanced message counters, not just the
+    counters from the original commissioning."""
+    suite_class: Type[PythonTestSuite] = PythonTestSuite.class_factory(
+        suite_type=SuiteType.COMMISSIONING,
+        name="SomeSuite",
+        python_test_version="Some version",
+        mandatory=False,
+    )
+    suite_instance = suite_class(TestSuiteExecution())
+    suite_instance.matter_config = mock.Mock()
+
+    with mock.patch.object(
+        target=suite_instance.sdk_container, attribute="is_running", return_value=True
+    ), mock.patch.object(
+        target=suite_instance.sdk_container, attribute="destroy"
+    ) as mock_destroy, mock.patch.object(
+        target=suite_instance.border_router, attribute="destroy_device"
+    ), mock.patch(
+        "test_collections.matter.sdk_tests.support.python_testing.models.test_suite"
+        ".capture_admin_storage_file"
+    ) as mock_capture:
+        await suite_instance.cleanup()
+
+    mock_capture.assert_called_once_with(
+        suite_instance.matter_config, test_engine_logger
+    )
+    mock_destroy.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_cleanup_skips_capture_when_container_not_running() -> None:
+    """If the container already died mid-suite, cleanup should skip the capture
+    attempt (nothing to copy from) rather than attempting it against a dead
+    container, but should still tear down the container and border router."""
+    suite_class: Type[PythonTestSuite] = PythonTestSuite.class_factory(
+        suite_type=SuiteType.COMMISSIONING,
+        name="SomeSuite",
+        python_test_version="Some version",
+        mandatory=False,
+    )
+    suite_instance = suite_class(TestSuiteExecution())
+    suite_instance.matter_config = mock.Mock()
+
+    with mock.patch.object(
+        target=suite_instance.sdk_container, attribute="is_running", return_value=False
+    ), mock.patch.object(
+        target=suite_instance.sdk_container, attribute="destroy"
+    ) as mock_destroy, mock.patch.object(
+        target=suite_instance.border_router, attribute="destroy_device"
+    ) as mock_destroy_device, mock.patch(
+        "test_collections.matter.sdk_tests.support.python_testing.models.test_suite"
+        ".capture_admin_storage_file"
+    ) as mock_capture:
+        await suite_instance.cleanup()
+
+    mock_capture.assert_not_called()
+    mock_destroy.assert_called_once()
+    mock_destroy_device.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_cleanup_skips_capture_when_matter_config_not_set() -> None:
+    """If setup failed before matter_config was assigned, cleanup should skip the
+    capture attempt (no config to resolve a storage path from) but still tear down
+    the container and border router."""
+    suite_class: Type[PythonTestSuite] = PythonTestSuite.class_factory(
+        suite_type=SuiteType.COMMISSIONING,
+        name="SomeSuite",
+        python_test_version="Some version",
+        mandatory=False,
+    )
+    suite_instance = suite_class(TestSuiteExecution())
+
+    with mock.patch.object(
+        target=suite_instance.sdk_container, attribute="destroy"
+    ) as mock_destroy, mock.patch.object(
+        target=suite_instance.border_router, attribute="destroy_device"
+    ) as mock_destroy_device, mock.patch(
+        "test_collections.matter.sdk_tests.support.python_testing.models.test_suite"
+        ".capture_admin_storage_file"
+    ) as mock_capture:
+        await suite_instance.cleanup()
+
+    mock_capture.assert_not_called()
+    mock_destroy.assert_called_once()
+    mock_destroy_device.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_cleanup_still_destroys_container_when_capture_fails() -> None:
+    """A capture failure must never prevent container/border-router teardown, and
+    must not be reported as a suite-level error."""
+    suite_class: Type[PythonTestSuite] = PythonTestSuite.class_factory(
+        suite_type=SuiteType.COMMISSIONING,
+        name="SomeSuite",
+        python_test_version="Some version",
+        mandatory=False,
+    )
+    suite_instance = suite_class(TestSuiteExecution())
+    suite_instance.matter_config = mock.Mock()
+
+    with mock.patch.object(
+        target=suite_instance.sdk_container, attribute="is_running", return_value=True
+    ), mock.patch.object(
+        target=suite_instance.sdk_container, attribute="destroy"
+    ) as mock_destroy, mock.patch.object(
+        target=suite_instance.border_router, attribute="destroy_device"
+    ) as mock_destroy_device, mock.patch(
+        "test_collections.matter.sdk_tests.support.python_testing.models.test_suite"
+        ".capture_admin_storage_file",
+        side_effect=RuntimeError("boom"),
+    ):
+        await suite_instance.cleanup()
+
+    mock_destroy.assert_called_once()
+    mock_destroy_device.assert_called_once()
+
+
 @pytest.mark.asyncio
 async def test_should_perform_new_commissioning_no() -> None:
     """Test that when should_perform_new_commissioning returns False,

test_collections/matter/sdk_tests/support/tests/python_tests/test_utils.py

@@ -31,6 +31,7 @@
     EXECUTABLE,
     RUNNER_CLASS_PATH,
     DUTCommissioningError,
+    capture_admin_storage_file,
     commission_device,
     generate_command_arguments,
 )
@@ -563,6 +564,37 @@ async def test_commission_device_failure() -> None:
     mock_handle_logs.assert_called_once()
 
 
+# ---------------------------------------------------------------------------
+# Tests for capture_admin_storage_file (re-snapshot at suite cleanup, issue #1070)
+# ---------------------------------------------------------------------------
+
+
+def test_capture_admin_storage_file_copies_from_container() -> None:
+    sdk_container: SDKContainer = SDKContainer()
+
+    with mock.patch.object(
+        target=sdk_container, attribute="copy_file_from_container"
+    ) as mock_copy:
+        capture_admin_storage_file(
+            default_environment_config, test_engine_logger  # type: ignore
+        )
+
+    mock_copy.assert_called_once()
+
+
+def test_capture_admin_storage_file_propagates_exceptions() -> None:
+    sdk_container: SDKContainer = SDKContainer()
+
+    with mock.patch.object(
+        target=sdk_container,
+        attribute="copy_file_from_container",
+        side_effect=RuntimeError("boom"),
+    ), pytest.raises(RuntimeError):
+        capture_admin_storage_file(
+            default_environment_config, test_engine_logger  # type: ignore
+        )
+
+
 # ---------------------------------------------------------------------------
 # Tests for the new typed-arg / json-arg handling in generate_command_arguments
 # ---------------------------------------------------------------------------