cleveragents-core/robot/helper_custom_sandbox_strategy.py

"""Helper utilities for custom sandbox strategy Robot integration tests.

Covers end-to-end integration of:
- SandboxStrategyProtocol and runtime checking
- SandboxStrategyRegistry registration and lookup
- BuiltInSandboxStrategyAdapter full lifecycle
- CustomStrategyConfig validation
- SandboxFactory custom strategy integration

Based on issue #586.
"""

from __future__ import annotations

import os
import sys
import tempfile
from datetime import datetime

from cleveragents.domain.models.core.resource import (
    PhysVirt,
    Resource,
    ResourceCapabilities,
)
from cleveragents.domain.models.core.resource import (
    SandboxStrategy as SandboxStrategyEnum,
)
from cleveragents.domain.models.core.sandbox_strategy import (
    DiffEntry,
    DiffView,
    SandboxRef,
    SandboxStrategyProtocol,
)
from cleveragents.infrastructure.plugins.exceptions import (
    ProtocolMismatchError,
)
from cleveragents.infrastructure.sandbox.factory import SandboxFactory
from cleveragents.infrastructure.sandbox.strategy_adapter import (
    BuiltInSandboxStrategyAdapter,
)
from cleveragents.infrastructure.sandbox.strategy_registry import (
    CustomStrategyConfig,
    SandboxStrategyRegistry,
)

# ---------------------------------------------------------------------------
# ULID helper
# ---------------------------------------------------------------------------

_CTR = 0


def _ulid(name: str) -> str:
    global _CTR
    _CTR += 1
    base = abs(hash(name)) % (10**20)
    raw = f"{_CTR:06d}{base:020d}"
    charset = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
    result = ""
    for ch in raw:
        result += charset[int(ch)]
    return (result + "0" * 26)[:26]


def _make_resource(name: str, location: str) -> Resource:
    return Resource(
        resource_id=_ulid(name),
        name=name,
        resource_type_name="fs-directory",
        classification=PhysVirt.PHYSICAL,
        sandbox_strategy=SandboxStrategyEnum.NONE,
        location=location,
        capabilities=ResourceCapabilities(
            readable=True,
            writable=True,
            sandboxable=True,
            checkpointable=False,
        ),
    )


# ---------------------------------------------------------------------------
# Test: SandboxRef creation and immutability
# ---------------------------------------------------------------------------


def _sandbox_ref_lifecycle() -> None:
    """Integration test: SandboxRef creation and properties."""
    ref = SandboxRef(
        sandbox_id="ref-001",
        plan_id="plan-001",
        resource_id="res-001",
        created_at=datetime.now(),
        metadata={"backend": "test"},
    )
    assert ref.sandbox_id == "ref-001"
    assert ref.plan_id == "plan-001"
    assert ref.metadata["backend"] == "test"

    # Frozen check
    try:
        ref.sandbox_id = "mutated"  # type: ignore[misc]
        print("FAIL: Expected FrozenInstanceError")
        return
    except (AttributeError, Exception):
        pass

    print("sandbox-ref-lifecycle-ok")


# ---------------------------------------------------------------------------
# Test: DiffView and DiffEntry
# ---------------------------------------------------------------------------


def _diff_view_lifecycle() -> None:
    """Integration test: DiffView and DiffEntry models."""
    entry = DiffEntry(path="test.py", operation="added", after=b"content")
    assert entry.path == "test.py"
    assert entry.operation == "added"

    view = DiffView(
        sandbox_id="dv-001",
        entries=[entry],
        summary="1 file added",
    )
    assert len(view.entries) == 1
    assert view.sandbox_id == "dv-001"

    # Empty path should fail
    try:
        DiffEntry(path="", operation="modified")
        print("FAIL: Expected ValidationError")
        return
    except Exception:
        pass

    print("diff-view-lifecycle-ok")


# ---------------------------------------------------------------------------
# Test: SandboxStrategyProtocol runtime check
# ---------------------------------------------------------------------------


def _protocol_check() -> None:
    """Integration test: Protocol runtime checking."""

    class _Good:
        def create(self, plan_id, resource):
            return None

        def read(self, ref, path):
            return b""

        def write(self, ref, path, content):
            return None

        def diff(self, ref):
            return None

        def commit(self, ref):
            pass

        def rollback(self, ref):
            pass

        def checkpoint(self, ref, checkpoint_id):
            pass

        def restore_checkpoint(self, ref, checkpoint_id):
            pass

        def cleanup(self, ref):
            pass

    class _Bad:
        def create(self, plan_id, resource):
            return None

    good = _Good()
    bad = _Bad()
    assert isinstance(good, SandboxStrategyProtocol)
    assert not isinstance(bad, SandboxStrategyProtocol)

    print("protocol-check-ok")


# ---------------------------------------------------------------------------
# Test: SandboxStrategyRegistry
# ---------------------------------------------------------------------------


def _registry_lifecycle() -> None:
    """Integration test: Registry registration, lookup, config, clear."""
    registry = SandboxStrategyRegistry(allowed_prefixes=("cleveragents.",))

    # Register
    cls = registry.register(
        "adapter",
        "cleveragents.infrastructure.sandbox.strategy_adapter",
        "BuiltInSandboxStrategyAdapter",
    )
    assert cls is BuiltInSandboxStrategyAdapter
    assert registry.has("adapter")
    assert "adapter" in registry.list_strategies()

    # Get
    assert registry.get("adapter") is BuiltInSandboxStrategyAdapter
    assert registry.get("nonexistent") is None

    # Non-protocol rejection
    try:
        registry.register(
            "bad",
            "cleveragents.infrastructure.sandbox.protocol",
            "SandboxError",
        )
        print("FAIL: Expected ProtocolMismatchError")
        return
    except ProtocolMismatchError:
        pass

    # Config-driven registration
    configs = {
        "cfg1": {
            "module": "cleveragents.infrastructure.sandbox.strategy_adapter",
            "class": "BuiltInSandboxStrategyAdapter",
        },
    }
    registered = registry.register_all_from_config(configs)
    assert len(registered) == 1

    # Clear
    registry.clear()
    assert len(registry.list_strategies()) == 0

    print("registry-lifecycle-ok")


# ---------------------------------------------------------------------------
# Test: BuiltInSandboxStrategyAdapter full lifecycle
# ---------------------------------------------------------------------------


def _adapter_lifecycle() -> None:
    """Integration test: Adapter create/read/write/diff/commit/cleanup."""
    with tempfile.TemporaryDirectory() as tmpdir:
        # Seed a file
        with open(os.path.join(tmpdir, "seed.txt"), "w") as f:
            f.write("seed")

        adapter = BuiltInSandboxStrategyAdapter(strategy_name="copy_on_write")
        resource = Resource(
            resource_id=_ulid("adapter-res"),
            name="adapter-res",
            resource_type_name="fs-directory",
            classification=PhysVirt.PHYSICAL,
            sandbox_strategy=SandboxStrategyEnum.COPY_ON_WRITE,
            location=tmpdir,
            capabilities=ResourceCapabilities(
                readable=True,
                writable=True,
                sandboxable=True,
                checkpointable=False,
            ),
        )

        # Create
        ref = adapter.create("plan-adapter", resource)
        assert isinstance(ref, SandboxRef)
        assert ref.plan_id == "plan-adapter"

        # Write
        entry = adapter.write(ref, "new_file.txt", b"hello")
        assert isinstance(entry, DiffEntry)
        assert entry.operation == "added"

        # Read
        data = adapter.read(ref, "new_file.txt")
        assert data == b"hello"

        # Diff
        view = adapter.diff(ref)
        assert isinstance(view, DiffView)

        # Cleanup
        adapter.cleanup(ref)
        assert ref.sandbox_id not in adapter._sandboxes

    print("adapter-lifecycle-ok")


# ---------------------------------------------------------------------------
# Test: Adapter checkpoint/restore
# ---------------------------------------------------------------------------


def _adapter_checkpoint() -> None:
    """Integration test: Adapter checkpoint and restore."""
    with tempfile.TemporaryDirectory() as tmpdir:
        with open(os.path.join(tmpdir, "seed.txt"), "w") as f:
            f.write("seed")

        adapter = BuiltInSandboxStrategyAdapter(strategy_name="copy_on_write")
        resource = Resource(
            resource_id=_ulid("cp-res"),
            name="cp-res",
            resource_type_name="fs-directory",
            classification=PhysVirt.PHYSICAL,
            sandbox_strategy=SandboxStrategyEnum.COPY_ON_WRITE,
            location=tmpdir,
            capabilities=ResourceCapabilities(
                readable=True,
                writable=True,
                sandboxable=True,
                checkpointable=False,
            ),
        )

        ref = adapter.create("plan-cp", resource)
        adapter.write(ref, "cp.txt", b"v1")
        adapter.checkpoint(ref, "cp-1")
        adapter.write(ref, "cp.txt", b"v2")

        # Verify v2
        assert adapter.read(ref, "cp.txt") == b"v2"

        # Restore
        adapter.restore_checkpoint(ref, "cp-1")
        assert adapter.read(ref, "cp.txt") == b"v1"

        adapter.cleanup(ref)

    print("adapter-checkpoint-ok")


# ---------------------------------------------------------------------------
# Test: CustomStrategyConfig validation
# ---------------------------------------------------------------------------


def _config_validation() -> None:
    """Integration test: CustomStrategyConfig validation."""
    # Valid
    cfg = CustomStrategyConfig(
        name="test", module="cleveragents.test", class_name="TestCls"
    )
    assert cfg.name == "test"

    # Empty name
    try:
        CustomStrategyConfig(name="", module="mod", class_name="Cls")
        print("FAIL: Expected ValueError for empty name")
        return
    except ValueError:
        pass

    # Empty module
    try:
        CustomStrategyConfig(name="x", module="", class_name="Cls")
        print("FAIL: Expected ValueError for empty module")
        return
    except ValueError:
        pass

    # Empty class
    try:
        CustomStrategyConfig(name="x", module="mod", class_name="")
        print("FAIL: Expected ValueError for empty class")
        return
    except ValueError:
        pass

    print("config-validation-ok")


# ---------------------------------------------------------------------------
# Test: SandboxFactory custom strategy integration
# ---------------------------------------------------------------------------


def _factory_custom_integration() -> None:
    """Integration test: SandboxFactory with custom strategy registry."""
    registry = SandboxStrategyRegistry(allowed_prefixes=("cleveragents.",))
    registry.register(
        "my_custom",
        "cleveragents.infrastructure.sandbox.strategy_adapter",
        "BuiltInSandboxStrategyAdapter",
    )

    factory = SandboxFactory(custom_registry=registry)
    assert factory.has_custom_strategy("my_custom")
    assert not factory.has_custom_strategy("nonexistent")

    cls = factory.get_custom_strategy_class("my_custom")
    assert cls is BuiltInSandboxStrategyAdapter

    # Factory without registry
    plain_factory = SandboxFactory()
    assert not plain_factory.has_custom_strategy("anything")
    assert plain_factory.get_custom_strategy_class("anything") is None

    print("factory-custom-integration-ok")


# ---------------------------------------------------------------------------
# Main dispatcher
# ---------------------------------------------------------------------------

_TESTS = {
    "sandbox-ref-lifecycle": _sandbox_ref_lifecycle,
    "diff-view-lifecycle": _diff_view_lifecycle,
    "protocol-check": _protocol_check,
    "registry-lifecycle": _registry_lifecycle,
    "adapter-lifecycle": _adapter_lifecycle,
    "adapter-checkpoint": _adapter_checkpoint,
    "config-validation": _config_validation,
    "factory-custom-integration": _factory_custom_integration,
}


def main() -> None:
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <test-name>")
        print(f"Available: {', '.join(sorted(_TESTS))}")
        sys.exit(1)

    test_name = sys.argv[1]
    if test_name not in _TESTS:
        print(f"Unknown test: {test_name}")
        print(f"Available: {', '.join(sorted(_TESTS))}")
        sys.exit(1)

    _TESTS[test_name]()


if __name__ == "__main__":
    main()