FabledGrimoire/tests/test_integration_shape_move.py

"""Real-Postgres integration tests for moving a directory's shape judgments.

Ledger rows are keyed by path. A directory moved in the tree has to carry its
rows to the new path, or the next sync vanishes every judgment and lists the
same shapes again, unjudged.
"""
import pytest
import pytest_asyncio
from sqlalchemy import select

from grimoire.models import async_session
from grimoire.models.code_shape import CodeShape, CodeShapeEvent
from grimoire.models.project import Project
from grimoire.services.shape_ledger import classify_shapes, move_shapes, sync_repo_shapes
from tests.helpers import ensure_user

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

REPO = "git.example.com/alice/widget"
OLD_TREE = [
    ("src/old/app.py", "sym", "make_app"),
    ("src/old/sub/util.py", "sym", "helper"),
    ("src/older/keep.py", "sym", "untouched"),
]
NEW_TREE = [
    ("src/new/app.py", "sym", "make_app"),
    ("src/new/sub/util.py", "sym", "helper"),
    ("src/older/keep.py", "sym", "untouched"),
]


@pytest_asyncio.fixture
async def seeded():
    """A project whose ledger was synced before the move, one shape judged."""
    async with async_session() as s:
        owner = await ensure_user(s, "shape_move_owner")
        other = await ensure_user(s, "shape_move_other")
        project = Project(user_id=owner.id, title="Moved directory")
        s.add(project)
        await s.flush()
        ids = {"owner": owner.id, "other": other.id, "pid": project.id}
        await s.commit()

    await sync_repo_shapes(ids["pid"], REPO, OLD_TREE, seen_marker="c1")
    await classify_shapes(ids["owner"], ids["pid"], [
        {"path": "src/old/sub/util.py", "symbol": "helper", "status": "exempt",
         "reason": "one caller"},
    ])
    return ids


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


@pytest.mark.integration
async def test_the_judgments_move_and_the_next_sync_finds_them(seeded):
    pid = seeded["pid"]
    result = await move_shapes(seeded["owner"], pid, "src/old/", "src/new")

    assert result == {"moved": 2, "replaced": 0, "kept": []}
    await sync_repo_shapes(pid, REPO, NEW_TREE, seen_marker="c2")
    assert await _ledger(pid) == {
        ("src/new/app.py", "make_app"): ("unclassified", True),
        ("src/new/sub/util.py", "helper"): ("exempt", True),
        ("src/older/keep.py", "untouched"): ("unclassified", True),
    }
    async with async_session() as s:
        moved = (await s.execute(
            select(CodeShapeEvent.path).where(
                CodeShapeEvent.project_id == pid, CodeShapeEvent.event == "moved",
            )
        )).scalars().all()
    assert sorted(moved) == ["src/old/app.py", "src/old/sub/util.py"]


@pytest.mark.integration
async def test_after_the_sync_got_there_first_the_judgment_replaces_the_fresh_row(seeded):
    pid = seeded["pid"]
    await sync_repo_shapes(pid, REPO, NEW_TREE, seen_marker="c2")

    result = await move_shapes(seeded["owner"], pid, "src/old", "src/new")

    assert result == {"moved": 2, "replaced": 2, "kept": []}
    assert await _ledger(pid) == {
        ("src/new/app.py", "make_app"): ("unclassified", True),
        ("src/new/sub/util.py", "helper"): ("exempt", True),
        ("src/older/keep.py", "untouched"): ("unclassified", True),
    }


@pytest.mark.integration
async def test_a_judgment_already_at_the_new_path_stands(seeded):
    pid = seeded["pid"]
    await sync_repo_shapes(pid, REPO, OLD_TREE + NEW_TREE, seen_marker="c2")
    await classify_shapes(seeded["owner"], pid, [
        {"path": "src/new/sub/util.py", "symbol": "helper", "status": "exempt",
         "reason": "judged again after the move"},
    ])

    result = await move_shapes(seeded["owner"], pid, "src/old", "src/new")

    assert result["moved"] == 1 and result["replaced"] == 1
    assert result["kept"] == [
        {"path": "src/old/sub/util.py", "symbol": "helper", "kind": "sym"},
    ]
    ledger = await _ledger(pid)
    assert ledger[("src/new/sub/util.py", "helper")] == ("exempt", True)
    assert ("src/old/sub/util.py", "helper") in ledger


@pytest.mark.integration
async def test_overlapping_and_unwritable_moves_are_refused(seeded):
    owner, other, pid = seeded["owner"], seeded["other"], seeded["pid"]
    with pytest.raises(ValueError, match="overlap"):
        await move_shapes(owner, pid, "src/old", "src/old/inner")
    assert await move_shapes(owner, pid, "src/gone", "src/new") == {
        "moved": 0, "replaced": 0, "kept": [],
    }
    with pytest.raises(ValueError, match="no write access"):
        await move_shapes(other, pid, "src/old", "src/new")
    assert ("src/old/app.py", "make_app") in await _ledger(pid)


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

    owner, pid = seeded["owner"], seeded["pid"]
    note = await snippets_svc.create_snippet(
        owner, name="move_helper", code="def helper():\n    pass\n",
        language="python", project_id=pid, locations=[
            {"repo": "Widget", "path": "src/old/sub/util.py", "symbol": "helper"},
            {"repo": "Widget", "path": "src/older/keep.py", "symbol": "untouched"},
        ],
    )

    result = await snippets_svc.move_locations(owner, pid, "src/old", "src/new")

    assert result == {"moved": [int(note.id)], "read_only": []}
    moved = await snippets_svc.get_snippet(owner, int(note.id))
    assert [loc["path"] for loc in snippets_svc.snippet_fields(moved)["locations"]] == [
        "src/new/sub/util.py", "src/older/keep.py",
    ]