FabledScribe/tests/test_integration_shape_classify.py

"""Real-Postgres integration tests for shape classification (#2789).

What mocks can't prove: the all-or-nothing batch against real rows, the
write-ACL gate, the todo query's filters, and the consumer map riding
get_snippet. Ledger rows are seeded through the same sync the coverage walk
uses — no forge needed, the sync takes extracted shapes directly.
"""
import pytest
import pytest_asyncio
from sqlalchemy import select

from scribe.models import async_session, engine
from scribe.models.code_shape import CodeShape
from scribe.models.project import Project
from scribe.models.user import User
from scribe.services.shape_ledger import (
    classify_shapes,
    list_project_shapes,
    snippet_consumers,
    sync_repo_shapes,
)

pytestmark = pytest.mark.integration

REPO = "git.example.com/alice/widget"
SHAPES = [
    ("src/app.py", "sym", "make_app"),
    ("src/app.py", "sym", "Config"),
    ("src/util.py", "sym", "helper"),
    ("web/button.css", "css", "btn"),
]


@pytest_asyncio.fixture(autouse=True)
async def _dispose_engine():
    """Dispose the app's module-level engine after each test.

    The engine pools asyncpg connections per event loop, but pytest-asyncio runs
    each test on a fresh loop — so without this, test 2 gets handed test 1's
    connection bound to a now-dead loop ("Future attached to a different loop").
    Disposing in the test's own loop teardown clears the pool cleanly.
    """
    yield
    await engine.dispose()


async def _user(session, username: str) -> User:
    existing = (
        await session.execute(select(User).where(User.username == username))
    ).scalar_one_or_none()
    if existing is not None:
        return existing
    user = User(username=username)
    session.add(user)
    await session.flush()
    return user


@pytest_asyncio.fixture
async def seeded():
    """Owner + outsider, a project with a synced 4-shape ledger, one snippet."""
    from scribe.services import snippets as snippets_svc

    async with async_session() as s:
        owner = await _user(s, "classify_owner")
        other = await _user(s, "classify_other")
        project = Project(user_id=owner.id, title="Classify target")
        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, SHAPES, seen_marker="main")
    snippet = await snippets_svc.create_snippet(
        ids["owner"], name="cls_make_app", code="def make_app():\n    pass\n",
        language="python", repo="Widget", path="src/factory.py",
        symbol="factory", project_id=ids["pid"],
    )
    ids["snippet"] = int(snippet.id)
    return ids


@pytest.mark.integration
async def test_classify_applies_judgments_and_reports_unmatched(seeded):
    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
    out = await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "instance",
         "snippet_id": sid},
        {"path": "src/util.py", "symbol": "helper", "status": "exempt",
         "reason": "test scaffolding, deliberately local"},
        {"path": "web/button.css", "symbol": "btn", "status": "variant",
         "snippet_id": sid, "reason": "darker focus ring for the toolbar"},
        {"path": "gone.py", "symbol": "nothing", "status": "exempt",
         "reason": "x"},
    ], via="audit")
    assert out["classified"] == 3
    assert out["unmatched"] == [{"path": "gone.py", "symbol": "nothing"}]

    rows, total = await list_project_shapes(owner, pid)
    by_symbol = {r.symbol: r for r in rows}
    assert total == 4
    assert by_symbol["make_app"].status == "instance"
    assert by_symbol["make_app"].snippet_id == sid
    assert by_symbol["make_app"].classified_by == "audit"
    assert by_symbol["helper"].status == "exempt"
    assert by_symbol["helper"].reason == "test scaffolding, deliberately local"
    assert by_symbol["btn"].status == "variant"
    assert by_symbol["Config"].status == "unclassified"

    # Withdrawing a judgment returns the shape to the todo, fields cleared.
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "unclassified"},
    ])
    rows, _ = await list_project_shapes(owner, pid, status="unclassified")
    assert {r.symbol for r in rows} == {"Config", "make_app"}
    make_app = next(r for r in rows if r.symbol == "make_app")
    assert make_app.snippet_id is None and make_app.classified_by is None


@pytest.mark.integration
async def test_a_bad_batch_applies_nothing(seeded):
    """All-or-nothing (#2709's lesson): a caller must never learn later that
    half a batch silently happened."""
    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
    with pytest.raises(ValueError) as err:
        await classify_shapes(owner, pid, [
            {"path": "src/app.py", "symbol": "make_app", "status": "instance",
             "snippet_id": sid},
            {"path": "src/util.py", "symbol": "helper", "status": "variant",
             "snippet_id": sid},  # variant with no reason: structural error
        ])
    assert "needs a reason" in str(err.value)
    rows, _ = await list_project_shapes(owner, pid, status="unclassified")
    assert len(rows) == 4  # including make_app — the valid half did NOT apply

    # A snippet target the caller can't read is the same: nothing applies.
    with pytest.raises(ValueError):
        await classify_shapes(owner, pid, [
            {"path": "src/app.py", "symbol": "make_app", "status": "instance",
             "snippet_id": 999999999},
        ])


@pytest.mark.integration
async def test_classification_is_write_gated_and_listing_read_gated(seeded):
    other, pid, sid = seeded["other"], seeded["pid"], seeded["snippet"]
    with pytest.raises(ValueError):
        await classify_shapes(other, pid, [
            {"path": "src/app.py", "symbol": "make_app", "status": "exempt",
             "reason": "not their call to make"},
        ])
    assert await list_project_shapes(other, pid) == ([], 0)


@pytest.mark.integration
async def test_list_filters_compose(seeded):
    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "instance",
         "snippet_id": sid},
    ])
    rows, total = await list_project_shapes(owner, pid, path="src")
    assert total == 3 and all(r.path.startswith("src/") for r in rows)
    # Directory semantics, not string prefix: "sr" matches nothing.
    assert (await list_project_shapes(owner, pid, path="sr"))[1] == 0
    rows, total = await list_project_shapes(owner, pid, snippet_id=sid)
    assert total == 1 and rows[0].symbol == "make_app"
    rows, total = await list_project_shapes(
        owner, pid, status="unclassified", limit=2
    )
    assert total == 3 and len(rows) == 2  # paged, with the true total


@pytest.mark.integration
async def test_get_snippet_carries_the_structured_consumer_map(seeded):
    from scribe.mcp._context import _user_id_ctx
    from scribe.mcp.tools.snippets import get_snippet

    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "instance",
         "snippet_id": sid},
        {"path": "web/button.css", "symbol": "btn", "status": "variant",
         "snippet_id": sid, "reason": "darker focus ring"},
    ])
    token = _user_id_ctx.set(owner)
    try:
        data = await get_snippet(snippet_id=sid)
    finally:
        _user_id_ctx.reset(token)
    assert [i["path"] for i in data["instances"]] == ["src/app.py"]
    assert data["variants"][0]["reason"] == "darker focus ring"

    # The map is caller-scoped: an outsider asking the service directly gets
    # silence, not another project's file layout.
    consumers = await snippet_consumers(seeded["other"], sid)
    assert consumers == {"instances": [], "variants": []}


@pytest.mark.integration
async def test_sync_refiles_rows_whose_snippet_was_purged(seeded):
    """The SET NULL companion (#2787): a judgment whose target is hard-deleted
    rejoins the todo on the next sync instead of dangling target-less."""
    owner, pid, sid = seeded["owner"], seeded["pid"], seeded["snippet"]
    await classify_shapes(owner, pid, [
        {"path": "src/app.py", "symbol": "make_app", "status": "instance",
         "snippet_id": sid},
    ])
    from scribe.models.note import Note

    async with async_session() as s:
        note = await s.get(Note, sid)
        await s.delete(note)  # hard delete, as purge_trash would
        await s.commit()
    await sync_repo_shapes(pid, REPO, SHAPES, seen_marker="main")
    async with async_session() as s:
        row = (await s.execute(select(CodeShape).where(
            CodeShape.project_id == pid, CodeShape.symbol == "make_app",
        ))).scalar_one()
    assert row.status == "unclassified"
    assert row.snippet_id is None