state-hub/tests/test_forge_projection.py

"""Deriving a projection from the forge (STATE-WP-0083-T01).

The properties that matter are identity and determinism: a forge-derived
projection must compute the same record identities as repo-manager, and the same
commit must always yield the same projection. Without both, the reset in T03
cannot be verified against anything.
"""

from __future__ import annotations

from pathlib import Path

import pytest

from api.services import forge_projection as fp


def test_identity_matches_the_fleet_derivation():
    """Same namespace as ADR-007, so overlay and forge agree on identity."""
    assert fp.derived_record_uuid("CUST-WP-0067") == "16249302-2767-55df-aec0-d92c2751c225"
    assert fp.derived_record_uuid("CUST-WP-0067-T01") == "f3608db4-20a5-58fb-a965-885eb14858af"


def _repo(tmp_path: Path) -> Path:
    root = tmp_path / "demo"
    (root / "workplans" / "archived").mkdir(parents=True)
    (root / "workplans" / "DEMO-WP-0001-a.md").write_text(
        "---\nid: DEMO-WP-0001\ntype: workplan\ntitle: \"First\"\nstatus: active\n---\n\n"
        "## Do the thing\n\n```task\nid: DEMO-WP-0001-T01\nstatus: todo\npriority: high\n```\n\n"
        "## Do the other\n\n```task\nid: DEMO-WP-0001-T02\nstatus: done\npriority: low\n```\n",
        encoding="utf-8",
    )
    (root / "workplans" / "archived" / "260101-DEMO-WP-0002-b.md").write_text(
        "---\nid: DEMO-WP-0002\ntype: workplan\ntitle: \"Second\"\nstatus: finished\n---\n\n# b\n",
        encoding="utf-8",
    )
    (root / "workplans" / "NOTES.md").write_text(
        "---\nid: NOT-A-WORKPLAN\ntype: note\n---\n\n# not a workplan\n", encoding="utf-8"
    )
    return root


def test_derives_workplans_tasks_and_archived_flag(tmp_path):
    p = fp.derive_from_checkout(_repo(tmp_path), "demo", "abc123")
    assert [w.record_id for w in p.workplans] == ["DEMO-WP-0001", "DEMO-WP-0002"]
    first, second = p.workplans
    assert first.status == "active" and first.archived is False
    assert second.archived is True
    assert p.task_count == 2
    assert [t.record_id for t in first.tasks] == ["DEMO-WP-0001-T01", "DEMO-WP-0001-T02"]


def test_ignores_files_that_are_not_workplans(tmp_path):
    """Selection is by `type: workplan`; anything else is not this hub's business."""
    p = fp.derive_from_checkout(_repo(tmp_path), "demo", "abc123")
    assert all(w.record_id != "NOT-A-WORKPLAN" for w in p.workplans)


def test_task_titles_fall_back_to_the_preceding_heading(tmp_path):
    p = fp.derive_from_checkout(_repo(tmp_path), "demo", "abc123")
    titles = [t.title for t in p.workplans[0].tasks]
    assert titles == ["Do the thing", "Do the other"]


def test_identifiers_are_derived_not_read_from_the_file(tmp_path):
    """A forge projection must not inherit whatever id a file happens to carry."""
    root = _repo(tmp_path)
    f = root / "workplans" / "DEMO-WP-0001-a.md"
    f.write_text(
        f.read_text(encoding="utf-8").replace(
            "status: active",
            'status: active\nstate_hub_workstream_id: "00000000-0000-4000-8000-000000000000"',
        ),
        encoding="utf-8",
    )
    p = fp.derive_from_checkout(root, "demo", "abc123")
    assert p.workplans[0].uuid == fp.derived_record_uuid("DEMO-WP-0001")
    assert p.workplans[0].uuid != "00000000-0000-4000-8000-000000000000"


def test_same_input_yields_identical_output(tmp_path):
    root = _repo(tmp_path)
    assert fp.derive_from_checkout(root, "demo", "abc").to_dict() == \
           fp.derive_from_checkout(root, "demo", "abc").to_dict()


def test_missing_workplans_directory_is_empty_not_an_error(tmp_path):
    (tmp_path / "bare").mkdir()
    p = fp.derive_from_checkout(tmp_path / "bare", "bare", "abc")
    assert p.workplans == [] and p.commit == "abc"


def test_clone_failure_is_reported_not_swallowed(monkeypatch):
    def boom(*a, **k):
        raise fp.ForgeDeriveError("repository not found")
    monkeypatch.setattr(fp, "_run_git", boom)
    with pytest.raises(fp.ForgeDeriveError, match="not found"):
        fp.derive_from_forge("nope")


class TestReset:
    """Applying the reset (STATE-WP-0083-T03).

    The properties worth guarding are the refusals, not the happy path. A reset
    that quietly retires a record someone still needs is worse than one that
    does nothing.
    """

    @staticmethod
    def _derived(*records):
        wps = []
        for rid, status, path in records:
            wps.append(
                fp.DerivedWorkplan(
                    record_id=rid, uuid=fp.derived_record_uuid(rid), title=rid,
                    status=status, relative_path=path, archived=False, tasks=[],
                )
            )
        return fp.DerivedProjection(repo_slug="demo", commit="c0ffee", workplans=wps)

    @pytest.mark.asyncio
    async def test_refuses_retirement_unless_acknowledged(self, monkeypatch):
        """A record that stops deriving may mean a deleted file — or a wrong branch."""
        session = _FakeSession(
            repo=_Repo(),
            rows=[_Row(slug="demo-wp-0001", status="active", path="workplans/a.md")],
        )
        out = await fp.reset_repository_projection(
            session, "demo", derived=self._derived()  # forge has nothing
        )
        assert out.status == "refused"
        assert out.refused and out.refused[0]["slug"] == "demo-wp-0001"
        assert out.retired == []
        assert session.committed is False

    @pytest.mark.asyncio
    async def test_retires_when_acknowledged(self):
        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
        session = _FakeSession(repo=_Repo(), rows=[row])
        out = await fp.reset_repository_projection(
            session, "demo", derived=self._derived(), acknowledge_retirements=True
        )
        assert out.status == "applied" and out.retired == ["demo-wp-0001"]
        assert row.projection_retired_at is not None
        assert row.projection_retired_reason == fp.RETIRE_REASON

    @pytest.mark.asyncio
    async def test_retirement_is_not_deletion(self):
        """Hub-native records reference workplans with RESTRICT; the row survives."""
        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
        session = _FakeSession(repo=_Repo(), rows=[row])
        await fp.reset_repository_projection(
            session, "demo", derived=self._derived(), acknowledge_retirements=True
        )
        assert row in session.rows
        assert session.deleted == []

    @pytest.mark.asyncio
    async def test_unretires_a_record_that_derives_again(self):
        from datetime import datetime, timezone
        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
        row.projection_retired_at = datetime.now(tz=timezone.utc)
        row.projection_retired_reason = fp.RETIRE_REASON
        session = _FakeSession(repo=_Repo(), rows=[row])
        out = await fp.reset_repository_projection(
            session, "demo",
            derived=self._derived(("DEMO-WP-0001", "active", "workplans/a.md")),
        )
        assert out.status == "applied"
        assert row.projection_retired_at is None

    @pytest.mark.asyncio
    async def test_unregistered_repository_is_refused_not_created(self):
        session = _FakeSession(repo=None, rows=[])
        out = await fp.reset_repository_projection(session, "nope", derived=self._derived())
        assert out.status == "refused"
        assert "not registered" in out.refused[0]["reason"]

    @pytest.mark.asyncio
    async def test_records_the_commit_it_derived_from(self):
        row = _Row(slug="demo-wp-0001", status="proposed", path="workplans/a.md")
        session = _FakeSession(repo=_Repo(), rows=[row])
        await fp.reset_repository_projection(
            session, "demo",
            derived=self._derived(("DEMO-WP-0001", "active", "workplans/a.md")),
        )
        assert row.derived_from_commit == "c0ffee"
        assert row.status == "active"


class _Repo:
    def __init__(self):
        import uuid as _u
        self.id = _u.uuid4()
        self.topic_id = None


class _Row:
    def __init__(self, slug, status, path):
        import uuid as _u
        self.id = _u.uuid4()
        self.slug = slug
        self.status = status
        self.backing_relative_path = path
        self.backing_filename = path.rsplit("/", 1)[-1]
        self.backing_archived = False
        self.projection_retired_at = None
        self.projection_retired_reason = None
        self.derived_from_commit = None


class _FakeSession:
    """Stands in for AsyncSession: enough to prove intent without a database."""

    def __init__(self, repo, rows, foreign=None, slug_clash=None):
        self._repo = repo
        self.rows = list(rows)
        self._foreign = list(foreign or [])
        self._slug_clash = list(slug_clash or [])
        self.added = []
        self.deleted = []
        self.committed = False
        self._calls = 0

    async def execute(self, *_a, **_k):
        self._calls += 1
        repo, rows = self._repo, self.rows
        # 1st call resolves the repo, 2nd loads its workplans, 3rd is the
        # foreign-identifier lookup.
        # 1 resolves the repo, 2 loads its workplans, 3 is the identifier
        # lookup, 4 the slug lookup.
        if self._calls == 2:
            payload = rows
        elif self._calls == 3:
            payload = self._foreign
        elif self._calls == 4:
            payload = self._slug_clash
        else:
            payload = []

        class R:
            def scalar_one_or_none(self_inner):
                return repo

            def scalars(self_inner):
                return iter(payload)

        return R()

    def add(self, obj):
        self.added.append(obj)

    def delete(self, obj):
        self.deleted.append(obj)

    async def flush(self):
        return None

    async def commit(self):
        self.committed = True

    async def rollback(self):
        self.rolled_back = True


class TestCollisionRefusal:
    """A colliding identifier must refuse, not raise (STATE-WP-0083-T03).

    net-kingdom's ADHOC-2026-08-23 derives to an identifier another repository
    already holds — the case CUST-WP-0066 documents. The reset previously failed
    on a database constraint, which tells the caller nothing they can act on.
    """

    @staticmethod
    def _derived(rid="DEMO-WP-0001"):
        return fp.DerivedProjection(
            repo_slug="demo", commit="c0ffee",
            workplans=[fp.DerivedWorkplan(
                record_id=rid, uuid=fp.derived_record_uuid(rid), title=rid,
                status="active", relative_path="workplans/a.md",
                archived=False, tasks=[])],
        )

    @pytest.mark.asyncio
    async def test_refuses_when_the_identifier_belongs_elsewhere(self):
        derived = self._derived()
        foreign = _Row(slug="held-by-someone-else", status="finished", path="workplans/x.md")
        foreign.id = __import__("uuid").UUID(derived.workplans[0].uuid)
        session = _FakeSession(repo=_Repo(), rows=[], foreign=[foreign])
        out = await fp.reset_repository_projection(session, "demo", derived=derived)
        assert out.status == "refused"
        assert out.refused[0]["reason"].startswith("derived identifier already belongs")
        assert out.refused[0]["held_by_slug"] == "held-by-someone-else"
        assert out.created == [] and session.added == []

    @pytest.mark.asyncio
    async def test_acknowledging_retirements_does_not_authorise_a_collision(self):
        """Different decision, different authorisation."""
        derived = self._derived()
        foreign = _Row(slug="held-by-someone-else", status="finished", path="workplans/x.md")
        foreign.id = __import__("uuid").UUID(derived.workplans[0].uuid)
        session = _FakeSession(repo=_Repo(), rows=[], foreign=[foreign])
        out = await fp.reset_repository_projection(
            session, "demo", derived=derived, acknowledge_retirements=True
        )
        assert out.status == "refused"
        assert session.added == []

    @pytest.mark.asyncio
    async def test_no_collision_still_creates(self):
        derived = self._derived()
        session = _FakeSession(repo=_Repo(), rows=[], foreign=[])
        out = await fp.reset_repository_projection(session, "demo", derived=derived)
        assert out.status == "applied" and out.created == ["DEMO-WP-0001"]


class TestFleetReset:
    """The fleet form must be a loop over the repository form (T04).

    Its value is entirely in what it does with failure: a wide reset that stops
    at the first refusal is one nobody can run, because there is always one
    unresolved repository somewhere.
    """

    class _Factory:
        def __init__(self, sessions):
            self._sessions = list(sessions)

        def __call__(self):
            session = self._sessions.pop(0)
            class Ctx:
                async def __aenter__(self_inner):
                    return session
                async def __aexit__(self_inner, *a):
                    return False
            return Ctx()

    @staticmethod
    def _derived(rid):
        return fp.DerivedProjection(
            repo_slug="x", commit="c0ffee",
            workplans=[fp.DerivedWorkplan(
                record_id=rid, uuid=fp.derived_record_uuid(rid), title=rid,
                status="active", relative_path="workplans/a.md", archived=False, tasks=[])])

    @pytest.mark.asyncio
    async def test_a_refusal_does_not_stop_the_pass(self, monkeypatch):
        calls = []

        async def fake(session, slug, **kw):
            calls.append(slug)
            out = fp.ResetOutcome(repo_slug=slug, commit="c0ffee", status="applied")
            if slug == "bad":
                out.status = "refused"
                out.refused.append({"reason": "would be retired", "slug": "x"})
            else:
                out.updated.append("A-WP-0001")
            return out

        monkeypatch.setattr(fp, "reset_repository_projection", fake)
        s = [_FakeSession(repo=_Repo(), rows=[]) for _ in range(3)]
        res = await fp.reset_fleet_projection(self._Factory(s), ["good", "bad", "also-good"])
        assert calls == ["good", "bad", "also-good"]
        d = res.to_dict()
        assert d["by_status"] == {"applied": 2, "refused": 1}
        assert d["totals"]["updated"] == 2

    @pytest.mark.asyncio
    async def test_an_error_does_not_stop_the_pass(self, monkeypatch):
        async def fake(session, slug, **kw):
            if slug == "boom":
                raise RuntimeError("clone failed")
            return fp.ResetOutcome(repo_slug=slug, commit="c", status="noop")

        monkeypatch.setattr(fp, "reset_repository_projection", fake)
        s = [_FakeSession(repo=_Repo(), rows=[]) for _ in range(3)]
        res = await fp.reset_fleet_projection(self._Factory(s), ["a", "boom", "b"])
        d = res.to_dict()
        assert d["errored"] == 1 and "clone failed" in res.errors["boom"]
        assert set(res.results) == {"a", "b"}

    @pytest.mark.asyncio
    async def test_only_applied_repositories_are_committed(self, monkeypatch):
        async def fake(session, slug, **kw):
            out = fp.ResetOutcome(repo_slug=slug, commit="c", status="applied")
            if slug == "refuser":
                out.status = "refused"
            else:
                out.updated.append("A-WP-0001")
            return out

        monkeypatch.setattr(fp, "reset_repository_projection", fake)
        s = [_FakeSession(repo=_Repo(), rows=[]) for _ in range(2)]
        await fp.reset_fleet_projection(self._Factory(s), ["applier", "refuser"])
        assert s[0].committed is True
        assert s[1].committed is False


class TestSlugCollisionRefusal:
    """slug carries its own unique constraint (STATE-WP-0083-T04).

    Checking the identifier alone left disaster-control raising IntegrityError:
    two repositories can derive different identifiers whose slugs still collide.
    """

    @pytest.mark.asyncio
    async def test_refuses_when_the_slug_belongs_elsewhere(self):
        derived = fp.DerivedProjection(
            repo_slug="demo", commit="c0ffee",
            workplans=[fp.DerivedWorkplan(
                record_id="REPO-WP-0001", uuid=fp.derived_record_uuid("REPO-WP-0001"),
                title="x", status="active", relative_path="workplans/a.md",
                archived=False, tasks=[])])
        clash = _Row(slug="repo-wp-0001", status="finished", path="workplans/other.md")
        session = _FakeSession(repo=_Repo(), rows=[], foreign=[], slug_clash=[clash])
        out = await fp.reset_repository_projection(session, "demo", derived=derived)
        assert out.status == "refused"
        assert out.refused[0]["reason"].startswith("slug already belongs")
        assert session.added == []


class TestUnreadableIsNotMissing:
    """STATE-WP-0084-T01.

    A repository central is not permitted to read and a repository whose
    records no longer derive authorise opposite things. Every test here exists
    to keep the retirement path unreachable from an answer that cannot support
    it — by construction, not by the reset happening to fail first.
    """

    @pytest.mark.parametrize(
        "stderr",
        [
            "fatal: could not read Username for 'https://forgejo.coulomb.social'",
            "remote: Invalid username or password.\nfatal: Authentication failed",
            "fatal: could not read Username for 'https://f': terminal prompts disabled",
            "fatal: repository 'https://forgejo.coulomb.social/rapp-openbao.git' not found",
            "fatal: unable to access '...': The requested URL returned error: 403",
        ],
    )
    def test_permission_shaped_failures_are_classified_unreadable(self, stderr, monkeypatch):
        def boom(*a, **k):
            raise fp.ForgeDeriveError(stderr)

        monkeypatch.setattr(fp, "_run_git", boom)
        with pytest.raises(fp.ForgeUnreadableError):
            fp.derive_from_forge("rapp-openbao")

    def test_a_genuine_fault_stays_a_plain_error(self, monkeypatch):
        def boom(*a, **k):
            raise fp.ForgeDeriveError("fatal: early EOF\nfatal: index-pack failed")

        monkeypatch.setattr(fp, "_run_git", boom)
        with pytest.raises(fp.ForgeDeriveError) as exc:
            fp.derive_from_forge("demo")
        assert not isinstance(exc.value, fp.ForgeUnreadableError)

    def test_unreadable_is_a_derive_error_so_old_callers_still_catch_it(self):
        assert issubclass(fp.ForgeUnreadableError, fp.ForgeDeriveError)

    def test_a_checkout_without_workplans_cannot_evidence_absence(self, tmp_path):
        (tmp_path / "bare").mkdir()
        p = fp.derive_from_checkout(tmp_path / "bare", "bare", "abc")
        assert p.workplans == []
        assert p.records_source_present is False
        assert p.retirement_eligible is False

    def test_a_real_checkout_can(self, tmp_path):
        p = fp.derive_from_checkout(_repo(tmp_path), "demo", "abc")
        assert p.records_source_present is True
        assert p.retirement_eligible is True

    def test_the_diff_withholds_stale_rather_than_computing_it(self):
        derived = fp.DerivedProjection(
            repo_slug="demo", commit="c0ffee", records_source_present=False
        )
        hub = [{"id": "11111111-1111-1111-1111-111111111111", "slug": "demo-wp-0001",
                "status": "active", "backing_relative_path": "workplans/a.md"}]
        d = fp.diff_against_hub(derived, hub, {})
        assert d.stale == []
        assert d.would_remove == 0
        assert d.stale_withheld

    @pytest.mark.asyncio
    async def test_an_empty_source_cannot_retire_even_when_acknowledged(self):
        derived = fp.DerivedProjection(
            repo_slug="demo", commit="c0ffee", records_source_present=False
        )
        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
        session = _FakeSession(repo=_Repo(), rows=[row])
        out = await fp.reset_repository_projection(
            session, "demo", derived=derived, acknowledge_retirements=True
        )
        assert out.status == "refused"
        assert out.retired == []
        assert row.projection_retired_at is None
        assert out.refused[0]["reason"].startswith("source produced no records")

    @pytest.mark.asyncio
    async def test_an_unreadable_repository_reports_unreadable_not_error(self, monkeypatch):
        def boom(*a, **k):
            raise fp.ForgeUnreadableError("rapp-openbao could not be read from the forge")

        monkeypatch.setattr(fp, "derive_from_forge", boom)
        session = _FakeSession(repo=_Repo(), rows=[])
        out = await fp.reset_repository_projection(session, "rapp-openbao")
        assert out.status == "unreadable"
        assert out.retired == [] and out.created == [] and out.updated == []

    @pytest.mark.asyncio
    async def test_the_fleet_keeps_unreadable_out_of_the_error_bucket(self, monkeypatch):
        async def fake(session, slug, **kw):
            if slug == "private":
                out = fp.ResetOutcome(repo_slug=slug, commit="", status="unreadable")
                out.refused.append({"reason": "repository could not be read from the forge",
                                    "slug": slug, "detail": "could not read Username"})
                return out
            if slug == "broken":
                raise RuntimeError("index-pack failed")
            out = fp.ResetOutcome(repo_slug=slug, commit="c0ffee", status="applied")
            out.updated.append("A-WP-0001")
            return out

        monkeypatch.setattr(fp, "reset_repository_projection", fake)
        sessions = [_FakeSession(repo=_Repo(), rows=[]) for _ in range(3)]
        outcome = await fp.reset_fleet_projection(
            TestFleetReset._Factory(sessions), ["ok", "private", "broken"]
        )
        assert list(outcome.unreadable) == ["private"]
        assert list(outcome.errors) == ["broken"]
        assert "private" not in outcome.results
        d = outcome.to_dict()
        assert d["unreadable_count"] == 1 and d["errored"] == 1
        assert d["repositories"] == 3