cleveragents-core/robot/helper_devcontainer_lifecycle.py

"""Helper utilities for devcontainer lifecycle Robot smoke tests.

Each command prints a single ``<command>-ok`` token on success so the
calling Robot test can assert on ``stdout``.
"""

from __future__ import annotations

import json
import sys
from dataclasses import dataclass, field

from cleveragents.application.services.cleanup_service import CleanupService
from cleveragents.domain.models.core.container_lifecycle import (
    ContainerLifecycleState,
    ContainerLifecycleTracker,
    transition_state,
)
from cleveragents.resource.handlers.devcontainer import (
    _parse_devcontainer_up_output,
    activate_container,
    clear_lifecycle_registry,
    get_lifecycle_tracker,
    rebuild_container,
    set_lifecycle_tracker,
    stop_all_active_containers,
    stop_container,
)

# ---------------------------------------------------------------------------
# F29 fix: removed conditional import with ``# type: ignore`` suppressions.
# The inline mock classes are always used — they are self-contained and
# avoid the PYTHONPATH dependency on ``features/``, which is not always
# available in standalone Robot runs.
# ---------------------------------------------------------------------------


@dataclass
class _MockResult:
    """Minimal subprocess.CompletedProcess stand-in."""

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


class _MockRunner:
    """Callable mock for subprocess.run used by lifecycle functions."""

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

    def __call__(self, args: list[str], **kwargs: object) -> _MockResult:
        self.calls.append((list(args), dict(kwargs)))
        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 _MockResult(returncode=0)

    def set_up_result(
        self,
        *,
        container_id: str = "aabbccddee0011223344",
        workspace: str = "/workspaces/project",
        returncode: int = 0,
    ) -> None:
        output = json.dumps(
            {
                "outcome": "success" if returncode == 0 else "error",
                "containerId": container_id,
                "remoteWorkspaceFolder": workspace,
            }
        )
        self.up_result = _MockResult(returncode=returncode, stdout=output)

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

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


# ---------------------------------------------------------------------------
# Test commands
# ---------------------------------------------------------------------------


def cmd_enum_values() -> None:
    """Verify all six lifecycle states exist."""
    expected = {"detected", "building", "running", "stopping", "stopped", "failed"}
    actual = {s.value for s in ContainerLifecycleState}
    assert actual == expected, f"Expected {expected}, got {actual}"
    print("enum-values-ok")


def cmd_transition_valid() -> None:
    """Verify detected->building transition."""
    clear_lifecycle_registry()
    tracker = ContainerLifecycleTracker(resource_id="01ROBOTTEST0000000000001")
    tracker = transition_state(tracker, ContainerLifecycleState.BUILDING, reason="test")
    assert tracker.current_state == ContainerLifecycleState.BUILDING
    assert len(tracker.transitions) == 1
    print("transition-valid-ok")


def cmd_transition_invalid() -> None:
    """Verify invalid transitions raise ValueError."""
    clear_lifecycle_registry()
    tracker = ContainerLifecycleTracker(resource_id="01ROBOTTEST0000000000002")
    try:
        transition_state(tracker, ContainerLifecycleState.RUNNING, reason="bad")
        raise AssertionError("Should have raised ValueError")
    except ValueError:
        pass
    print("transition-invalid-ok")


def cmd_lazy_activation() -> None:
    """Verify lazy activation with mock runner."""
    clear_lifecycle_registry()
    runner = _MockRunner()
    runner.set_up_result(container_id="aabbccddee0022334455", workspace="/ws")
    result = activate_container(
        "01ROBOTTEST0000000000010",
        "/workspace",
        run_command=runner,
    )
    assert result.current_state == ContainerLifecycleState.RUNNING
    assert result.container_id == "aabbccddee0022334455"
    assert len(runner.up_calls) == 1
    print("lazy-activation-ok")


def cmd_stop_container() -> None:
    """Verify stop transitions active to stopped."""
    clear_lifecycle_registry()
    runner = _MockRunner()
    tracker = ContainerLifecycleTracker(
        resource_id="01ROBOTTEST0000000000020",
        current_state=ContainerLifecycleState.RUNNING,
        container_id="ctr-stop",
    )
    set_lifecycle_tracker(tracker)
    stop_container("01ROBOTTEST0000000000020", run_command=runner)
    final = get_lifecycle_tracker("01ROBOTTEST0000000000020")
    assert final.current_state == ContainerLifecycleState.STOPPED
    assert len(runner.stop_calls) == 1
    print("stop-container-ok")


def cmd_rebuild_container() -> None:
    """Verify rebuild transitions stopped to running."""
    clear_lifecycle_registry()
    runner = _MockRunner()
    runner.set_up_result(container_id="aabbccddee0033445566", workspace="/ws")
    tracker = ContainerLifecycleTracker(
        resource_id="01ROBOTTEST0000000000030",
        current_state=ContainerLifecycleState.STOPPED,
    )
    set_lifecycle_tracker(tracker)
    # R16 fix: call rebuild_container instead of activate_container
    rebuild_container(
        "01ROBOTTEST0000000000030",
        "/workspace",
        run_command=runner,
    )
    final = get_lifecycle_tracker("01ROBOTTEST0000000000030")
    assert final.current_state == ContainerLifecycleState.RUNNING
    print("rebuild-container-ok")


def cmd_session_cleanup() -> None:
    """Verify session cleanup stops all active containers."""
    clear_lifecycle_registry()
    runner = _MockRunner()
    for rid in ("01ROBOTTEST0000000000040", "01ROBOTTEST0000000000041"):
        tracker = ContainerLifecycleTracker(
            resource_id=rid,
            current_state=ContainerLifecycleState.RUNNING,
            container_id=f"ctr-{rid[-3:]}",
        )
        set_lifecycle_tracker(tracker)
    stopped = stop_all_active_containers(run_command=runner, session_id="ses-001")
    assert len(stopped) == 2
    for rid in ("01ROBOTTEST0000000000040", "01ROBOTTEST0000000000041"):
        t = get_lifecycle_tracker(rid)
        assert t.current_state == ContainerLifecycleState.STOPPED
    print("session-cleanup-ok")


def cmd_json_parsing() -> None:
    """Verify devcontainer up JSON output parsing."""
    stdout = json.dumps(
        {
            "outcome": "success",
            "containerId": "aabbccddee0044556677",
            "remoteWorkspaceFolder": "/parsed/ws",
        }
    )
    ctr_id, ws = _parse_devcontainer_up_output(stdout)
    assert ctr_id == "aabbccddee0044556677"
    assert ws == "/parsed/ws"

    # Invalid JSON
    ctr_id2, ws2 = _parse_devcontainer_up_output("not json {{")
    assert ctr_id2 is None
    assert ws2 is None
    print("json-parsing-ok")


def cmd_registry_persist() -> None:
    """Verify tracker persists in registry."""
    clear_lifecycle_registry()
    tracker = ContainerLifecycleTracker(resource_id="01ROBOTTEST0000000000060")
    set_lifecycle_tracker(tracker)
    tracker2 = transition_state(tracker, ContainerLifecycleState.BUILDING, reason="t")
    set_lifecycle_tracker(tracker2)
    retrieved = get_lifecycle_tracker("01ROBOTTEST0000000000060")
    assert retrieved.current_state == ContainerLifecycleState.BUILDING
    print("registry-persist-ok")


def cmd_cleanup_method() -> None:
    """Verify CleanupService.stop_active_devcontainers exists."""
    assert hasattr(CleanupService, "stop_active_devcontainers")
    assert callable(CleanupService.stop_active_devcontainers)
    print("cleanup-method-ok")


_COMMANDS = {
    "enum-values": cmd_enum_values,
    "transition-valid": cmd_transition_valid,
    "transition-invalid": cmd_transition_invalid,
    "lazy-activation": cmd_lazy_activation,
    "stop-container": cmd_stop_container,
    "rebuild-container": cmd_rebuild_container,
    "session-cleanup": cmd_session_cleanup,
    "json-parsing": cmd_json_parsing,
    "registry-persist": cmd_registry_persist,
    "cleanup-method": cmd_cleanup_method,
}


def main() -> None:
    """Dispatch subcommand from argv."""
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <command>", file=sys.stderr)
        sys.exit(1)
    cmd = sys.argv[1]
    func = _COMMANDS.get(cmd)
    if func is None:
        print(f"Unknown command: {cmd}", file=sys.stderr)
        sys.exit(1)
    func()


if __name__ == "__main__":
    main()