cleveragents-core/robot/helper_project_migration.py

"""Helper utilities for project migration Robot integration tests.

Covers the integration between:
- NamespacedProjectModel <-> NamespacedProject domain model
- ProjectResourceLinkModel persistence
- Round-trip: domain -> SQLAlchemy -> domain
"""

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,
    ProjectResourceLinkModel,
    ResourceModel,
    ResourceTypeModel,
)


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


def _make_ulid(suffix: str = "0") -> str:
    return ("0" * (26 - len(suffix))) + suffix


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[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)
    factory = sessionmaker(bind=engine)
    session = factory()
    return session, engine


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


def _setup_resource(session: Session) -> str:
    rid = _make_ulid("1")
    r = ResourceModel()
    r.resource_id = rid
    r.namespaced_name = "local/test-repo"
    r.namespace = "local"
    r.type_name = "builtin/git-checkout"
    r.resource_kind = "physical"
    r.location = "/tmp/test-repo"
    r.created_at = _now_iso()
    r.updated_at = _now_iso()
    session.add(r)
    session.commit()
    return rid


def _project_round_trip() -> None:
    session, _ = _create_db()
    proj = NamespacedProjectModel(
        namespaced_name="local/round-trip-proj",
        namespace="local",
        description="Round trip test project",
        tags_json="[]",
        created_at=_now_iso(),
        updated_at=_now_iso(),
    )
    session.add(proj)
    session.commit()

    retrieved = session.get(NamespacedProjectModel, "local/round-trip-proj")
    assert retrieved is not None
    assert retrieved.namespace == "local"
    assert retrieved.description == "Round trip test project"
    print("project-round-trip-ok")


def _link_round_trip() -> None:
    session, _ = _create_db()
    _setup_resource_type(session)
    rid = _setup_resource(session)

    proj = NamespacedProjectModel(
        namespaced_name="local/link-test-proj",
        namespace="local",
        tags_json="[]",
        created_at=_now_iso(),
        updated_at=_now_iso(),
    )
    session.add(proj)
    session.commit()

    link = ProjectResourceLinkModel(
        link_id=_make_ulid("99"),
        project_name="local/link-test-proj",
        resource_id=rid,
        read_only=False,
        created_at=_now_iso(),
    )
    session.add(link)
    session.commit()

    retrieved = session.get(ProjectResourceLinkModel, _make_ulid("99"))
    assert retrieved is not None
    assert retrieved.project_name == "local/link-test-proj"
    assert retrieved.resource_id == rid
    print("link-round-trip-ok")


def _domain_conversion() -> None:
    session, _ = _create_db()

    # from_domain
    domain_project = NamespacedProject(
        name="domain-conv-proj",
        namespace="local",
        description="Domain conversion test",
    )
    model = NamespacedProjectModel.from_domain(domain_project)
    assert model.namespaced_name == "local/domain-conv-proj"
    assert model.namespace == "local"

    session.add(model)
    session.commit()

    # to_domain
    retrieved = session.get(NamespacedProjectModel, "local/domain-conv-proj")
    assert retrieved is not None
    domain_back = retrieved.to_domain()
    assert domain_back.name == "domain-conv-proj"
    assert domain_back.namespace == "local"
    assert domain_back.description == "Domain conversion test"
    print("domain-conversion-ok")


def main() -> None:
    if len(sys.argv) < 2:
        print("Usage: helper_project_migration.py <test-name>", file=sys.stderr)
        sys.exit(1)

    test_name = sys.argv[1]
    dispatch: dict[str, object] = {
        "project-round-trip": _project_round_trip,
        "link-round-trip": _link_round_trip,
        "domain-conversion": _domain_conversion,
    }

    handler = dispatch.get(test_name)
    if handler is None:
        print(f"Unknown test: {test_name}", file=sys.stderr)
        sys.exit(1)

    handler()  # type: ignore[operator]


if __name__ == "__main__":
    main()