"""Admin API: session-only access, invite flow → accept → login, tier/status updates, keys, last-admin guard,
user deletion, global usage, audit filters."""
from __future__ import annotations

import os
from urllib.parse import parse_qs, urlparse

from fastapi.testclient import TestClient

JSON = {"Content-Type": "application/json"}


def token_of(link: str) -> str:
    return parse_qs(urlparse(link).query)["token"][0]


def fresh_email(tag: str) -> str:
    return f"{tag}-{os.getpid()}-{os.urandom(3).hex()}@example.com"


def test_admin_requires_session_and_role(client, make_user, admin_web):
    """A6: an admin's API key does NOT open the admin area; plain users get FORBIDDEN."""
    _, key, _, _ = make_user()
    r = client.get("/v1/admin/users", headers={"Authorization": f"Bearer {key}"})
    assert r.status_code == 403 and r.json()["error"]["code"] == "SESSION_REQUIRED"
    assert client.get("/v1/admin/users").status_code == 401
    _, admin_key, _, _ = make_user(role="admin")
    r = client.get("/v1/admin/users", headers={"Authorization": f"Bearer {admin_key}"})
    assert r.status_code == 403 and r.json()["error"]["code"] == "SESSION_REQUIRED"
    web, _, _, _ = admin_web
    assert web.get("/v1/admin/users").status_code == 200
    # a signed-in plain user → FORBIDDEN
    _, _, email, pw = make_user(with_key=False)
    plain = TestClient(web.app)
    plain.post("/v1/auth/login", json={"email": email, "password": pw}, headers=JSON)
    r = plain.get("/v1/admin/users")
    assert r.status_code == 403 and r.json()["error"]["code"] == "FORBIDDEN"


def test_invite_flow(admin_web):
    web, _, _, _ = admin_web
    email = fresh_email("tim")
    r = web.post("/v1/admin/users", json={"email": email, "name": "Tim", "tier": "high_usage"}, headers=JSON)
    assert r.status_code == 201, r.text
    d = r.json()["data"]
    assert d["user"]["status"] == "invited" and d["user"]["tier"] == "high_usage" and d["user"]["keys_active"] == 1
    assert d["key"]["prefix"].startswith("hfmd_live_") and "key" not in d["key"]   # admins never see the full key here
    assert d["invitation"]["delivered"] is False and d["invitation"]["queued"] is False
    assert "/accept-invite?token=" in d["invitation"]["link"]
    uid = d["user"]["id"]
    # duplicate invite → 409
    r = web.post("/v1/admin/users", json={"email": email, "name": "Tim"}, headers=JSON)
    assert r.status_code == 409 and r.json()["error"]["code"] == "EMAIL_TAKEN"
    # detail shows the pending link, the keys and a 7-day usage series
    r = web.get(f"/v1/admin/users/{uid}")
    assert r.status_code == 200
    det = r.json()["data"]
    assert det["pending_invite_link"] == d["invitation"]["link"] and len(det["keys"]) == 1
    assert det["usage"]["range"] == "7d" and len(det["usage"]["points"]) == 168 and det["keys"][0]["scopes"] == ["data"]
    # user cannot log in yet (no password)
    guest = TestClient(web.app)
    r = guest.post("/v1/auth/login", json={"email": email, "password": "whatever-long-enough"}, headers=JSON)
    assert r.status_code == 401
    # re-send voids the first token
    r = web.post(f"/v1/admin/users/{uid}/invite", json={}, headers=JSON)
    assert r.status_code == 200
    link2 = r.json()["data"]["invitation"]["link"]
    r = guest.post("/v1/auth/accept-invite", json={"token": token_of(d["invitation"]["link"]), "password": "tim-passphrase-123"}, headers=JSON)
    assert r.status_code == 400 and r.json()["error"]["code"] == "INVALID_TOKEN"
    # accept with the fresh one → active + session
    r = guest.post("/v1/auth/accept-invite", json={"token": token_of(link2), "password": "tim-passphrase-123"}, headers=JSON)
    assert r.status_code == 200 and r.json()["data"]["status"] == "active" and r.json()["data"]["email_verified"] is True
    me = guest.get("/v1/me").json()["data"]
    assert me["user"]["email"] == email and me["limits"]["tier"] == "high_usage" and me["user"]["keys_active"] == 1
    guest.cookies.clear()
    r = guest.post("/v1/auth/login", json={"email": email, "password": "tim-passphrase-123"}, headers=JSON)
    assert r.status_code == 200
    guest.cookies.clear()
    # signing up again with an invited-then-active address → uniform 202, nothing changed
    r = guest.post("/v1/auth/signup", json={"email": email, "name": "Tim", "password": "another-passphrase"}, headers=JSON)
    assert r.status_code == 202 and "debug_link" not in r.json()["data"]
    assert guest.post("/v1/auth/login", json={"email": email, "password": "tim-passphrase-123"}, headers=JSON).status_code == 200


def test_signup_on_invited_address_resends_invitation(admin_web):
    web, _, _, _ = admin_web
    email = fresh_email("luca")
    web.post("/v1/admin/users", json={"email": email, "name": "Luca"}, headers=JSON)
    guest = TestClient(web.app)
    r = guest.post("/v1/auth/signup", json={"email": email, "name": "Luca", "password": "luca-passphrase-123"}, headers=JSON)
    assert r.status_code == 202 and r.json()["data"]["status"] == "verification_sent"     # uniform wording
    assert "/accept-invite?token=" in r.json()["data"]["debug_link"]                       # …but it IS the invitation


def test_create_active_user_update_tier_and_disable(admin_web):
    web, _, _, _ = admin_web
    email = fresh_email("ada")
    r = web.post("/v1/admin/users", json={"email": email, "name": "Ada", "password": "ada-passphrase-123"}, headers=JSON)
    assert r.status_code == 201 and r.json()["data"]["user"]["status"] == "active" and r.json()["data"]["invitation"] is None
    uid = r.json()["data"]["user"]["id"]
    # admin creates a key for her (shown once) with a lifetime, it honours the user tier then the override
    r = web.post(f"/v1/admin/users/{uid}/keys", json={"name": "handover", "expires_in_days": 90, "note": "handed over by mail"}, headers=JSON)
    assert r.status_code == 201 and r.json()["data"]["expires_at"] and r.json()["data"]["note"] == "handed over by mail"
    key = r.json()["data"]["key"]
    assert web.get("/v1/status", headers={"Authorization": f"Bearer {key}"}).headers["X-RateLimit-Limit-Requests"] == "120"
    r = web.patch(f"/v1/admin/users/{uid}", json={"tier": "high_usage"}, headers=JSON)
    assert r.status_code == 200 and r.json()["data"]["tier"] == "high_usage"
    assert web.get("/v1/status", headers={"Authorization": f"Bearer {key}"}).headers["X-RateLimit-Limit-Requests"] == "600"
    r = web.patch(f"/v1/admin/users/{uid}", json={"tier": "platinum"}, headers=JSON)
    assert r.status_code == 422
    # disable → key refused, login refused
    r = web.patch(f"/v1/admin/users/{uid}", json={"status": "disabled"}, headers=JSON)
    assert r.status_code == 200
    r = web.get("/v1/status", headers={"Authorization": f"Bearer {key}"})
    assert r.status_code == 403 and r.json()["error"]["code"] == "ACCOUNT_DISABLED"
    guest = TestClient(web.app)
    r = guest.post("/v1/auth/login", json={"email": email, "password": "ada-passphrase-123"}, headers=JSON)
    assert r.status_code == 403 and r.json()["error"]["code"] == "ACCOUNT_DISABLED"
    assert web.post(f"/v1/admin/users/{uid}/reset-password", json={}, headers=JSON).status_code == 409
    # re-enable and revoke via admin
    web.patch(f"/v1/admin/users/{uid}", json={"status": "active"}, headers=JSON)
    kid = web.get(f"/v1/admin/users/{uid}").json()["data"]["keys"][0]["id"]
    r = web.delete(f"/v1/admin/users/{uid}/keys/{kid}", headers=JSON)
    assert r.status_code == 200 and r.json()["data"]["status"] == "revoked"
    assert web.get("/v1/status", headers={"Authorization": f"Bearer {key}"}).status_code == 401
    # reset link for the user
    r = web.post(f"/v1/admin/users/{uid}/reset-password", json={}, headers=JSON)
    assert r.status_code == 200 and "/reset-password?token=" in r.json()["data"]["reset"]["link"]
    assert web.get(f"/v1/admin/users/{uid}").json()["data"]["pending_reset_link"] == r.json()["data"]["reset"]["link"]
    assert web.get("/v1/admin/users/999999").json()["error"]["code"] == "USER_NOT_FOUND"


def test_last_admin_guard_and_self_protection(admin_web, make_user):
    """A19."""
    from accounts.models import User
    from core.db import session
    from sqlalchemy import select, update
    web, admin_id, _, _ = admin_web
    # an admin cannot touch their own role / status
    r = web.patch(f"/v1/admin/users/{admin_id}", json={"role": "user"}, headers=JSON)
    assert r.status_code == 403 and r.json()["error"]["code"] == "FORBIDDEN"
    assert web.patch(f"/v1/admin/users/{admin_id}", json={"status": "disabled"}, headers=JSON).status_code == 403
    assert web.patch(f"/v1/admin/users/{admin_id}", json={"name": "Root"}, headers=JSON).status_code == 200
    assert web.delete(f"/v1/admin/users/{admin_id}", headers=JSON).status_code == 403
    # make our admin the ONLY active admin, then try to demote another (now demoted) admin… and ourselves through the service
    other_id, _, _, _ = make_user(role="admin", with_key=False)
    with session() as s:
        s.execute(update(User).where(User.role == "admin", User.status == "active", User.id.notin_([admin_id, other_id])).values(role="user"))
    r = web.patch(f"/v1/admin/users/{other_id}", json={"role": "user"}, headers=JSON)
    assert r.status_code == 200                                    # two admins → demoting one is fine
    assert web.get("/v1/admin/users?limit=1").json()["meta"]["admins_active"] == 1
    from accounts import service
    from core.errors import ApiError
    with session() as s:
        me = s.get(User, admin_id)
        for kwargs in ({"role": "user"}, {"status": "disabled"}):
            try:
                service.update_user(s, me, actor="test", **kwargs)
                assert False, "last admin must be protected"
            except ApiError as exc:
                assert exc.code == "LAST_ADMIN" and exc.status == 409
        try:
            service.delete_account(s, me, actor="test")
            assert False
        except ApiError as exc:
            assert exc.code == "LAST_ADMIN"
        s.rollback()
    # promoting someone lifts the guard
    web.patch(f"/v1/admin/users/{other_id}", json={"role": "admin"}, headers=JSON)
    assert web.get("/v1/admin/users?limit=1").json()["meta"]["admins_active"] == 2


def test_role_change_signs_the_user_out(admin_web, make_user, signin):
    """A11: bumping session_version on role change."""
    web, _, _, _ = admin_web
    uid, _, email, pw = make_user(with_key=False)
    user_web = TestClient(web.app)
    signin(user_web, email, pw)
    assert user_web.get("/v1/me").status_code == 200
    assert web.patch(f"/v1/admin/users/{uid}", json={"role": "admin"}, headers=JSON).status_code == 200
    assert user_web.get("/v1/me").status_code == 401


def test_admin_delete_user(admin_web, make_user, signin):
    web, _, _, _ = admin_web
    uid, key, email, pw = make_user()
    user_web = TestClient(web.app)
    signin(user_web, email, pw)
    r = web.delete(f"/v1/admin/users/{uid}", headers=JSON)
    assert r.status_code == 200 and r.json()["data"]["status"] == "deleted" and r.json()["data"]["email"].endswith("@deleted.invalid")
    assert user_web.get("/v1/me").status_code == 401
    assert web.get("/v1/status", headers={"Authorization": f"Bearer {key}"}).status_code == 401
    assert web.patch(f"/v1/admin/users/{uid}", json={"tier": "high_usage"}, headers=JSON).status_code == 409
    assert web.post(f"/v1/admin/users/{uid}/invite", json={}, headers=JSON).status_code == 409
    assert web.get(f"/v1/admin/users?status=deleted").json()["data"][0]["status"] == "deleted"


def test_list_users_pagination_and_search(admin_web):
    web, _, _, _ = admin_web
    emails = [fresh_email("page") for _ in range(3)]
    for e in emails:
        web.post("/v1/admin/users", json={"email": e, "name": "Paged Person"}, headers=JSON)
    r = web.get("/v1/admin/users?search=paged&limit=2")
    assert r.status_code == 200
    body = r.json()
    assert len(body["data"]) == 2 and body["meta"]["next_cursor"] and "admins_active" in body["meta"]
    r2 = web.get(f"/v1/admin/users?search=paged&limit=2&cursor={body['meta']['next_cursor']}")
    assert r2.status_code == 200 and len(r2.json()["data"]) >= 1
    ids = {u["id"] for u in body["data"]} | {u["id"] for u in r2.json()["data"]}
    assert len(ids) >= 3
    assert web.get("/v1/admin/users?limit=0").status_code == 422


def test_global_usage_and_audit(admin_web, make_user):
    web, admin_id, _, _ = admin_web
    uid, key, _, _ = make_user()
    web.get("/v1/_test/frame?n=100", headers={"Authorization": f"Bearer {key}"})
    import time

    from ratelimit import usage
    usage.fold(now_s=time.time() + 120)     # force the current minute to fold
    r = web.get("/v1/admin/usage?days=7&top=5")
    assert r.status_code == 200
    d = r.json()["data"]
    assert d["per_day"] and d["per_day"][-1]["requests"] >= 1 and d["totals"]["requests"] >= 1
    top = [t for t in d["top"] if t["user"] and t["user"]["id"] == uid]
    assert top and top[0]["rows"] >= 100 and top[0]["user"]["email"]
    r = web.get("/v1/admin/audit?limit=5")
    assert r.status_code == 200
    rows = r.json()["data"]
    assert rows and {"id", "ts", "actor", "action", "target", "meta"} <= set(rows[0])
    r = web.get("/v1/admin/audit?action=key.")
    assert all(x["action"].startswith("key.") for x in r.json()["data"])
    # B6: filters by actor / target
    r = web.get(f"/v1/admin/audit?actor=user:{admin_id}")
    assert r.json()["data"] and all(x["actor"] == f"user:{admin_id}" for x in r.json()["data"])
    r = web.get(f"/v1/admin/audit?target=user:{uid}")
    assert r.json()["data"] and all(x["target"] == f"user:{uid}" for x in r.json()["data"])
    assert web.get("/v1/admin/audit?target=user:0").json()["data"] == []