"""Integration: the admin ``/api/source-labels`` surface (phase 191, task 02).

The wire half of "Sources need a name, a label, tag(s)" (``TODO.md`` L4)
— the catalogue the Sources page loads, the upsert its Save drives and
the reset its Reset button drives. Driven through the real app with the
house ``TestClient`` cookie jar (the ``test_ui_settings_api.py`` phase-91
fixture pattern: the admin login and the token-user matrix), against the
real Postgres (``podman compose up -d db``); no LLM involved.

Contract under test:

* **the gate** — an anonymous caller AND a signed-in token USER (role
  ``"user"``) get 403 ``{"detail": "admin only"}`` on all three routes
  (the router-wide ``require_admin``, the phase-91/168 admin-surface
  shape, one fixed detail with nothing to enumerate), and nothing lands;
* **the catalogue GET** — the union of the ``git_sources`` rows of BOTH
  kinds and the indexed document sources no row derives (the uploads),
  de-duplicated and sorted by ``source``, each entry exactly
  ``{source, display_name, label, tags}`` with its label row or the
  ``null``/``[]`` defaults; an env-only deployment (empty table) lists
  the ``BOR_GIT_SOURCES`` names; three tables, three SELECTs — the
  assembly never queries per source;
* **the PUT** — the round trip (set all three → 200 → the GET reads them
  back), the partial-update shape (an omitted field is untouched, an
  explicit ``null`` clears, ``[]`` clears the tags, an all-omitted body
  is a 200 no-op), the trim-then-cap ordering, the case-insensitive tag
  de-dupe preserving the first case, the framework's standard 422 for a
  121-char name / 201-char label / 11 tags / 33-char tag / whitespace-only
  tag / a key outside the shape — each leaving a pre-seeded row
  BYTE-IDENTICAL — the 404 for a source that is neither a row name nor an
  indexed document (nothing created), the 422 for a > 120-char path, and
  the ``normalize_tags`` funnel (its ``ValueError`` is a 422, never a
  500);
* **the DELETE** — 200 with the entry's post-reset state, the GET shows
  the nulls after it, a second DELETE is 404;
* **the alias contract (D1)** — labelling a source touches
  ``source_labels`` and NOTHING else: ``documents.source``, the row's
  derived name and the catalogue's identity are byte-identical after a
  rename, which is what keeps a saved chat's ``source/path`` locator, a
  shared page and a verbatim caption valid across a rename.

``source_labels``, ``git_sources`` and the catalogue are global state:
truncated around every test.

Requires:  podman compose up -d db
"""
from __future__ import annotations

import uuid
from collections.abc import Callable, Iterator
from contextlib import contextmanager
from typing import Any

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import event, text
from sqlalchemy.orm import Session

from app.api import source_labels as source_labels_api
from app.config import Settings
from app.db import engine
from app.main import app as fastapi_app
from app.models import Document, GitSource
from tests.conftest import ADMIN_PASSWORD

#: The catalogue entry's exact wire shape (``extra="forbid"`` makes the
#: omission structural — the row's ``updated_at`` never rides along).
_ENTRY_KEYS = {"source", "display_name", "label", "tags"}


@pytest.fixture(autouse=True)
def clean_state(db: Session) -> Iterator[None]:
    """Every table this module reads or writes is global state (the
    catalogue is a FULL scan of the registry and the documents, so
    another suite's rows would move the pins)."""
    _wipe(db)
    yield
    _wipe(db)


def _wipe(db: Session) -> None:
    db.execute(text("DELETE FROM source_labels"))
    db.execute(text("TRUNCATE git_sources"))
    db.execute(text("TRUNCATE chunks, documents"))
    db.execute(text("TRUNCATE api_tokens"))
    db.commit()


def _login(client: TestClient) -> TestClient:
    client.post("/api/login", json={"password": ADMIN_PASSWORD})
    return client


@pytest.fixture()
def env_sources(
    monkeypatch: pytest.MonkeyPatch,
) -> Callable[[str], None]:
    """Pin the ``BOR_GIT_SOURCES`` fallback the way
    ``test_git_sources_api.py`` does — with a fresh ``Settings`` that
    ignores the operator's ``.env`` (a dev ``.env`` with a repo in it
    would otherwise BE the catalogue while the table is empty, the
    phase-35 fallback). Returns the setter so one test can name the list
    it wants; the empty string is the "nothing configured" state."""

    def _set(git_sources: str = "") -> None:
        monkeypatch.setattr(
            "app.rag.git_sources.get_settings",
            lambda: Settings(
                _env_file=None,  # pyright: ignore[reportCallIssue]
                git_sources=git_sources,
            ),
        )

    return _set


def _doc(db: Session, source: str, path: str) -> Document:
    """One indexed document — the catalogue's (b) arm reads
    ``documents.source`` and NOTHING else."""
    doc = Document(
        id=uuid.uuid4(),
        source=source,
        path=path,
        full_path=f"/srv/{source}/{path}",
        title=path.rsplit("/", 1)[-1],
        content=f"body of {path}",
        content_hash="0" * 64,
    )
    db.add(doc)
    return doc


def _seed_three_kinds(db: Session) -> None:
    """One git row, one local row and one upload-shaped document.

    The git row is a ``file://`` URL (the phase-135 E2E-fixture form) and
    carries the ``.git`` suffix, so the pin also proves the name is the
    importer's DERIVED one (``repo_name``) rather than the raw URL; a
    document exists under that derived name too, which is what the
    de-duplication has to collapse.
    """
    db.add_all(
        [
            GitSource(url="file:///srv/repos/gitea-forge.git", kind="git"),
            GitSource(
                url="/srv/docs/media-vault",
                kind="local",
                path="/srv/docs/media-vault",
            ),
        ]
    )
    db.add(_doc(db, "gitea-forge", "docs/app.md"))
    db.add(_doc(db, "corpus-upload", "notes/a.md"))
    db.commit()


@contextmanager
def _captured_statements() -> Iterator[list[str]]:
    """Every SQL statement the ENGINE runs inside the block.

    The house statement-capture idiom (``test_import_session_retention``
    / ``test_prune_cascade_statements``): a local listener on the shared
    engine, removed in ``finally`` — the request runs on another thread
    under ``TestClient``, so the engine (not the session) is the place
    both sides see.
    """
    captured: list[str] = []

    def _listen(
        conn, cursor, statement, parameters, context, executemany
    ) -> None:  # the engine's own signature — no types to restate
        captured.append(statement)

    event.listen(engine, "before_cursor_execute", _listen)
    try:
        yield captured
    finally:
        event.remove(engine, "before_cursor_execute", _listen)


def _selects(statements: list[str]) -> list[str]:
    """The SELECTs only, whitespace-collapsed for comparability, with
    the pool's ``SELECT 1`` health check dropped — it is a checkout
    artifact, not a read this endpoint asked for, and pinning it would
    make the count depend on how warm the pool happens to be."""
    return [
        collapsed
        for statement in statements
        if (collapsed := " ".join(statement.split())).upper().startswith("SELECT")
        and collapsed.upper() != "SELECT 1"
    ]


def _row(db: Session, source: str) -> dict[str, Any]:
    """One ``source_labels`` row read RAW — the pin never trusts the
    shape under test to report its own storage."""
    db.expire_all()
    found = db.execute(
        text(
            "SELECT source, display_name, label, tags FROM source_labels"
            " WHERE source = :s"
        ),
        {"s": source},
    ).first()
    if found is None:
        return {}
    return {
        "source": found[0],
        "display_name": found[1],
        "label": found[2],
        "tags": found[3],
    }


# ---------------------------------------------------------------------------
# The admin gate (the phase-91 shape)
# ---------------------------------------------------------------------------


def test_anonymous_is_403_on_every_route(client: TestClient, db: Session) -> None:
    """Router-wide ``require_admin``: one fixed detail on all three
    routes, and nothing landed (the ``test_ui_settings_api`` contract)."""
    r = client.get("/api/source-labels")
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}

    r = client.put("/api/source-labels/gitea-forge", json={"display_name": "nope"})
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}

    r = client.delete("/api/source-labels/gitea-forge")
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}

    assert db.execute(text("SELECT count(*) FROM source_labels")).scalar_one() == 0


def test_token_user_is_403_on_every_route(client: TestClient) -> None:
    """A signed-in token USER is not the admin: the owner's naming surface
    is closed to them on all three methods (the phase-79 token matrix —
    the same leg the phase-91/168 admin surfaces pin)."""
    admin = _login(TestClient(fastapi_app))
    r = admin.post("/api/tokens", json={"label": "alice"})
    assert r.status_code == 201, r.text
    token = r.json()["token"]
    assert client.post("/api/token-auth", json={"token": token}).status_code == 204
    assert client.get("/api/whoami").json() == {"authenticated": True, "role": "user"}

    assert client.get("/api/source-labels").status_code == 403
    r = client.put("/api/source-labels/gitea-forge", json={"display_name": "nope"})
    assert r.status_code == 403
    assert r.json() == {"detail": "admin only"}
    assert client.delete("/api/source-labels/gitea-forge").status_code == 403


# ---------------------------------------------------------------------------
# GET — the catalogue
# ---------------------------------------------------------------------------


def test_get_empty_deployment_is_an_empty_catalogue(
    admin_client: TestClient, env_sources: Callable[[str], None]
) -> None:
    """No rows and no env list: the response is the empty list, not a 404
    and not a missing key — the page always has a shape to render."""
    env_sources("")
    r = admin_client.get("/api/source-labels")
    assert r.status_code == 200, r.text
    assert r.json() == {"labels": []}


def test_get_lists_git_local_and_upload_sources(
    admin_client: TestClient, db: Session
) -> None:
    """The D5 union, sorted, de-duplicated, with the exact entry shape.

    ``gitea-forge`` (a git row, derived from the URL basename with the
    ``.git`` stripped) also has a document under that name — the entry
    still appears exactly ONCE. ``corpus-upload`` has documents and no
    row, so only arm (b) can list it (and the E2E needs it to).
    """
    _seed_three_kinds(db)

    r = admin_client.get("/api/source-labels")
    assert r.status_code == 200, r.text
    labels = r.json()["labels"]
    assert [entry["source"] for entry in labels] == [
        "corpus-upload",
        "gitea-forge",
        "media-vault",
    ]
    for entry in labels:
        assert set(entry) == _ENTRY_KEYS
        assert entry["display_name"] is None
        assert entry["label"] is None
        assert entry["tags"] == []


def test_get_carries_the_stored_label_on_one_entry_only(
    admin_client: TestClient, db: Session
) -> None:
    """One labelled row, two unlabelled ones: the label rides its own
    entry and the others keep the null/``[]`` defaults (a label row is
    data beside the name, never a rename — D1)."""
    _seed_three_kinds(db)
    assert (
        admin_client.put(
            "/api/source-labels/corpus-upload",
            json={"display_name": "the uploads", "tags": ["upload"]},
        ).status_code
        == 200
    )

    labels = admin_client.get("/api/source-labels").json()["labels"]
    by_source = {entry["source"]: entry for entry in labels}
    assert by_source["corpus-upload"] == {
        "source": "corpus-upload",
        "display_name": "the uploads",
        "label": None,
        "tags": ["upload"],
    }
    assert by_source["gitea-forge"]["display_name"] is None
    assert by_source["media-vault"]["tags"] == []


def test_get_lists_env_source_names_while_the_table_is_empty(
    admin_client: TestClient, env_sources: Callable[[str], None]
) -> None:
    """An env-only deployment (no ``git_sources`` rows): the
    ``BOR_GIT_SOURCES`` names are the catalogue, derived exactly as the
    importer would name what it clones — and one of them is labelable."""
    env_sources("https://a.example.com/one.git, git@b.example.com:two.git ,")

    labels = admin_client.get("/api/source-labels").json()["labels"]
    assert [entry["source"] for entry in labels] == ["one", "two"]

    r = admin_client.put("/api/source-labels/one", json={"label": "the first repo"})
    assert r.status_code == 200, r.text
    assert r.json()["label"] == "the first repo"


def test_get_reads_three_tables_and_never_one_per_source(
    admin_client: TestClient, db: Session
) -> None:
    """The one-query-per-table discipline: with four sources and one
    label row the GET costs exactly three SELECTs — the registry, the
    distinct document sources, the label rows. A per-source label lookup
    (or a third existence query) fails this pin."""
    _seed_three_kinds(db)
    db.add(_doc(db, "third-source", "b.md"))
    db.commit()
    admin_client.put("/api/source-labels/media-vault", json={"display_name": "vault"})

    with _captured_statements() as statements:
        body = admin_client.get("/api/source-labels").json()
        selects = _selects(statements)

    assert [entry["source"] for entry in body["labels"]] == [
        "corpus-upload",
        "gitea-forge",
        "media-vault",
        "third-source",
    ]
    assert len(selects) == 3, selects
    assert sum(1 for s in selects if "FROM git_sources" in s) == 1
    assert sum(1 for s in selects if "DISTINCT documents.source" in s) == 1
    assert sum(1 for s in selects if "FROM source_labels" in s) == 1
    # No whole-entity document read — the catalogue never hydrates content.
    assert not any("documents.content" in s for s in selects), selects


# ---------------------------------------------------------------------------
# PUT — the upsert
# ---------------------------------------------------------------------------


def test_put_round_trips_all_three_fields(
    admin_client: TestClient, db: Session
) -> None:
    """Set name + label + tags on a git source: 200 carries the stored
    shape, the row exists, and the GET reads it back."""
    _seed_three_kinds(db)

    r = admin_client.put(
        "/api/source-labels/gitea-forge",
        json={
            "display_name": "the git server",
            "label": "self-hosted forge + Gitea",
            "tags": ["gitea", "forge"],
        },
    )
    assert r.status_code == 200, r.text
    assert r.json() == {
        "source": "gitea-forge",
        "display_name": "the git server",
        "label": "self-hosted forge + Gitea",
        "tags": ["gitea", "forge"],
    }
    assert _row(db, "gitea-forge") == {
        "source": "gitea-forge",
        "display_name": "the git server",
        "label": "self-hosted forge + Gitea",
        "tags": ["gitea", "forge"],
    }
    entry = admin_client.get("/api/source-labels").json()["labels"]
    assert entry[1]["display_name"] == "the git server"


def test_put_partial_update_shape(admin_client: TestClient, db: Session) -> None:
    """D5 three ways: a value SETS, an OMITTED field is untouched, an
    explicit ``null`` CLEARS (and ``tags: []`` clears the tags while the
    strings stay)."""
    _seed_three_kinds(db)
    admin_client.put(
        "/api/source-labels/media-vault",
        json={"display_name": "the media box", "label": "jellyfin + arrs", "tags": ["jellyfin"]},
    )

    # Only the label moves — the name and the tags keep their values.
    r = admin_client.put("/api/source-labels/media-vault", json={"label": "arrs only"})
    assert r.status_code == 200, r.text
    assert r.json()["display_name"] == "the media box"
    assert r.json()["label"] == "arrs only"
    assert r.json()["tags"] == ["jellyfin"]

    # An explicit null clears JUST that field.
    r = admin_client.put("/api/source-labels/media-vault", json={"display_name": None})
    assert r.status_code == 200, r.text
    assert r.json() == {
        "source": "media-vault",
        "display_name": None,
        "label": "arrs only",
        "tags": ["jellyfin"],
    }

    # An empty tag list clears the tags; the label stays.
    r = admin_client.put("/api/source-labels/media-vault", json={"tags": []})
    assert r.json() == {
        "source": "media-vault",
        "display_name": None,
        "label": "arrs only",
        "tags": [],
    }
    assert _row(db, "media-vault")["tags"] == []


def test_put_all_omitted_body_is_a_200_no_op(
    admin_client: TestClient, db: Session
) -> None:
    """The ``GitSourcePatchIn`` "all absent → 200 no-op" precedent: an
    empty body changes nothing on a labelled row (and creates a defaults
    row on an unlabelled one)."""
    _seed_three_kinds(db)
    admin_client.put(
        "/api/source-labels/gitea-forge",
        json={"display_name": "the git server", "tags": ["gitea"]},
    )

    r = admin_client.put("/api/source-labels/gitea-forge", json={})
    assert r.status_code == 200, r.text
    assert r.json() == {
        "source": "gitea-forge",
        "display_name": "the git server",
        "label": None,
        "tags": ["gitea"],
    }
    assert _row(db, "gitea-forge")["display_name"] == "the git server"


def test_put_trims_values_and_dedupes_tags(
    admin_client: TestClient, db: Session
) -> None:
    """Trim before the caps (``GitSourceIn._trim_url``), and the ONE tag
    rule: case-insensitive de-dupe with the FIRST case kept, order
    otherwise preserved."""
    _seed_three_kinds(db)

    r = admin_client.put(
        "/api/source-labels/gitea-forge",
        json={
            "display_name": "   the git server   ",
            "label": "  self-hosted forge  ",
            "tags": [" Gitea ", "gitea", "FORGE", "forge"],
        },
    )
    assert r.status_code == 200, r.text
    assert r.json()["display_name"] == "the git server"
    assert r.json()["label"] == "self-hosted forge"
    assert r.json()["tags"] == ["Gitea", "FORGE"]


@pytest.mark.parametrize(
    ("field", "value", "needle"),
    [
        ("display_name", "x" * 121, "string_too_long"),
        ("label", "x" * 201, "string_too_long"),
        ("tags", ["tag"] * 11, "too_long"),
        ("tags", ["x" * 33], "string_too_long"),
        ("tags", ["   "], "string_too_short"),
        ("tags", [""], "string_too_short"),
    ],
)
def test_put_validation_422s_leave_the_row_unchanged(
    admin_client: TestClient,
    db: Session,
    field: str,
    value: Any,
    needle: str,
) -> None:
    """Every D5 cap is the framework's standard 422 (the pydantic body,
    not a fixed ``detail``), names the offending field, and writes
    NOTHING: the pre-seeded row is byte-identical afterwards.

    The trim runs BEFORE the cap, so a 121-char value that trims to 120
    is legal — the 120-char leg of each boundary is asserted too, or the
    cap could silently be one short."""
    _seed_three_kinds(db)
    admin_client.put(
        "/api/source-labels/gitea-forge",
        json={"display_name": "the git server", "label": "forge", "tags": ["gitea"]},
    )
    before = _row(db, "gitea-forge")

    if field == "display_name":
        legal = "y" * 120
    elif field == "label":
        legal = "y" * 200
    else:
        legal = ["tag"] * 10 if len(value) == 11 else ["y" * 32]
    r = admin_client.put("/api/source-labels/gitea-forge", json={field: value})
    assert r.status_code == 422, r.text
    detail = r.json()["detail"]
    assert isinstance(detail, list), r.text
    assert detail[0]["type"] == needle, detail
    assert field in detail[0]["loc"], detail
    assert _row(db, "gitea-forge") == before

    assert admin_client.put(
        "/api/source-labels/gitea-forge", json={field: legal}
    ).status_code == 200


def test_put_rejects_a_key_outside_the_shape(
    admin_client: TestClient, db: Session
) -> None:
    """``extra="forbid"`` (the phase-91 voice): an undeclared key is a
    422 and no row is created."""
    _seed_three_kinds(db)

    r = admin_client.put(
        "/api/source-labels/media-vault",
        json={"display_name": "vault", "source": "gitea-forge"},
    )
    assert r.status_code == 422, r.text
    assert r.json()["detail"][0]["type"] == "extra_forbidden"
    assert db.execute(text("SELECT count(*) FROM source_labels")).scalar_one() == 0


def test_put_rejects_a_malformed_tags_value(
    admin_client: TestClient, db: Session
) -> None:
    """A ``tags`` value that is not a list of strings is pydantic's own
    422, in its own voice (``list_type`` / ``string_type``): the trimmer
    passes anything it does not understand straight through rather than
    guessing, so the framework names the type and NOTHING is written."""
    _seed_three_kinds(db)

    r = admin_client.put("/api/source-labels/media-vault", json={"tags": "gitea"})
    assert r.status_code == 422, r.text
    assert r.json()["detail"][0]["type"] == "list_type"

    r = admin_client.put(
        "/api/source-labels/media-vault", json={"tags": ["gitea", 7]}
    )
    assert r.status_code == 422, r.text
    assert r.json()["detail"][0]["type"] == "string_type"
    assert db.execute(text("SELECT count(*) FROM source_labels")).scalar_one() == 0


def test_put_folds_empty_strings_to_null(admin_client: TestClient, db: Session) -> None:
    """An empty / whitespace-only ``display_name`` or ``label`` is the
    CLEAR leg, not a stored empty string: "no name" and "an empty name"
    must not be two different things for the prompt builders (task 05)."""
    _seed_three_kinds(db)
    admin_client.put(
        "/api/source-labels/corpus-upload",
        json={"display_name": "the uploads", "label": "the upload folder"},
    )

    r = admin_client.put(
        "/api/source-labels/corpus-upload",
        json={"display_name": "   ", "label": ""},
    )
    assert r.status_code == 200, r.text
    assert r.json()["display_name"] is None
    assert r.json()["label"] is None
    assert _row(db, "corpus-upload")["display_name"] is None


def test_put_404s_a_source_that_exists_nowhere(
    admin_client: TestClient, db: Session
) -> None:
    """D5's no-orphan-row arm: neither a registry name nor an indexed
    document → 404, and NOTHING is written (a dead label would boost
    nothing and never appear in the UI to be cleared)."""
    _seed_three_kinds(db)

    r = admin_client.put("/api/source-labels/no-such-source", json={"label": "x"})
    assert r.status_code == 404, r.text
    assert r.json() == {"detail": "source not found"}
    assert db.execute(text("SELECT count(*) FROM source_labels")).scalar_one() == 0

    # A near-miss on the derived name is still unknown (no fuzzy leg).
    r = admin_client.put("/api/source-labels/gitea-forg", json={"label": "x"})
    assert r.status_code == 404, r.text
    assert db.execute(text("SELECT count(*) FROM source_labels")).scalar_one() == 0


def test_put_422s_an_overlong_source_path(admin_client: TestClient) -> None:
    """The path param carries the ``documents.source`` 120 cap: a longer
    name is the framework's standard 422 on the ``path`` location (it
    never reaches the query)."""
    r = admin_client.put(f"/api/source-labels/{'x' * 121}", json={"label": "x"})
    assert r.status_code == 422, r.text
    detail = r.json()["detail"]
    assert detail[0]["type"] == "string_too_long"
    assert detail[0]["loc"] == ["path", "source"]


def test_put_maps_the_tag_normalizer_error_to_422(
    admin_client: TestClient,
    db: Session,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The belt: ``normalize_tags`` is the ONE tag rule and its
    ``ValueError`` is the framework's 422 — never a 500. The request
    shape caps the same numbers today, so the belt is pinned by
    substituting the raising normalizer (the row stays byte-identical,
    which is the part that must not regress)."""
    _seed_three_kinds(db)
    admin_client.put(
        "/api/source-labels/gitea-forge", json={"display_name": "the git server"}
    )
    before = _row(db, "gitea-forge")

    def _boom(_raw: list[str]) -> list[str]:
        raise ValueError("at most 10 tags are allowed")

    monkeypatch.setattr(source_labels_api, "normalize_tags", _boom)

    r = admin_client.put("/api/source-labels/gitea-forge", json={"tags": ["gitea"]})
    assert r.status_code == 422, r.text
    assert r.json() == {"detail": "at most 10 tags are allowed"}
    assert _row(db, "gitea-forge") == before


# ---------------------------------------------------------------------------
# DELETE — the reset
# ---------------------------------------------------------------------------


def test_delete_resets_then_404s(admin_client: TestClient, db: Session) -> None:
    """200 with the entry's post-reset state (the phase-168
    ``DELETE /api/admin/settings/{key}`` convention — the row re-renders
    from the response it already has), the GET shows the nulls, and a
    second DELETE is a 404 with one fixed detail."""
    _seed_three_kinds(db)
    admin_client.put(
        "/api/source-labels/gitea-forge",
        json={"display_name": "the git server", "label": "forge", "tags": ["gitea"]},
    )

    r = admin_client.delete("/api/source-labels/gitea-forge")
    assert r.status_code == 200, r.text
    assert r.json() == {
        "source": "gitea-forge",
        "display_name": None,
        "label": None,
        "tags": [],
    }
    assert db.execute(text("SELECT count(*) FROM source_labels")).scalar_one() == 0

    entry = {e["source"]: e for e in admin_client.get("/api/source-labels").json()["labels"]}
    assert entry["gitea-forge"] == {
        "source": "gitea-forge",
        "display_name": None,
        "label": None,
        "tags": [],
    }

    r = admin_client.delete("/api/source-labels/gitea-forge")
    assert r.status_code == 404, r.text
    assert r.json() == {"detail": "source label not found"}


def test_delete_works_for_a_source_that_has_vanished(
    admin_client: TestClient, db: Session
) -> None:
    """The reset never consults the registry: a label whose source has
    since been removed (phase 69's total removal) stays CLEARABLE, or a
    label could never be taken back off a gone source."""
    _seed_three_kinds(db)
    admin_client.put("/api/source-labels/corpus-upload", json={"label": "the upload"})
    db.execute(text("DELETE FROM documents WHERE source = 'corpus-upload'"))
    db.commit()

    r = admin_client.delete("/api/source-labels/corpus-upload")
    assert r.status_code == 200, r.text
    assert db.execute(text("SELECT count(*) FROM source_labels")).scalar_one() == 0


# ---------------------------------------------------------------------------
# The alias contract (D1)
# ---------------------------------------------------------------------------


def test_labelling_a_source_touches_nothing_but_the_label_row(
    admin_client: TestClient, db: Session
) -> None:
    """The owner-confirmed alias contract, at the API: a rename writes
    ``source_labels`` and leaves ``documents.source`` (and with it every
    ``source/path`` locator, citation and KB-tree node) byte-identical —
    and the catalogue's identity column stays the DERIVED name."""
    _seed_three_kinds(db)
    before = db.execute(
        text("SELECT source, path FROM documents ORDER BY source, path")
    ).fetchall()

    r = admin_client.put(
        "/api/source-labels/gitea-forge", json={"display_name": "the git server"}
    )
    assert r.status_code == 200, r.text

    db.expire_all()
    after = db.execute(
        text("SELECT source, path FROM documents ORDER BY source, path")
    ).fetchall()
    assert after == before
    assert [row[0] for row in after] == ["corpus-upload", "gitea-forge"]

    catalogue = admin_client.get("/api/source-labels").json()["labels"]
    assert [entry["source"] for entry in catalogue] == [
        "corpus-upload",
        "gitea-forge",
        "media-vault",
    ]
    assert catalogue[1]["display_name"] == "the git server"

    # And the PUT still keys on the DERIVED name after the rename (the
    # display name is never an address).
    assert admin_client.put(
        "/api/source-labels/the git server", json={"label": "nope"}
    ).status_code == 404