feat(placements): v0.11 P5 learner placement visibility (REQ-11-005) [5/10] — wave 5-1: gate + scoped-events + hired + revoked-null + mine arms + reads-stay-free tests

apps/ai-service/tests/placements/test_learner_status.py

@@ -0,0 +1,260 @@
+"""Learner application-status tests (D-104/G-322, REQ-11-005) — the
+mine-read discovery arms, the status read's scoped-history + hired
+placement arms, the revoked honest-null arm, and the reads-stay-free
+pin (no self-429, G-313)."""
+
+from __future__ import annotations
+
+import pytest
+from fastapi.testclient import TestClient
+
+from ai_service.marketplace.base import Posting
+from ai_service.marketplace.store import SQLiteMarketplaceStore
+
+from .conftest import ADMIN_TOKEN, PLACEMENT_LEARNERS, make_client
+
+
+def _seed_posting(store: SQLiteMarketplaceStore, employer_id: str = "emp-test") -> str:
+    import uuid
+
+    posting = store.insert_posting(
+        Posting(
+            id=f"post-{uuid.uuid4().hex[:12]}",
+            source="admin",
+            external_id=f"test-{uuid.uuid4().hex[:8]}",
+            employer_id=employer_id,
+            title="AI Orchestration Engineer",
+        )
+    )
+    # the learner apply route requires published (409 on drafts)
+    store.set_posting_status(posting.id, "published")
+    return posting.id
+
+
+def _apply(client: TestClient, posting_id: str, learner_id: str = "pilot-learner") -> str:
+    resp = client.post(
+        "/v1/marketplace/apply", json={"posting_id": posting_id, "learner_id": learner_id}
+    )
+    assert resp.status_code == 200, resp.text
+    return resp.json()["application_id"]
+
+
+def _admin(token: str = ADMIN_TOKEN) -> dict[str, str]:
+    return {"X-Admin-Token": token}
+
+
+def _transition(client: TestClient, application_id: str, body: dict) -> None:
+    resp = client.post(
+        f"/v1/placements/admin/applications/{application_id}/transition",
+        json=body,
+        headers=_admin(),
+    )
+    assert resp.status_code == 200, resp.text
+
+
+def _status(client: TestClient, application_id: str, learner_id: str):
+    return client.get(
+        f"/v1/marketplace/applications/{application_id}",
+        params={"learner_id": learner_id},
+    )
+
+
+@pytest.fixture()
+def scaffold(tmp_path, identity_store, marketplace_store):
+    client = make_client(tmp_path, identity_store, marketplace_store)
+    with client as entered:
+        posting_id = _seed_posting(marketplace_store)
+        application_id = _apply(entered, posting_id)
+        yield {
+            "client": entered,
+            "posting_id": posting_id,
+            "application_id": application_id,
+        }
+
+
+# -- the mine-read (G-322): discovery on revisit -------------------------------
+
+
+def test_mine_own_200(scaffold) -> None:
+    resp = scaffold["client"].get(
+        "/v1/marketplace/applications/mine",
+        params={"posting_id": scaffold["posting_id"], "learner_id": "pilot-learner"},
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["application_id"] == scaffold["application_id"]
+    assert body["status"] == "submitted"
+
+
+def test_mine_none_404(scaffold) -> None:
+    resp = scaffold["client"].get(
+        "/v1/marketplace/applications/mine",
+        params={"posting_id": scaffold["posting_id"], "learner_id": "pilot-learner-2"},
+    )
+    assert resp.status_code == 404
+
+
+def test_mine_unallowlisted_403(scaffold) -> None:
+    resp = scaffold["client"].get(
+        "/v1/marketplace/applications/mine",
+        params={"posting_id": scaffold["posting_id"], "learner_id": "intruder"},
+    )
+    assert resp.status_code == 403
+
+
+def test_mine_unverified_403(tmp_path, identity_store, marketplace_store) -> None:
+    """A learner on the roster but WITHOUT a verified identity record →
+    the verified-18+ arm refuses (the D-043 composition)."""
+    client = make_client(tmp_path, identity_store, marketplace_store)
+    with client as entered:
+        posting_id = _seed_posting(marketplace_store)
+        from ..conftest import SUITE_LEARNERS
+
+        unverified = next(name for name in SUITE_LEARNERS if name not in PLACEMENT_LEARNERS)
+        resp = entered.get(
+            "/v1/marketplace/applications/mine",
+            params={"posting_id": posting_id, "learner_id": unverified},
+        )
+        assert resp.status_code == 403
+
+
+# -- the status read (D-104): status + scoped history + placement ---------------
+
+
+def test_status_own_200_with_events(scaffold) -> None:
+    client = scaffold["client"]
+    _transition(client, scaffold["application_id"], {"to": "in_review"})
+    resp = _status(client, scaffold["application_id"], "pilot-learner")
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["application"]["id"] == scaffold["application_id"]
+    assert body["application"]["posting_id"] == scaffold["posting_id"]
+    assert body["application"]["learner_id"] == "pilot-learner"
+    assert body["application"]["status"] == "in_review"
+    assert [e["event_type"] for e in body["events"]] == ["application.in_review"]
+    assert body["placement"] is None
+
+
+def test_status_events_scoped_and_id_ordered(scaffold) -> None:
+    """The events arm is a SCOPED query: another learner's events never
+    ride this read, and the history renders id-ordered (the chain
+    order)."""
+    client = scaffold["client"]
+    import uuid
+
+    store = client.app.state.marketplace_store
+    posting2 = store.insert_posting(
+        Posting(
+            id=f"post-b-{uuid.uuid4().hex[:6]}",
+            source="admin",
+            external_id=f"b-{uuid.uuid4().hex[:8]}",
+            employer_id="emp-test",
+            title="Second posting",
+        )
+    )
+    store.set_posting_status(posting2.id, "published")
+    other = _apply(client, posting2.id, learner_id="pilot-learner-2")
+    _transition(client, other, {"to": "interview"})
+    _transition(client, scaffold["application_id"], {"to": "in_review"})
+    _transition(client, scaffold["application_id"], {"to": "interview"})
+    resp = _status(client, scaffold["application_id"], "pilot-learner")
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert [e["event_type"] for e in body["events"]] == [
+        "application.in_review",
+        "application.interview",
+    ]
+    assert all(
+        e["application_id"] == scaffold["application_id"] for e in body["events"]
+    )
+
+
+def test_status_hired_carries_placement(scaffold) -> None:
+    client = scaffold["client"]
+    _transition(
+        client,
+        scaffold["application_id"],
+        {"to": "hired", "role_title": "Placement Engineer", "start_date": "2026-11-02"},
+    )
+    resp = _status(client, scaffold["application_id"], "pilot-learner")
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["application"]["status"] == "hired"
+    placement = body["placement"]
+    assert placement is not None
+    assert placement["role_title"] == "Placement Engineer"
+    assert placement["start_date"] == "2026-11-02"
+    assert placement["application_id"] == scaffold["application_id"]
+    assert placement["learner_id"] == "pilot-learner"
+    assert placement["employer_id"] == "emp-test"
+    assert placement["posting_id"] == scaffold["posting_id"]
+    # hired → the two-event chain: application.hired + placement.created
+    assert [e["event_type"] for e in body["events"]] == [
+        "application.hired",
+        "placement.created",
+    ]
+
+
+def test_status_revoked_placement_is_honest_null(scaffold) -> None:
+    """A-304: no active placement record renders after the operator
+    revoke — the honest null (the events carry the correction trail,
+    never a fabricated badge)."""
+    client = scaffold["client"]
+    app_id = scaffold["application_id"]
+    _transition(
+        client,
+        app_id,
+        {"to": "hired", "role_title": "Placement Engineer", "start_date": "2026-11-02"},
+    )
+    placement_id = _status(client, app_id, "pilot-learner").json()["placement"]["id"]
+    resp = client.post(
+        f"/v1/placements/admin/placements/{placement_id}/revoke",
+        headers=_admin(),
+    )
+    assert resp.status_code == 200, resp.text
+    resp = _status(client, app_id, "pilot-learner")
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["placement"] is None
+    assert body["events"][-1]["event_type"] == "placement.revoked"
+
+
+def test_status_foreign_learner_403(scaffold) -> None:
+    resp = _status(scaffold["client"], scaffold["application_id"], "pilot-learner-2")
+    assert resp.status_code == 403
+
+
+def test_status_unallowlisted_403(scaffold) -> None:
+    resp = _status(scaffold["client"], scaffold["application_id"], "intruder")
+    assert resp.status_code == 403
+
+
+def test_status_unverified_403(tmp_path, identity_store, marketplace_store) -> None:
+    client = make_client(tmp_path, identity_store, marketplace_store)
+    with client as entered:
+        posting_id = _seed_posting(marketplace_store)
+        from ..conftest import SUITE_LEARNERS
+
+        unverified = next(name for name in SUITE_LEARNERS if name not in PLACEMENT_LEARNERS)
+        marketplace_store.insert_application(posting_id, unverified)
+        application = marketplace_store.list_applications(posting_id)[0]
+        resp = _status(entered, application.id, unverified)
+        assert resp.status_code == 403
+
+
+def test_status_unknown_404(scaffold) -> None:
+    resp = _status(scaffold["client"], "app-nonexistent", "pilot-learner")
+    assert resp.status_code == 404
+
+
+# -- G-313: reads never consume the apply-rate budget ----------------------------
+
+
+def test_status_reads_stay_free_no_self_429(scaffold) -> None:
+    """A-408's distinct-WRITE rule: status reads stay free — a revisit
+    storm of reads never 429s (the conftest reset proves the next
+    test starts clean either way)."""
+    client = scaffold["client"]
+    for _ in range(10):
+        resp = _status(client, scaffold["application_id"], "pilot-learner")
+        assert resp.status_code == 200, resp.text