"""Tests for GET /admin/api/profiles.

Each test builds an isolated admin router over a temp DB and a temp copy of
config.yaml so the real repo config and DB are never touched. The endpoint
must compute admission counts by delegating to ``routing.select_candidates``
and ``routing.restrict_to_from_profile`` — the tests include a source-level
guard asserting those function names appear in src/admin.py.
"""

from __future__ import annotations

import re

import sqlite3
from pathlib import Path

import pytest
from fastapi import FastAPI
from starlette.testclient import TestClient

from admin import build_router
from config import BUILTIN_PROFILES, load_config

ROOT = Path(__file__).resolve().parent.parent
SCHEMA_SQL = (ROOT / "config" / "schema.sql").read_text()
ADMIN_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS admin_model_overrides (
    model_id         TEXT NOT NULL,
    provider         TEXT NOT NULL,
    availability     TEXT NOT NULL,
    reason           TEXT,
    updated_at       TEXT NOT NULL,
    PRIMARY KEY (model_id, provider)
);
CREATE INDEX IF NOT EXISTS idx_admin_model_overrides_availability
    ON admin_model_overrides (availability);
"""


def _make_db(tmp_path: Path) -> sqlite3.Connection:
    conn = sqlite3.connect(str(tmp_path / "profiles.db"))
    conn.row_factory = sqlite3.Row
    return conn


def _seed_models(conn: sqlite3.Connection, rows: list[dict]) -> None:
    cols = [
        "model_id", "provider", "base_model_id", "display_name",
        "cost_per_1m_prompt", "cost_per_1m_completion", "context_window",
        "effective_context_window", "max_output_tokens", "tier",
        "supports_tools", "supports_json_mode", "supports_vision",
        "supports_reasoning", "reasoning_default_enabled", "latency_class",
        "reasoning_mode", "context_variant", "access_level", "deprecated",
        "availability", "last_updated", "eligible_categories",
    ]
    placeholders = ",".join(["?"] * len(cols))
    for r in rows:
        values = [r.get(c) for c in cols]
        conn.execute(
            f"INSERT INTO models ({','.join(cols)}) VALUES ({placeholders})",
            values,
        )
    conn.commit()


def _profile_client(
    tmp_path,
    profiles_yaml: str | None = None,
    overlay_profiles_yaml: str | None = None,
):
    """Build a TestClient for /admin with a temp DB and config copies."""
    (tmp_path / "config").mkdir(parents=True, exist_ok=True)
    config_yaml = tmp_path / "config" / "config.yaml"
    local_yaml = tmp_path / "config" / "config.local.yaml"

    base_text = (ROOT / "config" / "config.yaml").read_text()
    if profiles_yaml is not None:
        config_yaml.write_text(base_text + profiles_yaml)
    else:
        config_yaml.write_text(base_text)

    if overlay_profiles_yaml is not None:
        local_yaml.write_text(overlay_profiles_yaml)

    cfg = load_config(str(config_yaml))

    def _db_factory() -> sqlite3.Connection:
        return _make_db(tmp_path)

    router = build_router(cfg, _db_factory, base_dir=str(tmp_path))
    app = FastAPI()
    app.include_router(router, prefix="/admin")
    return TestClient(app), cfg, config_yaml, local_yaml


def _model_row(
    model_id: str,
    *,
    provider: str = "neuralwatt",
    tier: int,
    cost_completion: float = 1.0,
    availability: str = "active",
    deprecated: int = 0,
    latency_class: str = "standard",
    eligible_categories: str | None = None,
) -> dict:
    return {
        "eligible_categories": eligible_categories,
        "model_id": model_id,
        "provider": provider,
        "base_model_id": model_id,
        "display_name": model_id,
        "cost_per_1m_prompt": cost_completion * 0.5,
        "cost_per_1m_completion": cost_completion,
        "context_window": 131072,
        "effective_context_window": 65536,
        "max_output_tokens": 8192,
        "tier": tier,
        "supports_tools": 1,
        "supports_json_mode": 1,
        "supports_vision": 1,
        "supports_reasoning": 1,
        "reasoning_default_enabled": 1,
        "latency_class": latency_class,
        "reasoning_mode": "default",
        "context_variant": "full",
        "access_level": "public",
        "deprecated": deprecated,
        "availability": availability,
        "last_updated": "2026-08-22T00:00:00+00:00",
    }


@pytest.fixture
def client_no_models(tmp_path):
    """Fresh DB, no rows; tests zero_admit and 200 behaviour."""
    client, _cfg, _, _ = _profile_client(tmp_path)
    conn = _make_db(tmp_path)
    conn.executescript(SCHEMA_SQL)
    conn.executescript(ADMIN_TABLE_SQL)
    conn.close()
    yield client, _cfg


@pytest.fixture
def client_with_models(tmp_path):
    """DB seeded with tiered/cost-differentiated rows."""
    client, _cfg, _, _ = _profile_client(tmp_path)
    conn = _make_db(tmp_path)
    conn.executescript(SCHEMA_SQL)
    conn.executescript(ADMIN_TABLE_SQL)
    # 1 cheap tier-1 model within onlycheaps cost cap (0.5)
    # 2 tier-2 models above cost cap
    # 2 tier-3 frontier models
    # 1 flex tier-2 model (admitted under batch, excluded under interactive)
    rows = [
        _model_row("t1-cheap", tier=1, cost_completion=0.4),
        _model_row("t2-mid-a", tier=2, cost_completion=1.5),
        _model_row("t2-mid-b", tier=2, cost_completion=1.6),
        _model_row("t3-front-a", tier=3, cost_completion=2.0),
        _model_row("t3-front-b", tier=3, cost_completion=2.5),
        _model_row("t2-flex", tier=2, cost_completion=1.5, latency_class="flex"),
    ]
    _seed_models(conn, rows)
    conn.close()
    yield client, _cfg


@pytest.fixture
def client_with_local_only(tmp_path):
    """DB seeded with only an ollama-local model for the locality profile."""
    client, _cfg, _, _ = _profile_client(tmp_path)
    conn = _make_db(tmp_path)
    conn.executescript(SCHEMA_SQL)
    conn.executescript(ADMIN_TABLE_SQL)
    rows = [
        _model_row(
            "local-model",
            provider="ollama-local",
            tier=1,
            cost_completion=0.0,
        ),
    ]
    _seed_models(conn, rows)
    conn.close()
    yield client, _cfg


# ---------------------------------------------------------------------------
# Source-level guard (grep check)
# ---------------------------------------------------------------------------


def test_implementation_uses_routing_helpers():
    """src/admin.py must call select_candidates and restrict_to_from_profile.

    This guards against a future refactor that re-implements the profile
    predicate logic inside the endpoint.
    """
    text = (ROOT / "src" / "admin.py").read_text()
    assert "select_candidates" in text
    assert "restrict_to_from_profile" in text


# ---------------------------------------------------------------------------
# Enumeration and shape
# ---------------------------------------------------------------------------


def _by_name(body: list[dict]) -> dict[str, dict]:
    return {r["name"]: r for r in body}


def test_profiles_endpoint_lists_builtins_and_config_profiles(tmp_path):
    """Endpoint returns all 5 builtins plus a config-defined profile."""
    profiles_block = """
profiles:
  myconfig:
    min_tier: 1
    max_tier: 2
"""
    client, cfg, _, _ = _profile_client(tmp_path, profiles_block)
    conn = _make_db(tmp_path)
    conn.executescript(SCHEMA_SQL)
    conn.executescript(ADMIN_TABLE_SQL)
    conn.close()

    resp = client.get("/admin/api/profiles")
    assert resp.status_code == 200
    body = resp.json()
    names = [r["name"] for r in body]

    assert names == ["default", "batch", "locality", "bigboybritches", "onlycheaps", "myconfig"]
    by_name = _by_name(body)
    for builtin_name in BUILTIN_PROFILES:
        assert by_name[builtin_name]["source"] == "builtin"
    assert by_name["myconfig"]["source"] == "config"
    assert by_name["myconfig"]["definition"]["min_tier"] == 1
    assert by_name["myconfig"]["definition"]["max_tier"] == 2


def test_profiles_definition_serialises_allowed_model_ids_set(tmp_path):
    """allowed_model_ids is exposed as a sorted stable list, not a set."""
    profiles_block = """
profiles:
  allowlist:
    allowed_model_ids: [z-model, a-model, m-model]
"""
    client, cfg, _, _ = _profile_client(tmp_path, profiles_block)
    conn = _make_db(tmp_path)
    conn.executescript(SCHEMA_SQL)
    conn.executescript(ADMIN_TABLE_SQL)
    conn.close()

    resp = client.get("/admin/api/profiles")
    by_name = _by_name(resp.json())
    assert by_name["allowlist"]["definition"]["allowed_model_ids"] == ["a-model", "m-model", "z-model"]


# ---------------------------------------------------------------------------
# Admission counts
# ---------------------------------------------------------------------------


def test_profiles_exact_counts_by_profile(client_with_models):
    """Counts match the seeded set and the canonical probe definition."""
    client, cfg = client_with_models
    resp = client.get("/admin/api/profiles")
    assert resp.status_code == 200
    by_name = _by_name(resp.json())

    # default: allowed public/active; interactive drops flex -> 5 rows
    assert by_name["default"]["admitted_count"] == 5
    assert by_name["default"]["admitted_models"] == [
        "t1-cheap", "t2-mid-a", "t2-mid-b", "t3-front-a", "t3-front-b"
    ]
    # batch: same as default + flex -> 6
    assert by_name["batch"]["admitted_count"] == 6
    # bigboybritches: tier >= 3 -> 2
    assert by_name["bigboybritches"]["admitted_count"] == 2
    assert by_name["bigboybritches"]["admitted_models"] == ["t3-front-a", "t3-front-b"]
    # onlycheaps: completion <= 0.5 -> 1 (the tier-1 cheap model)
    assert by_name["onlycheaps"]["admitted_count"] == 1
    assert by_name["onlycheaps"]["admitted_models"] == ["t1-cheap"]


def test_locality_profile_with_local_model(client_with_local_only):
    """locality filters to ollama-local provider rows."""
    client, cfg = client_with_local_only
    resp = client.get("/admin/api/profiles")
    assert resp.status_code == 200
    by_name = _by_name(resp.json())
    # Known edge: the canonical probe passes task_category=None, and
    # eligible_categories is a restrict-only gate, so a row naming categories
    # is excluded when the task names none. The seed here has no
    # eligible_categories, so locality should succeed.
    assert by_name["locality"]["admitted_count"] == 1
    assert by_name["locality"]["admitted_models"] == ["local-model"]


# ---------------------------------------------------------------------------
# The zero-admit badge must not cry wolf on a category-gated profile
# ---------------------------------------------------------------------------


@pytest.fixture
def client_with_gated_local(tmp_path):
    """A cloud row plus the shape production actually has: a local row whose
    ``eligible_categories`` restricts it to the two categories it serves."""
    client, _cfg, _, _ = _profile_client(tmp_path)
    conn = _make_db(tmp_path)
    conn.executescript(SCHEMA_SQL)
    conn.executescript(ADMIN_TABLE_SQL)
    _seed_models(
        conn,
        [
            _model_row("t1-cheap", tier=1, cost_completion=0.4),
            _model_row(
                "qwen2.5-coder-router:14b",
                provider="ollama-local",
                tier=1,
                cost_completion=0.0,
                eligible_categories="file_summarization,diff_checking",
            ),
        ],
    )
    conn.close()
    yield client, _cfg


def test_locality_does_not_cry_wolf_while_a_local_row_is_active(
    client_with_gated_local,
):
    """The badge fired on a profile that works, which is worse than cosmetic.

    ``locality`` genuinely admits one model for the two categories it exists
    to serve, and zero for a category-less probe. Telling an operator that a
    working profile is broken trains people to ignore the badge on the day it
    is real — the day it is real being incident #3.
    """
    client, cfg = client_with_gated_local
    locality = _by_name(client.get("/admin/api/profiles").json())["locality"]

    assert locality["zero_admit"] is False, (
        "zero-admit alarm raised while an active ollama-local row serves "
        "its eligible categories"
    )


def test_locality_coverage_reports_the_honest_number(client_with_gated_local):
    """Not "admits 0 models" but "admits 1 model for 2 of N categories"."""
    client, cfg = client_with_gated_local
    locality = _by_name(client.get("/admin/api/profiles").json())["locality"]
    coverage = locality["category_coverage"]

    assert coverage["total"] == len(cfg.proficiency.categories)
    assert coverage["admitting"] == 2
    assert coverage["max_admitted"] == 1
    # Both stats come from the same probe: a card reading "1 admitted / 0
    # interactive" would contradict the very line explaining it.
    assert coverage["interactive_admitting"] == 2
    assert coverage["max_interactive_admitted"] == 1
    assert coverage["models"] == ["qwen2.5-coder-router:14b"]
    # The category-less probe is still reported, and is still 0 — it answers a
    # different question, and the page now labels it as such rather than
    # turning it into an alarm.
    assert locality["admitted_count"] == 0


def test_a_profile_admitting_nothing_anywhere_still_warns(client_with_gated_local):
    """The alarm has to survive being made honest, or it bought nothing."""
    client, _cfg = client_with_gated_local
    body = client.post(
        "/admin/api/profiles/",
        json={"name": "nowhere", "allowed_model_ids": ["does-not-exist"]},
    )
    assert body.status_code == 200
    assert body.json()["zero_admit"] is True


# ---------------------------------------------------------------------------
# Zero admission
# ---------------------------------------------------------------------------


def test_zero_admit_for_narrow_config_profile(tmp_path):
    """A config profile with an allowlist of non-existent models has zero admit."""
    profiles_block = """
profiles:
  empty:
    allowed_model_ids: [does-not-exist]
"""
    client, cfg, _, _ = _profile_client(tmp_path, profiles_block)
    conn = _make_db(tmp_path)
    conn.executescript(SCHEMA_SQL)
    conn.executescript(ADMIN_TABLE_SQL)
    _seed_models(conn, [_model_row("real", tier=1, cost_completion=0.1)])
    conn.close()

    resp = client.get("/admin/api/profiles")
    by_name = _by_name(resp.json())
    assert by_name["empty"]["admitted_count"] == 0
    assert by_name["empty"]["interactive_count"] == 0
    assert by_name["empty"]["zero_admit"] is True


def test_admin_override_removes_only_admitted_model(tmp_path):
    """Deprecating via admin_model_overrides drops a profile's count to zero."""
    profiles_block = """
profiles:
  single:
    allowed_model_ids: [only-me]
"""
    client, cfg, _, _ = _profile_client(tmp_path, profiles_block)
    conn = _make_db(tmp_path)
    conn.executescript(SCHEMA_SQL)
    conn.executescript(ADMIN_TABLE_SQL)
    _seed_models(conn, [_model_row("only-me", tier=1, cost_completion=0.1)])
    conn.close()

    resp = client.get("/admin/api/profiles")
    by_name = _by_name(resp.json())
    assert by_name["single"]["admitted_count"] == 1

    conn = _make_db(tmp_path)
    conn.execute(
        "INSERT INTO admin_model_overrides (model_id, provider, availability, updated_at) "
        "VALUES (?, ?, ?, ?)",
        ("only-me", "neuralwatt", "deprecated", "2026-08-22T00:00:00+00:00"),
    )
    conn.commit()
    conn.close()

    resp = client.get("/admin/api/profiles")
    by_name = _by_name(resp.json())
    assert by_name["single"]["admitted_count"] == 0
    assert by_name["single"]["zero_admit"] is True


def test_empty_models_table_returns_zero_admit_for_every_profile(client_no_models):
    """With no models every builtin profile returns 0 and 200, never 500."""
    client, cfg = client_no_models
    resp = client.get("/admin/api/profiles")
    assert resp.status_code == 200
    body = resp.json()
    assert len(body) == len(BUILTIN_PROFILES)
    for r in body:
        assert r["admitted_count"] == 0
        assert r["interactive_count"] == 0
        assert r["zero_admit"] is True
        assert r["admitted_models"] == []


# ---------------------------------------------------------------------------
# Interactive vs admitted
# ---------------------------------------------------------------------------


def test_batch_profile_admits_more_than_interactive(client_with_models):
    """A profile with latency_tolerance=batch returns admitted >= interactive+1."""
    client, cfg = client_with_models
    resp = client.get("/admin/api/profiles")
    by_name = _by_name(resp.json())
    assert by_name["batch"]["admitted_count"] == 6
    assert by_name["batch"]["interactive_count"] == 5


# ---------------------------------------------------------------------------
# Shape contract
# ---------------------------------------------------------------------------


def test_profiles_response_fields(client_with_models):
    """Every record contains the expected top-level keys."""
    client, cfg = client_with_models
    resp = client.get("/admin/api/profiles")
    body = resp.json()
    for r in body:
        assert set(r.keys()) == {
            "name",
            "source",
            "definition",
            "admitted_count",
            "interactive_count",
            "zero_admit",
            "admitted_models",
            "category_coverage",
            # Which profile bare `auto` resolves to, so the profiles page can
            # show and set it instead of sending the operator to a dropdown
            # on Controls. See test_admin_profile_is_default.py.
            "is_default",
        }
        assert r["source"] in {"builtin", "config"}
        assert isinstance(r["admitted_count"], int)
        assert isinstance(r["interactive_count"], int)
        assert isinstance(r["zero_admit"], bool)
        assert isinstance(r["admitted_models"], list)
        assert set(r["category_coverage"]) == {
            "total",
            "admitting",
            "interactive_admitting",
            "max_admitted",
            "max_interactive_admitted",
            "models",
        }


# ---------------------------------------------------------------------------
# CRUD endpoints
# ---------------------------------------------------------------------------


def _profile_client_with_models(tmp_path, profiles_yaml=None, overlay_profiles_yaml=None):
    client, cfg, config_yaml, local_yaml = _profile_client(
        tmp_path, profiles_yaml, overlay_profiles_yaml
    )
    conn = _make_db(tmp_path)
    conn.executescript(SCHEMA_SQL)
    conn.executescript(ADMIN_TABLE_SQL)
    rows = [
        _model_row("t1-cheap", tier=1, cost_completion=0.4),
        _model_row("t2-mid-a", tier=2, cost_completion=1.5),
        _model_row("t2-mid-b", tier=2, cost_completion=1.6),
        _model_row("t3-front-a", tier=3, cost_completion=2.0),
        _model_row("t3-front-b", tier=3, cost_completion=2.5),
        _model_row("t2-flex", tier=2, cost_completion=1.5, latency_class="flex"),
    ]
    _seed_models(conn, rows)
    conn.close()
    return client, cfg, config_yaml, local_yaml


def test_profile_crud_round_trip(tmp_path):
    """Create, update, delete an overlay profile and see it reflected in GET."""
    client, cfg, config_yaml, local_yaml = _profile_client_with_models(tmp_path)
    base_before = config_yaml.read_bytes()

    resp = client.post("/admin/api/profiles/", json={"name": "minit", "max_tier": 2})
    assert resp.status_code == 200
    body = resp.json()
    assert body["profile"]["name"] == "minit"
    assert body["profile"]["definition"]["max_tier"] == 2
    assert body["profile"]["source"] == "overlay"
    assert "zero_admit" in body

    resp = client.get("/admin/api/profiles")
    by_name = _by_name(resp.json())
    assert by_name["minit"]["source"] == "overlay"
    assert by_name["minit"]["definition"]["max_tier"] == 2

    resp = client.post("/admin/api/profiles/minit", json={"max_tier": 3})
    assert resp.status_code == 200
    assert resp.json()["profile"]["definition"]["max_tier"] == 3

    resp = client.get("/admin/api/profiles")
    by_name = _by_name(resp.json())
    assert by_name["minit"]["definition"]["max_tier"] == 3

    resp = client.delete("/admin/api/profiles/minit")
    assert resp.status_code == 200

    resp = client.get("/admin/api/profiles")
    assert "minit" not in _by_name(resp.json())

    assert load_config(str(config_yaml))
    assert config_yaml.read_bytes() == base_before
    assert "profiles:" not in config_yaml.read_text()
    assert "minit" not in local_yaml.read_text()


def test_builtins_are_read_only(tmp_path):
    client, cfg, _, _ = _profile_client_with_models(tmp_path)

    resp = client.post("/admin/api/profiles/", json={"name": "batch", "max_tier": 2})
    assert resp.status_code == 403

    resp = client.post("/admin/api/profiles/batch", json={"max_tier": 1})
    assert resp.status_code == 403

    resp = client.delete("/admin/api/profiles/batch")
    assert resp.status_code == 403


def test_delete_guard_current_default(tmp_path):
    client, cfg, _, _ = _profile_client_with_models(tmp_path)

    resp = client.post("/admin/api/profiles/", json={"name": "minit", "max_tier": 2})
    assert resp.status_code == 200

    resp = client.post(
        "/admin/api/config/routing.default_profile", json={"value": "minit"}
    )
    assert resp.status_code == 200

    resp = client.delete("/admin/api/profiles/minit")
    assert resp.status_code == 422
    assert "routing.default_profile" in resp.json()["detail"]


def test_rename_via_create_delete_refused_by_default(tmp_path):
    client, cfg, _, _ = _profile_client_with_models(tmp_path)

    resp = client.post("/admin/api/profiles/", json={"name": "oldname", "max_tier": 2})
    assert resp.status_code == 200
    client.post(
        "/admin/api/config/routing.default_profile", json={"value": "oldname"}
    ).raise_for_status()

    resp = client.post("/admin/api/profiles/", json={"name": "newname", "max_tier": 2})
    assert resp.status_code == 200

    resp = client.delete("/admin/api/profiles/oldname")
    assert resp.status_code == 422
    assert "routing.default_profile" in resp.json()["detail"]


def test_zero_admission_save_warning(tmp_path):
    client, cfg, _, _ = _profile_client_with_models(tmp_path)

    resp = client.post(
        "/admin/api/profiles/",
        json={"name": "ghost", "allowed_model_ids": ["does-not-exist"]},
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["zero_admit"] is True
    assert "warning" in body

    resp = client.post("/admin/api/profiles/", json={"name": "normal", "max_tier": 3})
    assert resp.status_code == 200
    body = resp.json()
    assert body["zero_admit"] is False
    assert "warning" not in body


def test_allowed_model_ids_null_round_trip(tmp_path):
    """allowed_model_ids: null persists as null, not a list, in the overlay."""
    client, cfg, config_yaml, local_yaml = _profile_client_with_models(tmp_path)

    resp = client.post("/admin/api/profiles/", json={"name": "nolist", "max_tier": 2})
    assert resp.status_code == 200

    base_text = config_yaml.read_text()
    local_text = local_yaml.read_text()
    assert re.search(
        r"^\s+allowed_model_ids:\s*$", base_text, re.MULTILINE
    ) is None
    assert re.search(
        r"^\s+allowed_model_ids:\s*$", local_text, re.MULTILINE
    ) is not None
    assert re.search(
        r"^\s+allowed_model_ids:\s*\[\]\s*$", local_text, re.MULTILINE
    ) is None

    resp = client.get("/admin/api/profiles")
    by_name = _by_name(resp.json())
    assert by_name["nolist"]["definition"]["allowed_model_ids"] is None


def test_profile_create_existing_returns_409(tmp_path):
    client, cfg, _, _ = _profile_client_with_models(tmp_path)

    resp = client.post("/admin/api/profiles/", json={"name": "minit", "max_tier": 2})
    assert resp.status_code == 200

    resp = client.post("/admin/api/profiles/", json={"name": "minit", "max_tier": 2})
    assert resp.status_code == 409
    assert "update" in resp.json()["detail"].lower()


def test_profile_update_unknown_returns_404(tmp_path):
    client, cfg, _, _ = _profile_client_with_models(tmp_path)
    resp = client.post("/admin/api/profiles/nosuch", json={"max_tier": 2})
    assert resp.status_code == 404


def test_profile_delete_unknown_returns_404(tmp_path):
    client, cfg, _, _ = _profile_client_with_models(tmp_path)
    resp = client.delete("/admin/api/profiles/nosuch")
    assert resp.status_code == 404


def test_profile_create_bad_tier_returns_422(tmp_path):
    client, cfg, _, _ = _profile_client_with_models(tmp_path)
    resp = client.post(
        "/admin/api/profiles/", json={"name": "bad", "min_tier": 9}
    )
    assert resp.status_code == 422


# ---------------------------------------------------------------------------
# Overlay-specific profile CRUD
# ---------------------------------------------------------------------------


def test_overlay_profile_is_visible_in_listing(tmp_path):
    base_block = """
profiles:
  shared:
    min_tier: 1
    max_tier: 2
"""
    overlay_block = """
profiles:
  onlyoverlay:
    min_tier: 1
    max_tier: 3
"""
    client, cfg, _, local_yaml = _profile_client_with_models(
        tmp_path, base_block, overlay_block
    )

    resp = client.get("/admin/api/profiles")
    assert resp.status_code == 200
    by_name = _by_name(resp.json())
    assert by_name["onlyoverlay"]["source"] == "overlay"
    assert by_name["onlyoverlay"]["definition"]["max_tier"] == 3


def test_overlay_profile_is_editable(tmp_path):
    overlay_block = """
profiles:
  onlyoverlay:
    min_tier: 1
    max_tier: 2
"""
    client, cfg, _, local_yaml = _profile_client_with_models(
        tmp_path, overlay_profiles_yaml=overlay_block
    )

    resp = client.post("/admin/api/profiles/onlyoverlay", json={"max_tier": 3})
    assert resp.status_code == 200
    assert resp.json()["profile"]["definition"]["max_tier"] == 3

    resp = client.get("/admin/api/profiles")
    assert resp.status_code == 200
    by_name = _by_name(resp.json())
    assert by_name["onlyoverlay"]["source"] == "overlay"
    assert by_name["onlyoverlay"]["definition"]["max_tier"] == 3


def test_overlay_profile_delete_removes_from_overlay(tmp_path):
    overlay_block = """
profiles:
  onlyoverlay:
    min_tier: 1
    max_tier: 2
"""
    client, cfg, _, local_yaml = _profile_client_with_models(
        tmp_path, overlay_profiles_yaml=overlay_block
    )

    resp = client.delete("/admin/api/profiles/onlyoverlay")
    assert resp.status_code == 200

    resp = client.get("/admin/api/profiles")
    assert resp.status_code == 200
    assert "onlyoverlay" not in _by_name(resp.json())
    assert "onlyoverlay" not in local_yaml.read_text()


def test_base_profile_update_returns_403_naming_base(tmp_path):
    base_block = """
profiles:
  baseonly:
    min_tier: 1
    max_tier: 2
"""
    client, cfg, config_yaml, local_yaml = _profile_client_with_models(
        tmp_path, base_block
    )
    base_before = config_yaml.read_bytes()

    resp = client.post("/admin/api/profiles/baseonly", json={"max_tier": 3})
    assert resp.status_code == 403
    assert "config/config.yaml" in resp.json()["detail"]
    assert config_yaml.read_bytes() == base_before


def test_base_profile_delete_returns_403_naming_base(tmp_path):
    base_block = """
profiles:
  baseonly:
    min_tier: 1
    max_tier: 2
"""
    client, cfg, config_yaml, local_yaml = _profile_client_with_models(
        tmp_path, base_block
    )
    base_before = config_yaml.read_bytes()

    resp = client.delete("/admin/api/profiles/baseonly")
    assert resp.status_code == 403
    assert "config/config.yaml" in resp.json()["detail"]
    assert config_yaml.read_bytes() == base_before


def test_create_shadowing_base_profile_returns_409_naming_base(tmp_path):
    base_block = """
profiles:
  baseonly:
    min_tier: 1
    max_tier: 2
"""
    client, cfg, config_yaml, local_yaml = _profile_client_with_models(
        tmp_path, base_block
    )
    base_before = config_yaml.read_bytes()

    resp = client.post(
        "/admin/api/profiles/", json={"name": "baseonly", "max_tier": 3}
    )
    assert resp.status_code == 409
    assert "config/config.yaml" in resp.json()["detail"]
    assert config_yaml.read_bytes() == base_before


def test_base_overlay_collision_resolves_to_overlay_and_labels_source(tmp_path):
    base_block = """
profiles:
  shared:
    min_tier: 1
    max_tier: 2
"""
    overlay_block = """
profiles:
  shared:
    min_tier: 1
    max_tier: 3
"""
    client, cfg, _, local_yaml = _profile_client_with_models(
        tmp_path, base_block, overlay_block
    )

    resp = client.get("/admin/api/profiles")
    assert resp.status_code == 200
    by_name = _by_name(resp.json())
    assert by_name["shared"]["source"] == "overlay"
    assert by_name["shared"]["definition"]["max_tier"] == 3


def test_profile_write_creates_overlay_with_header(tmp_path):
    client, cfg, config_yaml, local_yaml = _profile_client_with_models(tmp_path)

    resp = client.post("/admin/api/profiles/", json={"name": "newprof", "max_tier": 2})
    assert resp.status_code == 200

    assert local_yaml.exists()
    local_text = local_yaml.read_text()
    assert local_text.startswith("# Machine-local overlay")
    assert "profiles:" in local_text
    assert "newprof:" in local_text


def test_profile_write_creates_overlay_backup(tmp_path):
    client, cfg, config_yaml, local_yaml = _profile_client_with_models(tmp_path)

    resp = client.post("/admin/api/profiles/", json={"name": "newprof", "max_tier": 2})
    assert resp.status_code == 200

    backups = sorted(tmp_path.glob("config/config.local.yaml.bak.*"))
    assert len(backups) == 1
    assert backups[0].read_text().strip() != ""


# ---------------------------------------------------------------------------
# Config-file corruption: clean 503s, no writes, no crashes
# ---------------------------------------------------------------------------

# Two top-level `profiles:` maps: ruamel raises DuplicateKeyError on load.
_DUPLICATE_PROFILES_BLOCK = """
profiles:
  aaa:
    min_tier: 1
profiles:
  bbb:
    min_tier: 2
"""


def _corrupt_config_client(tmp_path, corrupt_block: str):
    """A client over a hand-corrupted temp config.yaml.

    The in-memory cfg is loaded from the CLEAN repo config, mirroring a
    running service whose startup cfg is fine but whose file has since been
    hand-broken. ``raise_server_exceptions=False`` pins an unhandled endpoint
    exception as a raw 500 response instead of raising in the test.
    """
    (tmp_path / "config").mkdir(parents=True, exist_ok=True)
    config_yaml = tmp_path / "config" / "config.yaml"
    config_yaml.write_text(
        (ROOT / "config" / "config.yaml").read_text() + corrupt_block
    )

    cfg = load_config(ROOT / "config" / "config.yaml")

    def _db_factory() -> sqlite3.Connection:
        return _make_db(tmp_path)

    router = build_router(cfg, _db_factory, base_dir=str(tmp_path))
    app = FastAPI()
    app.include_router(router, prefix="/admin")
    client = TestClient(app, raise_server_exceptions=False)

    conn = _make_db(tmp_path)
    conn.executescript(SCHEMA_SQL)
    conn.executescript(ADMIN_TABLE_SQL)
    conn.close()
    return client, cfg, config_yaml


def test_profiles_GET_duplicate_profiles_keys_returns_503(tmp_path):
    """A duplicate-key config.yaml refuses with a clean 503, not a crash."""
    client, cfg, _ = _corrupt_config_client(
        tmp_path, _DUPLICATE_PROFILES_BLOCK
    )

    resp = client.get("/admin/api/profiles")
    assert resp.status_code == 503
    assert "config.yaml" in resp.json()["detail"]


def test_profiles_GET_malformed_entry_names_the_profile(tmp_path):
    """A hand-edited invalid profile entry refuses with 503 naming it."""
    bad_block = """
profiles:
  badprofile:
    min_tier: 9
"""
    client, cfg, _ = _corrupt_config_client(tmp_path, bad_block)

    resp = client.get("/admin/api/profiles")
    assert resp.status_code == 503
    assert "badprofile" in resp.json()["detail"]


def test_profiles_GET_non_mapping_entry_names_the_profile(tmp_path):
    """A profile entry that is not a mapping at all refuses with 503 naming it."""
    bad_block = """
profiles:
  scalar: 5
"""
    client, cfg, _ = _corrupt_config_client(tmp_path, bad_block)

    resp = client.get("/admin/api/profiles")
    assert resp.status_code == 503
    assert "scalar" in resp.json()["detail"]


def test_profile_create_against_corrupt_config_refuses_and_writes_nothing(tmp_path):
    """A create on a corrupt config.yaml is a clean 503 and touches no bytes."""
    client, cfg, config_yaml = _corrupt_config_client(
        tmp_path, _DUPLICATE_PROFILES_BLOCK
    )
    before = config_yaml.read_bytes()

    resp = client.post(
        "/admin/api/profiles/", json={"name": "minit", "max_tier": 2}
    )
    assert resp.status_code == 503
    assert "config.yaml" in resp.json()["detail"]

    assert config_yaml.read_bytes() == before
    assert not list(tmp_path.glob("config/config.local.yaml.bak.*"))


def test_profile_delete_against_corrupt_config_refuses_and_writes_nothing(tmp_path):
    """A delete on a corrupt config.yaml is a clean 503 and touches no bytes."""
    client, cfg, config_yaml = _corrupt_config_client(
        tmp_path, _DUPLICATE_PROFILES_BLOCK
    )
    before = config_yaml.read_bytes()

    resp = client.delete("/admin/api/profiles/whatever")
    assert resp.status_code == 503
    assert "config.yaml" in resp.json()["detail"]

    assert config_yaml.read_bytes() == before
    assert not list(tmp_path.glob("config/config.local.yaml.bak.*"))