cleveragents-core/robot/helper_repo_indexing.py

"""Robot Framework helper for repo indexing integration tests.

Verifies that the RepoIndexingService can index a directory, refresh
incrementally, enforce policy limits, and clean up on removal.

Usage:
    python robot/helper_repo_indexing.py full-index
    python robot/helper_repo_indexing.py incremental-refresh
    python robot/helper_repo_indexing.py policy-enforcement
"""

from __future__ import annotations

import shutil
import sys
import tempfile
from collections.abc import Callable
from pathlib import Path

_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from sqlalchemy import create_engine  # noqa: E402
from sqlalchemy.orm import sessionmaker  # noqa: E402

from cleveragents.application.services.repo_indexing_service import (  # noqa: E402
    RepoIndexingService,
)
from cleveragents.infrastructure.database.models import Base  # noqa: E402

_RID = "01KK0D8WNATFNEX2JMG5GKF6FQ"


def _make_service() -> RepoIndexingService:
    engine = create_engine("sqlite:///:memory:", echo=False)
    Base.metadata.create_all(engine)
    factory = sessionmaker(bind=engine, expire_on_commit=False)
    return RepoIndexingService(session_factory=factory)


def _make_sample_dir() -> str:
    tmpdir = tempfile.mkdtemp()
    (Path(tmpdir) / "app.py").write_text("print('hello')\n")
    (Path(tmpdir) / "lib.py").write_text("def f(): return 1\n")
    (Path(tmpdir) / "README.md").write_text("# App\n")
    return tmpdir


def _cmd_full_index() -> int:
    svc = _make_service()
    tmpdir = _make_sample_dir()
    try:
        idx = svc.index_resource(_RID, tmpdir)
        if idx.metadata.status.value != "ready":
            print(f"FAIL: status={idx.metadata.status.value}")
            return 1
        if idx.metadata.file_count != 3:
            print(f"FAIL: file_count={idx.metadata.file_count}")
            return 1
        if idx.metadata.primary_language != "python":
            print(f"FAIL: language={idx.metadata.primary_language}")
            return 1

        # Verify persistence
        retrieved = svc.get_index(_RID)
        if retrieved is None:
            print("FAIL: get_index returned None")
            return 1
        if retrieved.metadata.file_count != 3:
            print(f"FAIL: retrieved file_count={retrieved.metadata.file_count}")
            return 1

        print("repo-indexing-full-ok")
        return 0
    finally:
        shutil.rmtree(tmpdir, ignore_errors=True)


def _cmd_incremental_refresh() -> int:
    svc = _make_service()
    tmpdir = _make_sample_dir()
    try:
        # Initial index
        original = svc.index_resource(_RID, tmpdir)
        orig_hash = next(
            (f.content_hash for f in original.files if f.path == "app.py"),
            None,
        )
        if orig_hash is None:
            print("FAIL: app.py not found in original index")
            return 1

        # Modify a file
        (Path(tmpdir) / "app.py").write_text("print('updated')\n# more\n")

        # Refresh
        refreshed = svc.refresh_index(_RID, tmpdir)
        new_hash = next(
            (f.content_hash for f in refreshed.files if f.path == "app.py"),
            None,
        )
        if new_hash is None:
            print("FAIL: app.py not found in refreshed index")
            return 1

        if orig_hash == new_hash:
            print("FAIL: hash should have changed")
            return 1

        # Unchanged file
        orig_lib = next(
            (f.content_hash for f in original.files if f.path == "lib.py"),
            None,
        )
        new_lib = next(
            (f.content_hash for f in refreshed.files if f.path == "lib.py"),
            None,
        )
        if orig_lib is None or new_lib is None:
            print("FAIL: lib.py not found in index")
            return 1
        if orig_lib != new_lib:
            print("FAIL: unchanged file hash changed")
            return 1

        print("repo-indexing-refresh-ok")
        return 0
    finally:
        shutil.rmtree(tmpdir, ignore_errors=True)


def _cmd_policy_enforcement() -> int:
    svc = _make_service()
    tmpdir = _make_sample_dir()
    try:
        (Path(tmpdir) / "big.dat").write_bytes(b"x" * 2_000_000)
        # Max file size
        idx = svc.index_resource(
            "01KK0D8WNATFNEX2JMG5GKF6FR", tmpdir, max_file_size=1_000_000
        )
        paths = [f.path for f in idx.files]
        if "big.dat" in paths:
            print("FAIL: big.dat should be excluded by max_file_size")
            return 1

        # Include globs
        idx2 = svc.index_resource(
            "01KK0DAE56DZVRZKRHA53K1MQG", tmpdir, include_globs=("*.py",)
        )
        if idx2.metadata.file_count != 2:
            print(f"FAIL: include_globs *.py got {idx2.metadata.file_count} files")
            return 1

        # Removal
        removed = svc.remove_index("01KK0D8WNATFNEX2JMG5GKF6FR")
        if not removed:
            print("FAIL: remove_index returned False")
            return 1
        if svc.get_index("01KK0D8WNATFNEX2JMG5GKF6FR") is not None:
            print("FAIL: index still exists after removal")
            return 1

        print("repo-indexing-policy-ok")
        return 0
    finally:
        shutil.rmtree(tmpdir, ignore_errors=True)


_COMMANDS: dict[str, Callable[[], int]] = {
    "full-index": _cmd_full_index,
    "incremental-refresh": _cmd_incremental_refresh,
    "policy-enforcement": _cmd_policy_enforcement,
}


def main() -> int:
    if len(sys.argv) < 2:
        print(
            "Usage: helper_repo_indexing.py "
            "<full-index|incremental-refresh|policy-enforcement>"
        )
        return 1
    command = sys.argv[1]
    handler = _COMMANDS.get(command)
    if handler is None:
        print(f"Unknown command: {command}")
        return 1
    return handler()


if __name__ == "__main__":
    sys.exit(main())