cleveragents-core/robot/helper_resource_handler_crud.py

"""Helper script for resource_handler_crud.robot integration tests.

Exercises the read -> write -> diff cycle on a real temporary git
repository using GitCheckoutHandler, and on a real temporary directory
using FsDirectoryHandler.

Issue #827: ResourceHandler CRUD and discovery methods.
"""

from __future__ import annotations

import subprocess
import sys
import tempfile
from collections.abc import Callable
from pathlib import Path
from typing import NoReturn

_ROOT = Path(__file__).resolve().parents[1]
_SRC = str(_ROOT / "src")
# Ensure local source takes precedence over installed package.
# Remove any existing cleveragents modules so importlib re-resolves
# from sys.path[0] (our local src).
sys.path.insert(0, _SRC)
for _mod_name in list(sys.modules):
    if _mod_name.startswith("cleveragents"):
        del sys.modules[_mod_name]

from cleveragents.domain.models.core.resource import (  # noqa: E402
    PhysVirt,
    Resource,
    ResourceCapabilities,
)
from cleveragents.resource.handlers.fs_directory import FsDirectoryHandler  # noqa: E402
from cleveragents.resource.handlers.git_checkout import GitCheckoutHandler  # noqa: E402


def _fail(msg: str) -> NoReturn:
    print(msg, file=sys.stderr)
    sys.exit(1)


def _make_resource(rtype: str, location: str) -> Resource:
    return Resource(
        resource_id="01KJ5C5TPMP8GGX3QC83E2MAQS",
        resource_type_name=rtype,
        classification=PhysVirt.PHYSICAL,
        location=location,
        capabilities=ResourceCapabilities(
            readable=True, writable=True, sandboxable=True
        ),
    )


# ---------------------------------------------------------------------------
# fs-read-write-diff: FsDirectoryHandler read -> write -> diff cycle
# ---------------------------------------------------------------------------


def fs_read_write_diff() -> None:
    """Test read -> write -> diff cycle on FsDirectoryHandler."""
    handler = FsDirectoryHandler()

    # Create two temp dirs: source and comparison
    src = tempfile.mkdtemp(prefix="crud_robot_src_")
    cmp_dir = tempfile.mkdtemp(prefix="crud_robot_cmp_")

    # Write identical files to both
    Path(src, "file.txt").write_text("original\n", encoding="utf-8")
    Path(cmp_dir, "file.txt").write_text("original\n", encoding="utf-8")

    resource = _make_resource("fs-directory", src)

    # 1. Read
    content = handler.read(resource=resource, path="file.txt")
    if content.text.strip() != "original":
        _fail(f"read failed: expected 'original', got '{content.text.strip()}'")

    # 2. Write (modify)
    result = handler.write(resource=resource, path="file.txt", data=b"modified\n")
    if not result.success:
        _fail(f"write failed: {result.message}")

    # 3. Diff against comparison dir (should detect change)
    diff = handler.diff(resource=resource, other_location=cmp_dir)
    if not diff.has_changes:
        _fail("diff should detect changes after write")
    if diff.files_changed < 1:
        _fail(f"diff files_changed={diff.files_changed}, expected >= 1")

    # 4. List children
    children = handler.list_children(resource=resource)
    if "file.txt" not in children:
        _fail(f"list_children missing 'file.txt': {children}")

    print("fs-read-write-diff-ok")


# ---------------------------------------------------------------------------
# git-read-write-diff: GitCheckoutHandler read -> write -> diff cycle
# ---------------------------------------------------------------------------


def git_read_write_diff() -> None:
    """Test read -> write -> diff cycle on GitCheckoutHandler."""
    handler = GitCheckoutHandler()

    # Create a temp git repo
    repo = tempfile.mkdtemp(prefix="crud_robot_git_")
    subprocess.run(["git", "init"], cwd=repo, capture_output=True, check=True)
    subprocess.run(
        ["git", "config", "user.email", "test@test.com"],
        cwd=repo,
        capture_output=True,
        check=True,
    )
    subprocess.run(
        ["git", "config", "user.name", "Test"],
        cwd=repo,
        capture_output=True,
        check=True,
    )
    Path(repo, "code.py").write_text("print('hello')\n", encoding="utf-8")
    subprocess.run(["git", "add", "."], cwd=repo, capture_output=True, check=True)
    subprocess.run(
        ["git", "commit", "-m", "init"],
        cwd=repo,
        capture_output=True,
        check=True,
    )

    resource = _make_resource("git-checkout", repo)

    # Comparison dir
    cmp_dir = tempfile.mkdtemp(prefix="crud_robot_cmp_")
    Path(cmp_dir, "code.py").write_text("print('hello')\n", encoding="utf-8")

    # 1. Read
    content = handler.read(resource=resource, path="code.py")
    if "hello" not in content.text:
        _fail(f"read failed: 'hello' not in '{content.text}'")

    # 2. Write
    result = handler.write(resource=resource, path="code.py", data=b"print('world')\n")
    if not result.success:
        _fail(f"write failed: {result.message}")

    # 3. Diff (should detect change since we modified the file)
    diff = handler.diff(resource=resource, other_location=cmp_dir)
    if not diff.has_changes:
        _fail("diff should detect changes after write")

    # 4. List children
    children = handler.list_children(resource=resource)
    if "code.py" not in children:
        _fail(f"list_children missing 'code.py': {children}")

    # 5. Discover children (no subdirs expected)
    handler.discover_children(resource=resource)
    # May or may not have children — just ensure no crash

    print("git-read-write-diff-ok")


# ---------------------------------------------------------------------------
# Dispatcher
# ---------------------------------------------------------------------------

_COMMANDS: dict[str, Callable[[], None]] = {
    "fs-read-write-diff": fs_read_write_diff,
    "git-read-write-diff": git_read_write_diff,
}

if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(
            f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>",
            file=sys.stderr,
        )
        sys.exit(1)
    _COMMANDS[sys.argv[1]]()