cleveragents-core/robot/helper_devcontainer_handler.py

"""Helper utilities for devcontainer handler 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
import tempfile
from pathlib import Path
from typing import Any

from cleveragents.domain.models.core.resource_type import ResourceTypeSpec
from cleveragents.resource.handlers.devcontainer import DevcontainerHandler
from cleveragents.resource.handlers.discovery import (
    DevcontainerDiscoveryResult,
    discover_devcontainers,
)
from cleveragents.resource.handlers.protocol import ResourceHandler
from cleveragents.resource.handlers.resolver import (
    clear_handler_cache,
    resolve_handler,
)

_VALID_DC_JSON: str = json.dumps(
    {"name": "test-devcontainer", "image": "ubuntu:latest"}
)

# -- Built-in type definitions for DAG checks --------------------------------

_BUILTIN_TYPES: list[dict[str, Any]] = [
    {
        "name": "git-checkout",
        "description": "Git checkout",
        "resource_kind": "physical",
        "sandbox_strategy": "git_worktree",
        "built_in": True,
        "child_types": [
            "fs-directory",
            "fs-file",
            "devcontainer-instance",
            "devcontainer-file",
        ],
    },
    {
        "name": "devcontainer-instance",
        "description": "Devcontainer instance",
        "resource_kind": "physical",
        "sandbox_strategy": "snapshot",
        "built_in": True,
        "parent_types": ["git-checkout", "fs-directory"],
        "child_types": ["devcontainer-file"],
    },
    {
        "name": "devcontainer-file",
        "description": "Devcontainer config file",
        "resource_kind": "physical",
        "sandbox_strategy": "copy_on_write",
        "built_in": True,
        "parent_types": ["devcontainer-instance"],
        "child_types": [],
    },
]


def cmd_protocol_check() -> None:
    """Verify DevcontainerHandler satisfies ResourceHandler."""
    handler = DevcontainerHandler()
    assert isinstance(handler, ResourceHandler), "Protocol check failed"
    print("protocol-check-ok")


def cmd_strategy_check() -> None:
    """Verify DevcontainerHandler default strategy.

    F22/F25 fix: handler uses ``none`` because SandboxFactory has not yet
    implemented ``snapshot``.  The container itself provides isolation.
    Known limitation — will switch to ``snapshot`` once implemented.
    """
    handler = DevcontainerHandler()
    assert handler._default_strategy.value == "none"
    print("strategy-check-ok")


def cmd_discovery_valid() -> None:
    """Discover devcontainer from valid config."""
    with tempfile.TemporaryDirectory() as tmp:
        dc_dir = Path(tmp) / ".devcontainer"
        dc_dir.mkdir()
        (dc_dir / "devcontainer.json").write_text(_VALID_DC_JSON, encoding="utf-8")
        results = discover_devcontainers(tmp, "git-checkout")
        assert len(results) == 1, f"Expected 1 result, got {len(results)}"
        assert str(results[0].config_path).endswith("devcontainer.json")
    print("discovery-valid-ok")


def cmd_discovery_invalid() -> None:
    """Skip invalid devcontainer.json."""
    with tempfile.TemporaryDirectory() as tmp:
        dc_dir = Path(tmp) / ".devcontainer"
        dc_dir.mkdir()
        (dc_dir / "devcontainer.json").write_text("not valid json {{", encoding="utf-8")
        results = discover_devcontainers(tmp, "git-checkout")
        assert len(results) == 0, f"Expected 0 results, got {len(results)}"
    print("discovery-invalid-ok")


def cmd_discovery_root() -> None:
    """Discover root-level .devcontainer.json."""
    with tempfile.TemporaryDirectory() as tmp:
        (Path(tmp) / ".devcontainer.json").write_text(_VALID_DC_JSON, encoding="utf-8")
        results = discover_devcontainers(tmp, "fs-directory")
        assert len(results) == 1, f"Expected 1 result, got {len(results)}"
    print("discovery-root-ok")


def cmd_discovery_nontrigger() -> None:
    """No discovery for non-trigger types."""
    with tempfile.TemporaryDirectory() as tmp:
        dc_dir = Path(tmp) / ".devcontainer"
        dc_dir.mkdir()
        (dc_dir / "devcontainer.json").write_text(_VALID_DC_JSON, encoding="utf-8")
        results = discover_devcontainers(tmp, "fs-file")
        assert len(results) == 0, f"Expected 0 results, got {len(results)}"
    print("discovery-nontrigger-ok")


def cmd_builtin_types() -> None:
    """Verify devcontainer types in BUILTIN_NAMES."""
    names = ResourceTypeSpec.BUILTIN_NAMES
    assert "devcontainer-instance" in names
    assert "devcontainer-file" in names
    assert "container-instance" in names
    print("builtin-types-ok")


def cmd_dag_hierarchy() -> None:
    """Validate devcontainer DAG parent/child constraints."""
    for type_def in _BUILTIN_TYPES:
        name = type_def["name"]
        if name == "devcontainer-instance":
            assert "git-checkout" in type_def["parent_types"]
            assert "fs-directory" in type_def["parent_types"]
            assert "devcontainer-file" in type_def["child_types"]
        elif name == "devcontainer-file":
            assert "devcontainer-instance" in type_def["parent_types"]
            assert len(type_def["child_types"]) == 0
        elif name == "git-checkout":
            assert "devcontainer-instance" in type_def["child_types"]
    print("dag-hierarchy-ok")


def cmd_resolver_import() -> None:
    """Verify resolve_handler loads DevcontainerHandler."""
    clear_handler_cache()
    handler = resolve_handler(
        "cleveragents.resource.handlers.devcontainer:DevcontainerHandler"
    )
    assert isinstance(handler, ResourceHandler)
    assert isinstance(handler, DevcontainerHandler)
    print("resolver-import-ok")


def cmd_result_validation() -> None:
    """Verify DevcontainerDiscoveryResult validates inputs."""
    with tempfile.TemporaryDirectory() as tmp:
        dc_file = Path(tmp) / "devcontainer.json"
        dc_file.write_text(_VALID_DC_JSON, encoding="utf-8")

        # Valid creation
        result = DevcontainerDiscoveryResult(
            config_path=dc_file,
            config_data=json.loads(_VALID_DC_JSON),
            parent_location=tmp,
        )
        assert result.config_path == dc_file

        # Empty parent_location should raise
        try:
            DevcontainerDiscoveryResult(
                config_path=dc_file,
                config_data=json.loads(_VALID_DC_JSON),
                parent_location="",
            )
            raise AssertionError("Should have raised ValueError")
        except ValueError:
            pass

    print("result-validation-ok")


_COMMANDS = {
    "protocol-check": cmd_protocol_check,
    "strategy-check": cmd_strategy_check,
    "discovery-valid": cmd_discovery_valid,
    "discovery-invalid": cmd_discovery_invalid,
    "discovery-root": cmd_discovery_root,
    "discovery-nontrigger": cmd_discovery_nontrigger,
    "builtin-types": cmd_builtin_types,
    "dag-hierarchy": cmd_dag_hierarchy,
    "resolver-import": cmd_resolver_import,
    "result-validation": cmd_result_validation,
}


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()