cleveragents-core/robot/helper_project_repository.py

"""Helper for project repository Robot integration tests.

Tests NamespacedProjectRepository and ProjectResourceLinkRepository
with an in-memory SQLite database.
"""

from __future__ import annotations

import sys
from datetime import UTC, datetime

from sqlalchemy import create_engine, event, text
from sqlalchemy.orm import Session, sessionmaker

from cleveragents.domain.models.core.project import NamespacedProject
from cleveragents.infrastructure.database.models import (
    Base,
    NamespacedProjectModel,  # noqa: F401
    ProjectResourceLinkModel,  # noqa: F401
    ResourceModel,
    ResourceTypeModel,
)
from cleveragents.infrastructure.database.repositories import (
    DuplicateLinkError,
    NamespacedProjectRepository,
    ProjectNotFoundError,
    ProjectResourceLinkRepository,
)


def _now_iso() -> str:
    return datetime.now(tz=UTC).isoformat()


def _set_sqlite_pragma(dbapi_conn: object, _connection_record: object) -> None:
    cursor = dbapi_conn.cursor()  # type: ignore[union-attr]
    cursor.execute("PRAGMA foreign_keys = ON")
    cursor.close()


def _create_db() -> tuple[sessionmaker[Session], object]:
    engine = create_engine("sqlite:///:memory:")
    event.listen(engine, "connect", _set_sqlite_pragma)
    with engine.connect() as conn:
        conn.execute(text("PRAGMA foreign_keys = ON"))
        conn.commit()
    Base.metadata.create_all(engine)
    sf = sessionmaker(bind=engine)
    return sf, engine


def _setup_resource_type(session: Session) -> None:
    rt = ResourceTypeModel()
    rt.name = "builtin/git-checkout"  # type: ignore[assignment]
    rt.namespace = "builtin"  # type: ignore[assignment]
    rt.resource_kind = "physical"  # type: ignore[assignment]
    rt.user_addable = True  # type: ignore[assignment]
    rt.created_at = _now_iso()  # type: ignore[assignment]
    rt.updated_at = _now_iso()  # type: ignore[assignment]
    session.add(rt)
    session.commit()


def _create_resource(session: Session, resource_id: str) -> None:
    r = ResourceModel()
    r.resource_id = resource_id  # type: ignore[assignment]
    r.type_name = "builtin/git-checkout"  # type: ignore[assignment]
    r.resource_kind = "physical"  # type: ignore[assignment]
    r.created_at = _now_iso()  # type: ignore[assignment]
    r.updated_at = _now_iso()  # type: ignore[assignment]
    session.add(r)
    session.commit()


def test_project_crud() -> None:
    """Test project create, get, list, update, delete."""
    sf, _engine = _create_db()
    session = sf()
    _setup_resource_type(session)

    repo = NamespacedProjectRepository(session_factory=lambda: session)

    # Create
    project = NamespacedProject(name="test-proj", namespace="local")
    repo.create(project)
    session.commit()

    # Get
    fetched = repo.get("local/test-proj")
    assert fetched.name == "test-proj"
    assert fetched.namespace == "local"

    # List
    projects = repo.list_projects()
    assert len(projects) == 1

    # Update
    updated = fetched.model_copy(update={"description": "Updated"})
    repo.update(updated)
    session.commit()
    re_fetched = repo.get("local/test-proj")
    assert re_fetched.description == "Updated"

    # Delete
    deleted = repo.delete("local/test-proj")
    session.commit()
    assert deleted is True

    # Verify gone
    try:
        repo.get("local/test-proj")
        raise AssertionError("Expected ProjectNotFoundError")
    except ProjectNotFoundError:
        pass

    print("project-crud-ok")


def test_link_crud() -> None:
    """Test link create, list, get, remove, alias uniqueness."""
    sf, _engine = _create_db()
    session = sf()
    _setup_resource_type(session)

    project_repo = NamespacedProjectRepository(session_factory=lambda: session)
    link_repo = ProjectResourceLinkRepository(session_factory=lambda: session)

    # Create project
    project = NamespacedProject(name="link-proj", namespace="local")
    project_repo.create(project)
    session.commit()

    # Create resources
    _create_resource(session, "00000000000000000000000001")
    _create_resource(session, "00000000000000000000000002")

    # Create links
    link1 = link_repo.create_link(
        project_name="local/link-proj",
        resource_id="00000000000000000000000001",
        alias="repo-a",
    )
    session.commit()
    assert link1.link_id is not None
    assert len(str(link1.link_id)) == 26

    link_repo.create_link(
        project_name="local/link-proj",
        resource_id="00000000000000000000000002",
        alias="repo-b",
    )
    session.commit()

    # List
    links = link_repo.list_links("local/link-proj")
    assert len(links) == 2

    # Get
    fetched = link_repo.get_link(str(link1.link_id))
    assert fetched is not None

    # Duplicate alias
    try:
        _create_resource(session, "00000000000000000000000003")
        link_repo.create_link(
            project_name="local/link-proj",
            resource_id="00000000000000000000000003",
            alias="repo-a",
        )
        raise AssertionError("Expected DuplicateLinkError")
    except DuplicateLinkError:
        pass

    # Remove
    removed = link_repo.remove_link(str(link1.link_id))
    session.commit()
    assert removed is True

    # Verify removed
    remaining = link_repo.list_links("local/link-proj")
    assert len(remaining) == 1

    print("link-crud-ok")


if __name__ == "__main__":
    test_name = sys.argv[1] if len(sys.argv) > 1 else "project-crud"
    if test_name == "project-crud":
        test_project_crud()
    elif test_name == "link-crud":
        test_link_crud()
    else:
        print(f"Unknown test: {test_name}", file=sys.stderr)
        sys.exit(1)