FabledGrimoire/tests/test_integration_repo_move.py

"""Real-Postgres integration tests for moving a renamed repo's binding.

A repo renamed on its forge has a new remote. The binding and every ledger row
are keyed by the remote, so the move has to carry both, in one transaction, and
refuse anything that would merge two ledgers or move another project's repo.
"""
import pytest
import pytest_asyncio
from sqlalchemy import select

from grimoire.models import async_session
from grimoire.models.code_shape import CodeShape
from grimoire.models.project import Project
from grimoire.services import repo_bindings as svc
from grimoire.services.shape_ledger import classify_shapes, sync_repo_shapes
from tests.helpers import ensure_user

pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]

OLD = "https://git.example.com/alice/OldName.git"
NEW = "git@git.example.com:alice/NewName.git"
OLD_KEY = "git.example.com/alice/oldname"
NEW_KEY = "git.example.com/alice/newname"
SHAPES = [("src/app.py", "sym", "make_app"), ("src/util.py", "sym", "helper")]


@pytest_asyncio.fixture
async def bound():
    """A project bound under the old name, its two-shape ledger, one judged."""
    async with async_session() as s:
        owner = await ensure_user(s, "repo_move_owner")
        project = Project(user_id=owner.id, title="Renamed repo")
        other = Project(user_id=owner.id, title="Some other project")
        s.add_all([project, other])
        await s.flush()
        ids = {"owner": owner.id, "pid": project.id, "other": other.id}
        await s.commit()

    await svc.set_binding(ids["owner"], OLD, ids["pid"], "dev")
    await sync_repo_shapes(ids["pid"], OLD_KEY, SHAPES, seen_marker="dev")
    await classify_shapes(ids["owner"], ids["pid"], [
        {"path": "src/util.py", "symbol": "helper", "status": "exempt",
         "reason": "one caller"},
    ])
    yield ids
    await svc.delete_binding(ids["owner"], OLD)
    await svc.delete_binding(ids["owner"], NEW)


async def _ledger(pid: int) -> dict[str, tuple[str, str]]:
    async with async_session() as s:
        rows = await s.execute(select(CodeShape).where(CodeShape.project_id == pid))
        return {r.symbol: (r.repo_key, r.status) for r in rows.scalars()}


@pytest.mark.integration
async def test_the_binding_and_every_judgment_move_to_the_new_name(bound):
    binding, moved = await svc.move_binding(bound["owner"], OLD, NEW, bound["pid"])

    assert moved == 2
    assert binding.repo_key == NEW_KEY
    assert binding.project_id == bound["pid"]
    assert binding.ref == "dev"
    assert await svc.resolve_project(bound["owner"], NEW) == bound["pid"]
    assert await svc.resolve_project(bound["owner"], OLD) is None
    assert await _ledger(bound["pid"]) == {
        "make_app": (NEW_KEY, "unclassified"),
        "helper": (NEW_KEY, "exempt"),
    }


@pytest.mark.integration
async def test_it_refuses_a_repo_bound_to_another_project(bound):
    with pytest.raises(ValueError, match="not bound to project"):
        await svc.move_binding(bound["owner"], OLD, NEW, bound["other"])
    assert {k for k, _ in (await _ledger(bound["pid"])).values()} == {OLD_KEY}


@pytest.mark.integration
async def test_it_refuses_when_the_new_name_is_already_bound(bound):
    await svc.set_binding(bound["owner"], NEW, bound["pid"])
    with pytest.raises(ValueError, match="already bound"):
        await svc.move_binding(bound["owner"], OLD, NEW, bound["pid"])
    assert {k for k, _ in (await _ledger(bound["pid"])).values()} == {OLD_KEY}


@pytest.mark.integration
async def test_it_refuses_to_merge_into_a_ledger_the_new_name_already_has(bound):
    await sync_repo_shapes(
        bound["pid"], NEW_KEY, [("src/new.py", "sym", "fresh")], seen_marker="dev",
    )
    with pytest.raises(ValueError, match="already has a ledger"):
        await svc.move_binding(bound["owner"], OLD, NEW, bound["pid"])
    assert await svc.resolve_project(bound["owner"], OLD) == bound["pid"]


@pytest.mark.integration
async def test_the_same_repo_under_both_names_is_refused(bound):
    with pytest.raises(ValueError, match="same repo"):
        await svc.move_binding(
            bound["owner"], OLD, "git@git.example.com:alice/oldname", bound["pid"],
        )


@pytest.mark.integration
async def test_the_snippets_recorded_in_the_old_repo_move_with_it(bound):
    from grimoire.services import snippets as snippets_svc

    owner, pid = bound["owner"], bound["pid"]

    async def snippet(name, *repos):
        note = await snippets_svc.create_snippet(
            owner, name=name, code=f"def {name}():\n    pass\n", language="python",
            project_id=pid,
            locations=[{"repo": r, "path": "src/app.py", "symbol": name} for r in repos],
        )
        return int(note.id)

    moved_one = await snippet("moved_one", "OldName", "alice/oldname", OLD)
    mixed = await snippet("mixed", "git.example.com/alice/oldname", "elsewhere/gadget")
    untouched = await snippet("untouched", "Gadget", "git.example.com/alice/oldnamesake")

    result = await snippets_svc.move_location_repo(owner, pid, OLD, NEW)

    assert sorted(result["moved"]) == sorted([moved_one, mixed])
    assert result["read_only"] == []

    async def repos(snippet_id):
        note = await snippets_svc.get_snippet(owner, snippet_id)
        return [loc["repo"] for loc in snippets_svc.snippet_fields(note)["locations"]]

    assert await repos(moved_one) == [NEW_KEY, NEW_KEY, NEW_KEY]
    assert await repos(mixed) == [NEW_KEY, "elsewhere/gadget"]
    assert await repos(untouched) == ["Gadget", "git.example.com/alice/oldnamesake"]