cleveragents-core/robot/helper_checkpoint_real_rollback.py

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

Each subcommand exercises CheckpointService.rollback_to_checkpoint() against
a real git repository to verify that the method executes ``git reset --hard``
and actually restores file system state.
"""

from __future__ import annotations

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

# Ensure local source tree AND robot/ directory are importable.
_ROOT = Path(__file__).resolve().parents[1]
_SRC = str(_ROOT / "src")
_ROBOT = str(_ROOT / "robot")
for _p in (_SRC, _ROBOT):
    if _p not in sys.path:
        sys.path.insert(0, _p)

from cleveragents.application.services.checkpoint_service import (  # noqa: E402
    CheckpointService,
)
from cleveragents.infrastructure.events.models import DomainEvent  # noqa: E402
from cleveragents.infrastructure.events.types import EventType  # noqa: E402

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

_INITIAL_CONTENT = "initial content\n"
_MODIFIED_CONTENT = "modified after checkpoint\n"
_TRACKED_FILENAME = "tracked.txt"
_NEW_FILENAME = "extra.txt"
_PLAN_ID = "01JBG822CHKPT000PXAN000000"


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _fail(msg: str) -> NoReturn:
    """Print failure message to stderr and exit with code 1."""
    print(msg, file=sys.stderr)
    sys.exit(1)


def _run_git(args: list[str], cwd: str) -> subprocess.CompletedProcess[str]:
    """Run a git command and return the completed process."""
    return subprocess.run(
        ["git", *args],
        cwd=cwd,
        capture_output=True,
        text=True,
        check=True,
        timeout=30,
    )


def _get_head_sha(cwd: str) -> str:
    """Return the HEAD commit SHA of a git repository."""
    result = _run_git(["rev-parse", "HEAD"], cwd=cwd)
    return result.stdout.strip()


def _create_workspace() -> str:
    """Create a temporary git workspace with an initial committed file.

    Returns the path to the temporary directory.
    """
    tmpdir = tempfile.mkdtemp(prefix="checkpoint_822_robot_")
    try:
        _run_git(["init", "--initial-branch=main"], cwd=tmpdir)
        _run_git(["config", "user.email", "test@example.com"], cwd=tmpdir)
        _run_git(["config", "user.name", "Test"], cwd=tmpdir)

        tracked_path = os.path.join(tmpdir, _TRACKED_FILENAME)
        Path(tracked_path).write_text(_INITIAL_CONTENT, encoding="utf-8")
        _run_git(["add", _TRACKED_FILENAME], cwd=tmpdir)
        _run_git(["commit", "-m", "Initial commit"], cwd=tmpdir)
    except Exception:
        shutil.rmtree(tmpdir, ignore_errors=True)
        raise
    return tmpdir


class _RecordingEventBus:
    """In-memory event bus that records emitted events."""

    def __init__(self) -> None:
        self.events: list[DomainEvent] = []

    def emit(self, event: DomainEvent) -> None:
        self.events.append(event)

    def subscribe(self, event_type: object, handler: object) -> None:
        pass


# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------


def rollback_restores_content() -> None:
    """Verify that rollback reverts a modified tracked file."""
    tmpdir = _create_workspace()
    try:
        tracked_path = os.path.join(tmpdir, _TRACKED_FILENAME)
        initial_sha = _get_head_sha(tmpdir)

        service = CheckpointService()
        service.register_sandbox(_PLAN_ID, tmpdir)
        checkpoint = service.create_checkpoint(
            plan_id=_PLAN_ID,
            sandbox_ref=initial_sha,
            reason="Integration test checkpoint",
            checkpoint_type="manual",
        )

        # Modify the tracked file and commit
        Path(tracked_path).write_text(_MODIFIED_CONTENT)
        _run_git(["add", _TRACKED_FILENAME], cwd=tmpdir)
        _run_git(["commit", "-m", "Modify tracked file"], cwd=tmpdir)

        # Invoke rollback
        service.rollback_to_checkpoint(
            plan_id=_PLAN_ID,
            checkpoint_id=checkpoint.checkpoint_id,
        )

        # Assert file content matches checkpoint state
        actual = Path(tracked_path).read_text()
        if actual != _INITIAL_CONTENT:
            _fail(
                f"Tracked file not reverted after rollback. "
                f"Expected: {_INITIAL_CONTENT!r}, Got: {actual!r}."
            )

        print("checkpoint-rollback-restores-content-ok")
    finally:
        shutil.rmtree(tmpdir, ignore_errors=True)


def rollback_removes_added_files() -> None:
    """Verify that rollback removes files added after the checkpoint."""
    tmpdir = _create_workspace()
    try:
        initial_sha = _get_head_sha(tmpdir)

        service = CheckpointService()
        service.register_sandbox(_PLAN_ID, tmpdir)
        checkpoint = service.create_checkpoint(
            plan_id=_PLAN_ID,
            sandbox_ref=initial_sha,
            reason="Integration test checkpoint",
            checkpoint_type="manual",
        )

        # Add a new file and commit
        new_path = os.path.join(tmpdir, _NEW_FILENAME)
        Path(new_path).write_text("new file content\n")
        _run_git(["add", _NEW_FILENAME], cwd=tmpdir)
        _run_git(["commit", "-m", "Add new file"], cwd=tmpdir)

        # Invoke rollback
        service.rollback_to_checkpoint(
            plan_id=_PLAN_ID,
            checkpoint_id=checkpoint.checkpoint_id,
        )

        # Assert the new file no longer exists
        if os.path.exists(new_path):
            _fail(f"New file still exists after rollback: {new_path}.")

        print("checkpoint-rollback-removes-added-files-ok")
    finally:
        shutil.rmtree(tmpdir, ignore_errors=True)


def rollback_emits_event() -> None:
    """Verify that rollback emits a CHECKPOINT_RESTORED domain event."""
    tmpdir = _create_workspace()
    try:
        initial_sha = _get_head_sha(tmpdir)
        event_bus = _RecordingEventBus()

        service = CheckpointService(event_bus=event_bus)  # type: ignore[arg-type]
        service.register_sandbox(_PLAN_ID, tmpdir)
        checkpoint = service.create_checkpoint(
            plan_id=_PLAN_ID,
            sandbox_ref=initial_sha,
            reason="Integration test checkpoint",
            checkpoint_type="manual",
        )

        # Modify and commit so rollback has something to do
        tracked_path = os.path.join(tmpdir, _TRACKED_FILENAME)
        Path(tracked_path).write_text(_MODIFIED_CONTENT)
        _run_git(["add", _TRACKED_FILENAME], cwd=tmpdir)
        _run_git(["commit", "-m", "Modify tracked file"], cwd=tmpdir)

        service.rollback_to_checkpoint(
            plan_id=_PLAN_ID,
            checkpoint_id=checkpoint.checkpoint_id,
        )

        restored_events = [
            e for e in event_bus.events if e.event_type == EventType.CHECKPOINT_RESTORED
        ]
        if len(restored_events) != 1:
            _fail(f"Expected 1 CHECKPOINT_RESTORED event, got {len(restored_events)}.")

        print("checkpoint-rollback-emits-event-ok")
    finally:
        shutil.rmtree(tmpdir, ignore_errors=True)


def rollback_rejects_nonexistent_sandbox() -> None:
    """Verify that rollback rejects a nonexistent sandbox path."""
    from cleveragents.core.exceptions import BusinessRuleViolation

    tmpdir = _create_workspace()
    try:
        initial_sha = _get_head_sha(tmpdir)

        service = CheckpointService()
        service.register_sandbox(_PLAN_ID, "/tmp/nonexistent_sandbox_path_822")
        checkpoint = service.create_checkpoint(
            plan_id=_PLAN_ID,
            sandbox_ref=initial_sha,
            reason="Integration test checkpoint",
            checkpoint_type="manual",
        )

        try:
            service.rollback_to_checkpoint(
                plan_id=_PLAN_ID,
                checkpoint_id=checkpoint.checkpoint_id,
            )
            _fail("Expected BusinessRuleViolation for nonexistent sandbox path")
        except BusinessRuleViolation as exc:
            if "sandbox path does not exist" not in str(exc):
                _fail(f"Unexpected error message: {exc}")

        print("checkpoint-rollback-rejects-nonexistent-sandbox-ok")
    finally:
        shutil.rmtree(tmpdir, ignore_errors=True)


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

_COMMANDS: dict[str, Callable[[], None]] = {
    "rollback-restores-content": rollback_restores_content,
    "rollback-removes-added-files": rollback_removes_added_files,
    "rollback-emits-event": rollback_emits_event,
    "rollback-rejects-nonexistent-sandbox": rollback_rejects_nonexistent_sandbox,
}

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)
    cmd = _COMMANDS[sys.argv[1]]
    cmd()