cleveragents-core/robot/helper_tdd_checkpoint_real_rollback.py

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

Each subcommand exercises CheckpointService.rollback_to_checkpoint() against
a real git repository to prove that the method does **not** execute
``git reset --hard``.  Files modified after a checkpoint remain unchanged
after rollback, demonstrating bug #822.

The helper exits 0 with a sentinel when the rollback correctly restores
file state (bug fixed), and exits 1 when the bug is still present.
The ``tdd_expected_fail_listener`` on the Robot side handles pass/fail
inversion while the bug remains open.
"""

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

# ---------------------------------------------------------------------------
# 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="tdd_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


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


def rollback_restores_content() -> None:
    """Verify that rollback reverts a modified tracked file.

    Exits 0 with sentinel when the file is correctly reverted (bug fixed).
    Exits 1 when the bug is still present (file remains modified).
    """
    tmpdir = _create_workspace()
    try:
        tracked_path = os.path.join(tmpdir, _TRACKED_FILENAME)
        initial_sha = _get_head_sha(tmpdir)

        # Create checkpoint from current state
        service = CheckpointService()
        service.register_sandbox(_PLAN_ID, tmpdir)
        checkpoint = service.create_checkpoint(
            plan_id=_PLAN_ID,
            sandbox_ref=initial_sha,
            reason="TDD checkpoint for bug #822",
            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}. "
                f"Bug #822: rollback did not execute git reset --hard."
            )

        print("tdd-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.

    Exits 0 with sentinel when the new file is removed (bug fixed).
    Exits 1 when the bug is still present (new file still exists).
    """
    tmpdir = _create_workspace()
    try:
        initial_sha = _get_head_sha(tmpdir)

        # Create checkpoint from current state
        service = CheckpointService()
        service.register_sandbox(_PLAN_ID, tmpdir)
        checkpoint = service.create_checkpoint(
            plan_id=_PLAN_ID,
            sandbox_ref=initial_sha,
            reason="TDD checkpoint for bug #822",
            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}. "
                f"Bug #822: rollback did not execute git reset --hard."
            )

        print("tdd-checkpoint-rollback-removes-added-files-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,
}

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