223 lines
7.3 KiB
Python
223 lines
7.3 KiB
Python
"""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()
|