cleveragents-core/features/mocks/mock_devcontainer_cli.py

"""Mock subprocess runner for devcontainer CLI commands.

Provides configurable mock responses for ``devcontainer up``,
``devcontainer exec``, and ``docker stop`` commands used in
lifecycle BDD tests.

All mocks reside in ``features/mocks/`` per project convention.
"""

from __future__ import annotations

import json
from dataclasses import dataclass, field


@dataclass
class MockProcessResult:
    """Mimics ``subprocess.CompletedProcess`` for test assertions."""

    returncode: int = 0
    stdout: str = ""
    stderr: str = ""
    args: list[str] = field(default_factory=list)


class MockDevcontainerRunner:
    """Configurable mock for subprocess.run calls.

    Records all invocations and returns pre-configured results
    based on the command being run.

    Usage::

        runner = MockDevcontainerRunner()
        runner.set_up_result(container_id="abcdef012345", workspace="/ws")
        # pass runner as ``run_command=runner`` in handler calls

    Attributes:
        calls: List of all recorded (args, kwargs) invocations.
        up_result: Result returned for ``devcontainer up`` commands.
        exec_result: Result returned for ``devcontainer exec`` commands.
        stop_result: Result returned for ``docker stop`` commands.
    """

    def __init__(self) -> None:
        self.calls: list[tuple[list[str], dict[str, object]]] = []
        self.up_result: MockProcessResult = MockProcessResult()
        self.exec_result: MockProcessResult = MockProcessResult()
        self.stop_result: MockProcessResult = MockProcessResult()

    #: TEST-4 fix: expected subprocess kwargs that production code must pass.
    #: If production code sets ``check=True`` (causing CalledProcessError on
    #: non-zero rc) or omits ``capture_output``/``text``, this mock will
    #: flag the mismatch.
    _EXPECTED_KWARGS = frozenset({"capture_output", "text", "timeout", "check"})

    def __call__(
        self,
        args: list[str],
        **kwargs: object,
    ) -> MockProcessResult:
        """Handle a subprocess.run invocation.

        Args:
            args: Command-line arguments.
            **kwargs: Additional subprocess kwargs — validated against
                expected production kwargs (TEST-4 fix).

        Returns:
            Mock result based on the command.

        Raises:
            AssertionError: If required kwargs are missing or ``check``
                is set to ``True`` (which would raise
                ``CalledProcessError`` in production).
        """
        self.calls.append((list(args), dict(kwargs)))

        # TEST-4 fix: validate subprocess kwargs.
        missing = self._EXPECTED_KWARGS - set(kwargs)
        if missing:
            raise AssertionError(
                f"MockDevcontainerRunner: missing expected subprocess kwargs "
                f"{missing} in call {args}"
            )
        if kwargs.get("check") is True:
            raise AssertionError(
                f"MockDevcontainerRunner: check=True would raise "
                f"CalledProcessError in production for {args}"
            )

        if len(args) >= 2 and args[0] == "devcontainer" and args[1] == "up":
            return self.up_result
        if len(args) >= 2 and args[0] == "devcontainer" and args[1] == "exec":
            return self.exec_result
        if len(args) >= 2 and args[0] == "docker" and args[1] == "stop":
            return self.stop_result

        return MockProcessResult(returncode=0)

    def set_up_result(
        self,
        *,
        container_id: str = "aabbccddee0011223344",
        workspace: str = "/workspaces/project",
        returncode: int = 0,
        stderr: str = "",
    ) -> None:
        """Configure the result for ``devcontainer up``.

        Args:
            container_id: Container ID in the JSON output.
            workspace: Workspace path in the JSON output.
            returncode: Exit code.
            stderr: Stderr text.
        """
        output = json.dumps(
            {
                "outcome": "success" if returncode == 0 else "error",
                "containerId": container_id,
                "remoteWorkspaceFolder": workspace,
            }
        )
        self.up_result = MockProcessResult(
            returncode=returncode,
            stdout=output,
            stderr=stderr,
        )

    def set_up_no_container_id(self) -> None:
        """Configure ``devcontainer up`` to succeed without a containerId."""
        output = json.dumps({"outcome": "success"})
        self.up_result = MockProcessResult(
            returncode=0,
            stdout=output,
            stderr="",
        )

    def set_up_failure(self, stderr: str = "container build failed") -> None:
        """Configure ``devcontainer up`` to fail.

        Args:
            stderr: Error message.
        """
        self.up_result = MockProcessResult(
            returncode=1,
            stdout="",
            stderr=stderr,
        )

    def set_exec_result(self, *, returncode: int = 0, stdout: str = "ping") -> None:
        """Configure the result for ``devcontainer exec``.

        Args:
            returncode: Exit code.
            stdout: Stdout text.
        """
        self.exec_result = MockProcessResult(
            returncode=returncode,
            stdout=stdout,
        )

    def set_exec_failure(self) -> None:
        """Configure ``devcontainer exec`` to fail."""
        self.exec_result = MockProcessResult(
            returncode=1,
            stdout="",
            stderr="container not responding",
        )

    def set_stop_result(self, *, returncode: int = 0) -> None:
        """Configure the result for ``docker stop``.

        Args:
            returncode: Exit code.
        """
        self.stop_result = MockProcessResult(returncode=returncode)

    @property
    def call_count(self) -> int:
        """Total number of invocations."""
        return len(self.calls)

    @property
    def up_calls(self) -> list[tuple[list[str], dict[str, object]]]:
        """Calls to ``devcontainer up``."""
        return [
            (a, k)
            for a, k in self.calls
            if len(a) >= 2 and a[0] == "devcontainer" and a[1] == "up"
        ]

    @property
    def exec_calls(self) -> list[tuple[list[str], dict[str, object]]]:
        """Calls to ``devcontainer exec``."""
        return [
            (a, k)
            for a, k in self.calls
            if len(a) >= 2 and a[0] == "devcontainer" and a[1] == "exec"
        ]

    @property
    def stop_calls(self) -> list[tuple[list[str], dict[str, object]]]:
        """Calls to ``docker stop``."""
        return [
            (a, k)
            for a, k in self.calls
            if len(a) >= 2 and a[0] == "docker" and a[1] == "stop"
        ]