FabledGrimoire/tests/test_integration_platforms.py

"""Real-Postgres checks for platform membership.

The one invariant: detection only ever ADDS, and only where nobody has
answered. A refresh that overwrote a "no" would put a project back into a
family it was taken out of; one that overwrote a "yes" would erase who said
it. Each is shown here against the real writer, beside the person-facing
writes it must defer to.
"""
import pytest
import pytest_asyncio
from sqlalchemy import select

from grimoire.models import async_session
from grimoire.models.family import Platform
from grimoire.models.project import Project
from grimoire.services import inception as inception_svc
from grimoire.services import platforms as platforms_svc
from tests.helpers import ensure_user

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

OWNER = "platforms_owner"


async def _id(slug: str) -> int:
    async with async_session() as s:
        return await s.scalar(
            select(Platform.id).where(Platform.slug == slug, Platform.deleted_at.is_(None))
        )


@pytest_asyncio.fixture
async def seeded():
    """An owner, an outsider and a fresh project. Each test gets its own
    project, so answers never leak between tests through a shared row."""
    async with async_session() as s:
        owner = await ensure_user(s, OWNER)
        outsider = await ensure_user(s, "platforms_outsider")
        project = Project(user_id=owner.id, title="an app with a server")
        s.add(project)
        await s.flush()
        ids = {"owner": owner.id, "outsider": outsider.id, "pid": project.id}
        await s.commit()
    return ids


def _states(rows: list[dict]) -> dict[str, str]:
    return {r["slug"]: r["state"] for r in rows}


async def test_detection_adds_only_where_nobody_has_answered(seeded):
    owner, pid = seeded["owner"], seeded["pid"]
    await platforms_svc.set_project_platforms(owner, pid, {
        "go": "declared", "rust": "rejected",
    })
    added = await platforms_svc.detect_for_project(pid, [
        "server/go.mod", "core/Cargo.toml", "app/src/main/AndroidManifest.xml",
    ])
    # Only android was unanswered; go and rust keep the person's answers.
    assert added == [await _id("android-app")]
    states = _states(await platforms_svc.project_platforms(owner, pid))
    assert states == {"go": "declared", "rust": "rejected", "android-app": "detected"}


async def test_detection_never_removes_a_platform_whose_marker_is_gone(seeded):
    owner, pid = seeded["owner"], seeded["pid"]
    await platforms_svc.detect_for_project(pid, ["server/go.mod"])
    await platforms_svc.detect_for_project(pid, ["README.md"])
    assert _states(await platforms_svc.project_platforms(owner, pid)) == {"go": "detected"}


async def test_a_withdrawn_answer_lets_detection_decide_again(seeded):
    owner, pid = seeded["owner"], seeded["pid"]
    await platforms_svc.set_project_platforms(owner, pid, {"go": "rejected"})
    assert await platforms_svc.detect_for_project(pid, ["go.mod"]) == []
    await platforms_svc.set_project_platforms(owner, pid, {"go": None})
    assert await platforms_svc.detect_for_project(pid, ["go.mod"]) == [await _id("go")]


async def test_a_bad_update_writes_nothing(seeded):
    owner, pid = seeded["owner"], seeded["pid"]
    with pytest.raises(ValueError, match="unknown platform"):
        await platforms_svc.set_project_platforms(owner, pid, {
            "go": "declared", "not-a-platform": "declared",
        })
    assert await platforms_svc.project_platforms(owner, pid) == []
    with pytest.raises(ValueError, match="no write access"):
        await platforms_svc.set_project_platforms(seeded["outsider"], pid, {"go": "declared"})
    assert await platforms_svc.project_platforms(seeded["outsider"], pid) is None


async def test_inception_records_slugs_and_rejects_what_it_leaves_out(seeded):
    """The inception list is the WHOLE answer. Detection found Android and
    Go; the person says the project is Go and Python — so Android becomes a
    recorded "no", and the next refresh cannot bring it back."""
    owner, pid = seeded["owner"], seeded["pid"]
    await platforms_svc.detect_for_project(pid, ["app/AndroidManifest.xml", "go.mod"])
    defaults = await inception_svc.current_defaults(owner, pid)
    assert {"android-app", "go"} <= {p["slug"] for p in defaults["project_platforms"]}

    out = await inception_svc.decide(owner, pid, via="ui", choices={
        "seed_systems": False, "platforms": ["python", "go"],
    })
    assert _states(out["effects"]["platforms"]) == {
        "android-app": "rejected", "go": "declared", "python": "declared",
    }
    async with async_session() as s:
        project = await s.get(Project, pid)
    assert project.inception["choices"]["platforms"] == ["go", "python"]
    assert await platforms_svc.detect_for_project(pid, ["app/AndroidManifest.xml"]) == []


async def test_an_unknown_inception_platform_applies_nothing(seeded):
    owner, pid = seeded["owner"], seeded["pid"]
    with pytest.raises(ValueError, match="unknown platform"):
        await inception_svc.decide(owner, pid, via="mcp", choices={
            "seed_systems": True, "platforms": ["go", "cobol-mainframe"],
        })
    assert await platforms_svc.project_platforms(owner, pid) == []


async def test_the_catalog_is_admin_written_and_duplicate_gated():
    async with async_session() as s:
        admin = await ensure_user(s, "platforms_admin", role="admin")
        user = await ensure_user(s, OWNER)
        await s.commit()
        admin_id, user_id = admin.id, user.id
    assert await platforms_svc.create_platform(user_id, "Kotlin Multiplatform") is None
    # "Android App" reduces to the seeded android-app: the existing one comes
    # back rather than a second spelling of it.
    dup = await platforms_svc.create_platform(admin_id, "Android App")
    assert dup["duplicate"] and dup["existing_id"] == await _id("android-app")
    with pytest.raises(ValueError, match="repo-relative"):
        await platforms_svc.create_platform(admin_id, "Odd", markers=["/abs"])