tests/test_admin_endpoints.py

"""Tests for the /admin/* surface in api.py.

These endpoints power the deploy/restore CLI (scripts/admin_deploy.py) and the
help-a-stuck-user runbook (regenerate-token, scrub-plays, force-link, etc.). A careless
edit to /admin/import would clobber every profile silently — this file is the regression
gate that catches it.

Pattern: in-memory SQLite per test (copies the bundled seed), fresh `api` module reload,
TestClient. Mirrors `tests/test_auth_login.py:14-27`.
"""
from __future__ import annotations

import io
import shutil
import sqlite3
import sys
from pathlib import Path
from unittest.mock import patch

import pytest
from fastapi.testclient import TestClient

_SEED = Path(__file__).resolve().parent.parent / "deploy/hf/seed/hf_bundled_store.db"
_ADMIN_SECRET = "test-secret-12345"


@pytest.fixture()
def admin_api(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    """Returns (api_module, client, store_path) with admin auth wired up."""
    assert _SEED.is_file(), f"Missing seed DB: {_SEED}"
    dest = tmp_path / "admin_test_store.db"
    shutil.copyfile(_SEED, dest)
    monkeypatch.setenv("KINK_STORE_PATH", str(dest))
    monkeypatch.setenv("KINK_SKIP_HEAVY_WARM", "1")
    monkeypatch.setenv("KINK_HF_REQUIRE_FULL_CATALOG", "0")
    monkeypatch.setenv("KINK_ADMIN_SECRET", _ADMIN_SECRET)
    sys.modules.pop("api", None)
    import api as api_mod

    api_mod._backend_impl = None
    api_mod.ADMIN_SECRET = _ADMIN_SECRET  # picks up new env via re-import, but be explicit
    api_mod._get_backend()
    client = TestClient(api_mod.app)
    return api_mod, client, dest


def _hdr(secret: str = _ADMIN_SECRET) -> dict:
    return {"x-admin-secret": secret}


def _make_user(client: TestClient) -> dict:
    return client.post("/users", json={}).json()


def _save_play(client: TestClient, user: dict, kink_id: str, rating: str = "love",
               directions: list[str] | None = None) -> None:
    r = client.post(
        f"/users/{user['id']}/plays",
        json={"kink_id": kink_id, "interest_state": rating, "directions": directions or ["together"]},
        headers={"x-private-token": user["private_token"]},
    )
    assert r.status_code in (200, 201), r.text


# ─────────────────────────────────────────────────────────────────────────────
# Phase 1.4 — auth gate sweep
# ─────────────────────────────────────────────────────────────────────────────


def test_every_admin_route_rejects_missing_secret(admin_api) -> None:
    """If a future endpoint forgets ``require_admin``, this test fails. Sweeps the actual
    FastAPI route table — no hand-maintained list to drift."""
    api_mod, client, _ = admin_api
    paths_methods: list[tuple[str, str]] = []
    for route in api_mod.app.routes:
        path = getattr(route, "path", "")
        if not path.startswith("/admin/"):
            continue
        methods = (getattr(route, "methods", set()) or set()) - {"HEAD", "OPTIONS"}
        for m in methods:
            paths_methods.append((m, path))

    assert paths_methods, "no admin routes discovered — registration broken?"

    # Concrete-path substitutions for routes with placeholders so we hit handler not 404.
    substitutions = {
        "{user_id}": "real-bunny-72",
        "{kink_id}": "demo_kink_5",
        "{partner_id}": "exact-wahoo-46",
        "{user_a}": "real-bunny-72",
        "{user_b}": "exact-wahoo-46",
    }

    for method, path in paths_methods:
        concrete = path
        for k, v in substitutions.items():
            concrete = concrete.replace(k, v)
        r = client.request(method, concrete)
        assert r.status_code == 403, f"{method} {concrete} returned {r.status_code} (expected 403)"


def test_admin_rejects_wrong_secret(admin_api) -> None:
    _, client, _ = admin_api
    r = client.get("/admin/users", headers={"x-admin-secret": "not-the-real-secret"})
    assert r.status_code == 403


# ─────────────────────────────────────────────────────────────────────────────
# Phase 1.2 — coverage matrix
# ─────────────────────────────────────────────────────────────────────────────


def test_list_users_returns_counts(admin_api) -> None:
    _, client, _ = admin_api
    u1 = _make_user(client)
    u2 = _make_user(client)
    _save_play(client, u1, "demo_kink_3")
    _save_play(client, u1, "demo_kink_5")
    _save_play(client, u2, "demo_kink_1")

    r = client.get("/admin/users", headers=_hdr())
    assert r.status_code == 200, r.text
    body = r.json()
    by_id = {it["user_id"]: it for it in body["items"]}
    assert by_id[u1["id"]]["plays"] == 2
    assert by_id[u2["id"]]["plays"] == 1
    assert body["total"] >= 2


def test_get_user_includes_private_token(admin_api) -> None:
    _, client, _ = admin_api
    u = _make_user(client)
    r = client.get(f"/admin/users/{u['id']}", headers=_hdr())
    assert r.status_code == 200
    body = r.json()
    assert body["id"] == u["id"]
    assert body["private_token"] == u["private_token"]


def test_get_user_404_for_unknown(admin_api) -> None:
    _, client, _ = admin_api
    r = client.get("/admin/users/no-such-user", headers=_hdr())
    assert r.status_code == 404


def test_regenerate_token_invalidates_old_login(admin_api) -> None:
    _, client, _ = admin_api
    u = _make_user(client)
    old_token = u["private_token"]
    r = client.post(f"/admin/users/{u['id']}/regenerate-token", headers=_hdr())
    assert r.status_code == 200
    new_token = r.json()["private_token"]
    assert new_token != old_token

    # Old token rejected on /auth/login
    bad = client.post("/auth/login", json={"user_id": u["id"], "private_token": old_token})
    assert bad.status_code == 401
    # New token accepted
    good = client.post("/auth/login", json={"user_id": u["id"], "private_token": new_token})
    assert good.status_code == 200


def test_admin_delete_user_cascades(admin_api) -> None:
    _, client, _ = admin_api
    u = _make_user(client)
    _save_play(client, u, "demo_kink_3")
    r = client.delete(f"/admin/users/{u['id']}", headers=_hdr())
    assert r.status_code == 200

    # Subsequent /admin/users/{uid} → 404
    r2 = client.get(f"/admin/users/{u['id']}", headers=_hdr())
    assert r2.status_code == 404


def test_admin_delete_user_idempotent(admin_api) -> None:
    _, client, _ = admin_api
    r = client.delete("/admin/users/never-existed", headers=_hdr())
    assert r.status_code == 200  # no-op is fine


def test_scrub_plays_keeps_profile(admin_api) -> None:
    _, client, _ = admin_api
    u = _make_user(client)
    _save_play(client, u, "demo_kink_3")
    _save_play(client, u, "demo_kink_5")

    r = client.post(f"/admin/users/{u['id']}/scrub-plays", headers=_hdr())
    assert r.status_code == 200
    body = r.json()
    assert body["deleted"] == 2

    full = client.get(f"/admin/users/{u['id']}", headers=_hdr()).json()
    assert full["id"] == u["id"]
    assert full["plays"] == {}


def test_admin_delete_one_play(admin_api) -> None:
    _, client, _ = admin_api
    u = _make_user(client)
    _save_play(client, u, "demo_kink_3")
    _save_play(client, u, "demo_kink_5")
    r = client.delete(f"/admin/users/{u['id']}/plays/demo_kink_3", headers=_hdr())
    assert r.status_code == 200

    full = client.get(f"/admin/users/{u['id']}", headers=_hdr()).json()
    assert "demo_kink_3" not in full["plays"]
    assert "demo_kink_5" in full["plays"]


def test_force_link_creates_partner_link(admin_api) -> None:
    _, client, _ = admin_api
    a = _make_user(client)
    b = _make_user(client)
    r = client.post(f"/admin/users/{a['id']}/partners/{b['id']}/force-link", headers=_hdr())
    assert r.status_code == 200
    body = r.json()
    assert body["created"] is True

    a_full = client.get(f"/admin/users/{a['id']}", headers=_hdr()).json()
    assert b["id"] in a_full["partners"]


def test_force_link_idempotent(admin_api) -> None:
    _, client, _ = admin_api
    a = _make_user(client)
    b = _make_user(client)
    client.post(f"/admin/users/{a['id']}/partners/{b['id']}/force-link", headers=_hdr())
    r = client.post(f"/admin/users/{a['id']}/partners/{b['id']}/force-link", headers=_hdr())
    assert r.status_code == 200
    assert r.json()["created"] is False  # already exists


def test_force_link_rejects_self(admin_api) -> None:
    _, client, _ = admin_api
    u = _make_user(client)
    r = client.post(f"/admin/users/{u['id']}/partners/{u['id']}/force-link", headers=_hdr())
    assert r.status_code == 400


def test_unlink_drops_partner_link(admin_api) -> None:
    _, client, _ = admin_api
    a = _make_user(client)
    b = _make_user(client)
    client.post(f"/admin/users/{a['id']}/partners/{b['id']}/force-link", headers=_hdr())
    r = client.delete(f"/admin/partners/{a['id']}/{b['id']}", headers=_hdr())
    assert r.status_code == 200
    assert r.json()["removed"] is True

    a_full = client.get(f"/admin/users/{a['id']}", headers=_hdr()).json()
    assert b["id"] not in a_full["partners"]


def test_unlink_idempotent_when_no_link(admin_api) -> None:
    _, client, _ = admin_api
    a = _make_user(client)
    b = _make_user(client)
    r = client.delete(f"/admin/partners/{a['id']}/{b['id']}", headers=_hdr())
    assert r.status_code == 200
    assert r.json()["removed"] is False


def test_export_returns_user_state_db(admin_api, tmp_path: Path) -> None:
    _, client, _ = admin_api
    u = _make_user(client)
    _save_play(client, u, "demo_kink_3")

    r = client.get("/admin/export", headers=_hdr())
    assert r.status_code == 200
    assert r.headers["content-type"] == "application/octet-stream"
    body = r.content
    assert len(body) > 0

    # Round-trip: write to a fresh fresh sqlite, confirm USER_TABLES present.
    out = tmp_path / "exported.db"
    out.write_bytes(body)
    conn = sqlite3.connect(f"file:{out}?mode=ro", uri=True)
    try:
        users = conn.execute("SELECT id FROM user").fetchall()
        assert any(row[0] == u["id"] for row in users)
        plays = conn.execute(
            'SELECT user_id, kink_id FROM playpreference WHERE user_id = ?', (u["id"],)
        ).fetchall()
        assert ("demo_kink_3" in {p[1] for p in plays})
    finally:
        conn.close()


def test_import_applies_user_state(admin_api, tmp_path: Path) -> None:
    api_mod, client, store_path = admin_api
    # Create a known user, export it, then delete the user, then import — verify restored.
    u = _make_user(client)
    _save_play(client, u, "demo_kink_3")
    export_bytes = client.get("/admin/export", headers=_hdr()).content
    client.delete(f"/admin/users/{u['id']}", headers=_hdr())

    # User is gone now
    assert client.get(f"/admin/users/{u['id']}", headers=_hdr()).status_code == 404

    # Import the snapshot back
    r = client.post(
        "/admin/import",
        content=export_bytes,
        headers={**_hdr(), "Content-Type": "application/octet-stream"},
    )
    assert r.status_code == 200
    body = r.json()
    assert body["ok"] is True
    assert body["total_rows"] >= 2  # at minimum: 1 user + 1 play

    # User is back
    restored = client.get(f"/admin/users/{u['id']}", headers=_hdr()).json()
    assert restored["id"] == u["id"]
    assert "demo_kink_3" in restored["plays"]


def test_import_rejects_empty_body(admin_api) -> None:
    _, client, _ = admin_api
    r = client.post(
        "/admin/import",
        content=b"",
        headers={**_hdr(), "Content-Type": "application/octet-stream"},
    )
    assert r.status_code == 400


def test_stats_reflects_live_data(admin_api) -> None:
    _, client, _ = admin_api
    u1 = _make_user(client)
    u2 = _make_user(client)
    _save_play(client, u1, "demo_kink_3")
    _save_play(client, u1, "demo_kink_5")
    _save_play(client, u2, "demo_kink_3")

    r = client.get("/admin/stats", headers=_hdr())
    assert r.status_code == 200
    body = r.json()
    assert body["row_counts"]["user"] >= 2
    assert body["row_counts"]["playpreference"] >= 3
    # top_kinks: demo_kink_3 has 2 saves, demo_kink_5 has 1 → 247 ranks first
    if body["top_kinks"]:
        assert body["top_kinks"][0]["save_count"] >= body["top_kinks"][-1]["save_count"]


def test_admin_health_payload_shape(admin_api) -> None:
    _, client, _ = admin_api
    r = client.get("/admin/health", headers=_hdr())
    assert r.status_code == 200
    body = r.json()
    assert "ok" in body
    assert "store_path" in body
    snap = body["snapshot"]
    for key in ("enabled", "repo", "filename", "has_hf_token", "last_restore"):
        assert key in snap, f"snapshot missing {key}"


# ─────────────────────────────────────────────────────────────────────────────
# Phase 1.3 — snapshot push/pull (Hub-bound — mocked)
# ─────────────────────────────────────────────────────────────────────────────


def test_snapshot_push_disabled_returns_409(admin_api) -> None:
    _, client, _ = admin_api
    with patch("backend.user_snapshot_hub.snapshot_enabled", return_value=False):
        r = client.post("/admin/snapshot/push", headers=_hdr())
    assert r.status_code == 409


def test_snapshot_push_skips_empty_store(admin_api, tmp_path: Path, monkeypatch) -> None:
    """When the local store has no users, refuse to overwrite the Hub copy with empty bytes."""
    # Wipe all users from the seeded store first
    api_mod, client, store_path = admin_api
    body_users = client.get("/admin/users", headers=_hdr()).json()
    for it in body_users["items"]:
        client.delete(f"/admin/users/{it['user_id']}", headers=_hdr())

    with patch("backend.user_snapshot_hub.snapshot_enabled", return_value=True), \
         patch("backend.user_snapshot_hub.push_user_snapshot") as push_mock:
        r = client.post("/admin/snapshot/push", headers=_hdr())
    assert r.status_code == 200
    body = r.json()
    assert body["ok"] is False
    assert "skipped" in body
    push_mock.assert_not_called()


def test_snapshot_push_calls_hub_push(admin_api) -> None:
    _, client, _ = admin_api
    _make_user(client)  # ensure non-empty
    with patch("backend.user_snapshot_hub.snapshot_enabled", return_value=True), \
         patch("backend.user_snapshot_hub.push_user_snapshot", return_value=True) as push_mock:
        r = client.post("/admin/snapshot/push", headers=_hdr())
    assert r.status_code == 200
    body = r.json()
    assert body["ok"] is True
    push_mock.assert_called_once()


def test_snapshot_pull_disabled_returns_409(admin_api) -> None:
    _, client, _ = admin_api
    with patch("backend.user_snapshot_hub.snapshot_enabled", return_value=False):
        r = client.post("/admin/snapshot/pull", headers=_hdr())
    assert r.status_code == 409


def test_snapshot_pull_calls_hub_pull(admin_api, tmp_path: Path) -> None:
    """Mock pull_user_snapshot to write a small valid snapshot to the target path, then verify
    restore_user_state runs and refresh_cache is invoked."""
    _, client, _ = admin_api
    # Pre-seed a user, export bytes, then have the mocked pull_user_snapshot drop those bytes
    # at the requested target_path so restore_user_state has something legitimate to apply.
    u = _make_user(client)
    snapshot_bytes = client.get("/admin/export", headers=_hdr()).content

    def fake_pull(target_path):
        Path(target_path).write_bytes(snapshot_bytes)
        return True

    with patch("backend.user_snapshot_hub.snapshot_enabled", return_value=True), \
         patch("backend.user_snapshot_hub.pull_user_snapshot", side_effect=fake_pull) as pull_mock:
        r = client.post("/admin/snapshot/pull", headers=_hdr())
    assert r.status_code == 200
    body = r.json()
    assert body["ok"] is True
    assert body["total_rows"] >= 1
    pull_mock.assert_called_once()