"""Self-service account flows: signup → verify → login → key → use → revoke → 401, reset, CSRF, /me, account security.

Every test here that documents a fixed vulnerability is marked in its docstring with the finding id (A1…A24)."""
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_full_lifecycle(web):
    email = fresh_email("ada")
    pw = "very-secure-passphrase"
    # signup → 202 with (dev only) debug link
    r = web.post("/v1/auth/signup", json={"email": email, "name": "Ada", "password": pw}, headers=JSON)
    assert r.status_code == 202, r.text
    d = r.json()["data"]
    assert d["status"] == "verification_sent" and "debug_link" in d and "/verify?token=" in d["debug_link"]
    # cannot log in before verifying
    r = web.post("/v1/auth/login", json={"email": email, "password": pw}, headers=JSON)
    assert r.status_code == 403 and r.json()["error"]["code"] == "EMAIL_NOT_VERIFIED"
    # A9/A16: GET verify only redirects to the web page (no consumption, no cookie)
    r = web.get("/v1/auth/verify", params={"token": token_of(d["debug_link"])}, follow_redirects=False)
    assert r.status_code == 302 and r.headers["location"].endswith(f"/verify?token={token_of(d['debug_link'])}")
    assert "hfmd_session" not in r.cookies
    # POST verify → session cookie
    r = web.post("/v1/auth/verify", json={"token": token_of(d["debug_link"])}, headers=JSON)
    assert r.status_code == 200 and r.json()["data"]["email_verified"] is True and r.json()["data"]["kind"] == "verify"
    assert "hfmd_session" in r.cookies
    # token is single use
    assert web.post("/v1/auth/verify", json={"token": token_of(d["debug_link"])}, headers=JSON).json()["error"]["code"] == "INVALID_TOKEN"
    # /me with the cookie
    r = web.get("/v1/me")
    assert r.status_code == 200
    me = r.json()["data"]
    assert me["user"]["email"] == email and me["limits"]["tier"] == "free" and me["limits"]["requests"] == 120
    assert me["user"]["keys_active"] == 0 and me["auth"] == "session"
    # create key (shown once) with note + expiry
    r = web.post("/v1/me/keys", json={"name": "laptop", "note": "CI runner", "expires_in_days": 30}, headers=JSON)
    assert r.status_code == 201, r.text
    k = r.json()["data"]
    assert k["key"].startswith("hfmd_live_") and len(k["key"]) == len("hfmd_live_") + 32
    assert k["prefix"] == k["key"][:18] and k["name"] == "laptop" and k["note"] == "CI runner" and k["scopes"] == ["data"]
    assert k["expires_at"] and k["expired"] is False
    listed = web.get("/v1/me/keys").json()["data"]
    assert len(listed) == 1 and "key" not in listed[0] and listed[0]["prefix"] == k["prefix"]
    # use the key on a data endpoint → free-tier headers, last_used_at + hashed ip recorded
    r = web.get("/v1/status", headers={"Authorization": f"Bearer {k['key']}", "X-Forwarded-For": "203.0.113.7"})
    assert r.status_code == 200 and r.headers["X-RateLimit-Limit-Requests"] == "120"
    listed = web.get("/v1/me/keys").json()["data"]
    assert listed[0]["last_used_at"] and len(listed[0]["last_used_ip"]) == 16
    # A6: the key may READ the account but not mutate it
    anon = TestClient(web.app)
    r = anon.get("/v1/me", headers={"Authorization": f"Bearer {k['key']}"})
    assert r.status_code == 200 and r.json()["data"]["auth"] == "key"
    assert anon.get("/v1/me/usage?range=24h", headers={"Authorization": f"Bearer {k['key']}"}).status_code == 200
    assert anon.get("/v1/me/limits", headers={"Authorization": f"Bearer {k['key']}"}).status_code == 200
    r = anon.post(f"/v1/me/keys/{k['id']}/rotate", json={}, headers={**JSON, "Authorization": f"Bearer {k['key']}"})
    assert r.status_code == 403 and r.json()["error"]["code"] == "SESSION_REQUIRED"
    assert anon.get("/v1/me/keys", headers={"Authorization": f"Bearer {k['key']}"}).status_code == 403
    # rotate (session) → new key keeps name/note/expiry, old one dead
    r = web.post(f"/v1/me/keys/{k['id']}/rotate", json={}, headers=JSON)
    assert r.status_code == 201
    k2 = r.json()["data"]
    assert k2["key"] != k["key"] and k2["rotated_from"] == k["id"] and k2["name"] == "laptop" and k2["note"] == "CI runner"
    assert k2["expires_at"] == k["expires_at"]
    r = web.get("/v1/status", headers={"Authorization": f"Bearer {k['key']}"})
    assert r.status_code == 401 and r.json()["error"]["code"] == "INVALID_API_KEY"
    # A3: revoke with an EMPTY body + X-Requested-With (what the SPA sends) is accepted
    r = web.delete(f"/v1/me/keys/{k2['id']}", headers={"X-Requested-With": "hfmd"})
    assert r.status_code == 200 and r.json()["data"]["status"] == "revoked"
    assert web.get("/v1/status", headers={"Authorization": f"Bearer {k2['key']}"}).status_code == 401
    # usage endpoint (session)
    u = web.get("/v1/me/usage?range=24h").json()["data"]
    assert u["totals"]["requests"] >= 1 and len(u["principals"]) == 3 and any(p.startswith("user:") for p in u["principals"]) and u["key_id"] is None
    assert {"requests", "rows", "status_429", "bytes", "rows_parquet"} <= set(u["points"][0])
    # logout with empty body + header, cookie gone
    r = web.post("/v1/auth/logout", headers={"X-Requested-With": "hfmd"})
    assert r.status_code == 200
    assert web.get("/v1/me").status_code == 401


def test_signup_never_touches_existing_account_and_is_uniform(web, make_user, outbox):
    """A4 (pre-hijack) + A10 (enumeration): signup on an existing address changes nothing and answers 202."""
    from accounts import security
    from accounts.models import User
    from core.db import session
    uid, _, email, pw = make_user()
    with session() as s:
        before = (s.get(User, uid).password_hash, s.get(User, uid).name)
    r = web.post("/v1/auth/signup", json={"email": email, "name": "Mallory", "password": "attacker-passphrase"}, headers=JSON)
    assert r.status_code == 202 and r.json()["data"]["status"] == "verification_sent" and "debug_link" not in r.json()["data"]
    with session() as s:
        u = s.get(User, uid)
        assert (u.password_hash, u.name) == before
        assert security.verify_password(pw, u.password_hash) and not security.verify_password("attacker-passphrase", u.password_hash)
    assert [m.kind for m in outbox] == ["already_registered"] and outbox[0].to == email
    # unverified signup: a second signup re-sends the verification but keeps the ORIGINAL password
    email2 = fresh_email("unverified")
    web.post("/v1/auth/signup", json={"email": email2, "name": "Real", "password": "the-real-passphrase"}, headers=JSON)
    r = web.post("/v1/auth/signup", json={"email": email2, "name": "Mallory", "password": "attacker-passphrase"}, headers=JSON)
    assert r.status_code == 202 and "/verify?token=" in r.json()["data"]["debug_link"]
    with session() as s:
        u = s.execute(__import__("sqlalchemy").select(User).where(User.email == email2)).scalars().first()
        assert u.name == "Real" and security.verify_password("the-real-passphrase", u.password_hash)
    # validation errors are still explicit
    r = web.post("/v1/auth/signup", json={"email": fresh_email("w"), "name": "x", "password": "short"}, headers=JSON)
    assert r.status_code == 400 and r.json()["error"]["code"] == "WEAK_PASSWORD"
    r = web.post("/v1/auth/signup", json={"email": "not-an-email", "name": "x", "password": "long-enough-password"}, headers=JSON)
    assert r.status_code == 400 and r.json()["error"]["code"] == "INVALID_PARAMETER"


def test_csrf_guard(web, make_user, signin):
    """A3: JSON always OK; empty body needs X-Requested-With / Sec-Fetch-Site / same Origin; forms are refused."""
    _, _, email, pw = make_user()
    signin(web, email, pw)
    r = web.post("/v1/me/keys", content="name=x", headers={"Content-Type": "application/x-www-form-urlencoded"})
    assert r.status_code == 415 and r.json()["error"]["code"] == "UNSUPPORTED_MEDIA_TYPE"
    r = web.post("/v1/auth/logout", headers={"Content-Type": "text/plain"}, content="x")
    assert r.status_code == 415
    # bodiless POST without any proof → 415 (a cross-site <form> with no fields looks like this)
    assert web.post("/v1/me/sessions/revoke-all", headers={"Origin": "https://evil.example"}).status_code == 415
    # bodiless with browser proofs → accepted
    assert web.post("/v1/me/sessions/revoke-all", headers={"Sec-Fetch-Site": "same-origin"}).status_code == 200
    assert web.post("/v1/me/sessions/revoke-all", headers={"Origin": "http://testserver"}).status_code == 200
    assert web.post("/v1/me/sessions/revoke-all", headers={"X-Requested-With": "hfmd"}).status_code == 200


def test_login_errors_constant_cost_and_lockout(web, make_user, outbox):
    """A5 (lockout), A10 (dummy hash), A15 (password length), A22 (login_failed audit)."""
    from accounts import security
    from accounts.models import AuditLog, User
    from core.db import session
    from sqlalchemy import select
    uid, _, email, pw = make_user()
    assert security.verify_password("anything", None) is False            # dummy-hash path returns False
    r = web.post("/v1/auth/login", json={"email": "ghost@example.com", "password": pw}, headers=JSON)
    assert r.status_code == 401 and r.json()["error"]["code"] == "INVALID_CREDENTIALS"
    assert web.post("/v1/auth/login", json={"email": email, "password": "x" * 300}, headers=JSON).status_code == 422
    for i in range(4):
        r = web.post("/v1/auth/login", json={"email": email, "password": "wrong-password"}, headers=JSON)
        assert r.status_code == 401, i
    r = web.post("/v1/auth/login", json={"email": email, "password": "wrong-password"}, headers=JSON)   # 5th → lock 30 s
    assert r.status_code == 423 and r.json()["error"]["code"] == "ACCOUNT_LOCKED" and 1 <= int(r.headers["Retry-After"]) <= 30
    r = web.post("/v1/auth/login", json={"email": email, "password": pw}, headers=JSON)                 # even the right password
    assert r.status_code == 423
    with session() as s:
        u = s.get(User, uid)
        assert u.failed_logins == 5 and u.locked_until is not None
        fails = list(s.execute(select(AuditLog).where(AuditLog.action == "user.login_failed", AuditLog.target == f"user:{uid}")).scalars())
        assert len(fails) == 5 and '"ip"' in fails[-1].meta
        u.locked_until = None
        u.failed_logins = 9
    r = web.post("/v1/auth/login", json={"email": email, "password": "wrong-password"}, headers=JSON)   # 10th → warning mail, 1 h lock
    assert r.status_code == 423 and int(r.headers["Retry-After"]) > 600
    assert any(m.kind == "suspicious_login" and m.to == email for m in outbox)
    with session() as s:
        s.get(User, uid).locked_until = None
    r = web.post("/v1/auth/login", json={"email": email, "password": pw}, headers=JSON)
    assert r.status_code == 200
    with session() as s:
        assert s.get(User, uid).failed_logins == 0
    assert web.get("/v1/me").status_code == 200


def test_forgot_and_reset_revokes_sessions_and_keys(web, make_user, signin):
    """A4/A11: a password reset signs out every session and (by default) revokes the keys."""
    _, key, email, pw = make_user()
    other = TestClient(web.app)
    signin(other, email, pw)                              # a second browser
    assert other.get("/v1/me").status_code == 200
    # unknown address: same 202, no link
    r = web.post("/v1/auth/forgot", json={"email": "unknown@example.com"}, headers=JSON)
    assert r.status_code == 202 and "debug_link" not in r.json()["data"]
    r = web.post("/v1/auth/forgot", json={"email": email}, headers=JSON)
    assert r.status_code == 202
    link = r.json()["data"]["debug_link"]
    assert "/reset-password?token=" in link
    r = web.post("/v1/auth/reset", json={"token": token_of(link), "password": "short"}, headers=JSON)
    assert r.status_code == 400 and r.json()["error"]["code"] == "WEAK_PASSWORD"
    r = web.post("/v1/auth/reset", json={"token": token_of(link), "password": "brand-new-passphrase"}, headers=JSON)
    assert r.status_code == 200 and r.json()["data"]["keys_revoked"] == 1
    assert other.get("/v1/me").status_code == 401                      # old session dead
    assert web.get("/v1/status", headers={"Authorization": f"Bearer {key}"}).status_code == 401   # key revoked
    web.cookies.clear()
    r = web.post("/v1/auth/login", json={"email": email, "password": "brand-new-passphrase"}, headers=JSON)
    assert r.status_code == 200
    web.cookies.clear()
    r = web.post("/v1/auth/reset", json={"token": token_of(link), "password": "brand-new-passphrase"}, headers=JSON)
    assert r.status_code == 400 and r.json()["error"]["code"] == "INVALID_TOKEN"
    # revoke_keys=false keeps the keys
    _, key2, email2, _ = make_user()
    link2 = web.post("/v1/auth/forgot", json={"email": email2}, headers=JSON).json()["data"]["debug_link"]
    r = web.post("/v1/auth/reset", json={"token": token_of(link2), "password": "another-new-passphrase", "revoke_keys": False}, headers=JSON)
    assert r.status_code == 200 and r.json()["data"]["keys_revoked"] == 0
    assert web.get("/v1/status", headers={"Authorization": f"Bearer {key2}"}).status_code == 200


def test_forgot_per_email_throttle(web, make_user):
    _, _, email, _ = make_user(with_key=False)
    for _ in range(5):
        assert web.post("/v1/auth/forgot", json={"email": email}, headers=JSON).status_code == 202
    r = web.post("/v1/auth/forgot", json={"email": email}, headers=JSON)
    assert r.status_code == 429 and "Retry-After" in r.headers


def test_key_limit_and_validation(web, make_user, signin):
    """A18: cap enforced (count checked after insert too); expiry / scopes validated."""
    _, key, email, pw = make_user()
    signin(web, email, pw)
    for i in range(9):
        assert web.post("/v1/me/keys", json={"name": f"k{i}"}, headers=JSON).status_code == 201
    r = web.post("/v1/me/keys", json={"name": "one-too-many"}, headers=JSON)
    assert r.status_code == 409 and r.json()["error"]["code"] == "KEY_LIMIT_REACHED"
    r = web.delete("/v1/me/keys/999999", headers=JSON)
    assert r.status_code == 404 and r.json()["error"]["code"] == "KEY_NOT_FOUND"
    r = web.post("/v1/me/keys", json={"name": "bad", "scopes": ["admin"]}, headers=JSON)
    assert r.status_code == 400 and r.json()["error"]["code"] == "INVALID_PARAMETER"
    assert web.post("/v1/me/keys", json={"name": "bad", "expires_in_days": 0}, headers=JSON).status_code == 422
    # rename + note
    kid = web.get("/v1/me/keys").json()["data"][0]["id"]
    r = web.patch(f"/v1/me/keys/{kid}", json={"name": "renamed", "note": "moved to the NAS"}, headers=JSON)
    assert r.status_code == 200 and r.json()["data"]["name"] == "renamed" and r.json()["data"]["note"] == "moved to the NAS"


def test_expired_key_is_refused(web, make_user, signin):
    from datetime import timedelta

    from accounts.models import ApiKey
    from core.db import session
    _, _, email, pw = make_user(with_key=False)
    signin(web, email, pw)
    k = web.post("/v1/me/keys", json={"name": "short-lived", "expires_in_days": 1}, headers=JSON).json()["data"]
    assert web.get("/v1/status", headers={"Authorization": f"Bearer {k['key']}"}).status_code == 200
    with session() as s:
        s.get(ApiKey, k["id"]).expires_at = s.get(ApiKey, k["id"]).created_at - timedelta(seconds=1)
    from ratelimit.middleware import invalidate_key_cache
    invalidate_key_cache()
    r = web.get("/v1/status", headers={"Authorization": f"Bearer {k['key']}"})
    assert r.status_code == 401 and "expired" in r.json()["error"]["message"]
    assert web.get("/v1/me/keys").json()["data"][0]["expired"] is True


def test_disabled_user_key_is_refused(client, make_user):
    uid, key, _, _ = make_user()
    from accounts import service
    from accounts.models import User
    from core.db import session
    with session() as s:
        service.update_user(s, s.get(User, uid), actor="test", status="disabled")
    r = client.get("/v1/status", headers={"Authorization": f"Bearer {key}"})
    assert r.status_code == 403 and r.json()["error"]["code"] == "ACCOUNT_DISABLED"


def test_disabled_user_cannot_reset_or_accept_invite(web, make_user):
    """A17."""
    from accounts import service
    from accounts.models import User
    from core.db import session
    uid, _, email, _ = make_user(with_key=False)
    link = web.post("/v1/auth/forgot", json={"email": email}, headers=JSON).json()["data"]["debug_link"]
    with session() as s:
        u = s.get(User, uid)
        service.update_user(s, u, actor="test", status="disabled")
        inv = service.issue_token(s, u, "invite", actor="test").link
    r = web.post("/v1/auth/reset", json={"token": token_of(link), "password": "brand-new-passphrase"}, headers=JSON)
    assert r.status_code == 403 and r.json()["error"]["code"] == "ACCOUNT_DISABLED"
    r = web.post("/v1/auth/accept-invite", json={"token": token_of(inv), "password": "brand-new-passphrase"}, headers=JSON)
    assert r.status_code == 403
    assert web.post("/v1/auth/forgot", json={"email": email}, headers=JSON).json()["data"].get("debug_link") is None


def test_session_version_revoke_all_and_disable(web, make_user, signin):
    """A11: bumping session_version kills every cookie; the current browser gets a fresh one."""
    from accounts import service
    from accounts.models import User
    from core.db import session
    uid, _, email, pw = make_user()
    signin(web, email, pw)
    old_cookie = web.cookies.get("hfmd_session")
    r = web.post("/v1/me/sessions/revoke-all", json={}, headers=JSON)
    assert r.status_code == 200 and r.json()["data"]["status"] == "sessions_revoked"
    assert web.cookies.get("hfmd_session") != old_cookie
    assert web.get("/v1/me").status_code == 200                                  # this browser stays signed in
    stale = TestClient(web.app)
    assert stale.get("/v1/me", headers={"Cookie": f"hfmd_session={old_cookie}"}).status_code == 401   # A11
    # disabling the account (admin/CLI) signs it out too
    with session() as s:
        service.update_user(s, s.get(User, uid), actor="test", status="disabled")
    assert web.get("/v1/me").status_code == 401


def test_change_password_and_email(web, make_user, signin, outbox):
    """B2 + A22: password change (with mail + sessions), e-mail change via confirmation link."""
    from accounts.models import AuditLog
    from core.db import session
    from sqlalchemy import select
    uid, _, email, pw = make_user()
    signin(web, email, pw)
    other = TestClient(web.app)
    signin(other, email, pw)
    r = web.post("/v1/me/password", json={"current_password": "nope-nope-nope", "new_password": "a-brand-new-passphrase"}, headers=JSON)
    assert r.status_code == 401 and r.json()["error"]["code"] == "INVALID_CREDENTIALS"
    r = web.post("/v1/me/password", json={"current_password": pw, "new_password": "short"}, headers=JSON)
    assert r.status_code == 400
    r = web.post("/v1/me/password", json={"current_password": pw, "new_password": "a-brand-new-passphrase"}, headers=JSON)
    assert r.status_code == 200 and web.get("/v1/me").status_code == 200 and other.get("/v1/me").status_code == 401
    assert any(m.kind == "password_changed" for m in outbox)
    # e-mail change: link goes to the NEW address; the account keeps the old one until confirmation
    new_email = fresh_email("new")
    r = web.post("/v1/me/email", json={"new_email": new_email, "password": "a-brand-new-passphrase"}, headers=JSON)
    assert r.status_code == 202 and r.json()["data"]["status"] == "confirmation_sent"
    link = r.json()["data"]["debug_link"]
    mail = [m for m in outbox if m.kind == "email_change"][0]
    assert mail.to == new_email and "/verify?token=" in link
    assert web.get("/v1/me").json()["data"]["user"]["email"] == email
    r = web.post("/v1/auth/verify", json={"token": token_of(link)}, headers=JSON)
    assert r.status_code == 200 and r.json()["data"]["kind"] == "email_change" and r.json()["data"]["email"] == new_email
    assert any(m.kind == "email_changed" and m.to == email for m in outbox)
    # the address of another user: uniform 202, that user is notified, nothing changes at confirmation time
    _, _, taken, _ = make_user(with_key=False)
    r = web.post("/v1/me/email", json={"new_email": taken, "password": "a-brand-new-passphrase"}, headers=JSON)
    assert r.status_code == 202 and "debug_link" not in r.json()["data"]
    assert any(m.kind == "already_registered" and m.to == taken for m in outbox)
    with session() as s:
        actions = {a for (a,) in s.execute(select(AuditLog.action).where(AuditLog.target == f"user:{uid}")).all()}
    assert {"account.password_change", "account.email_change", "account.email_change_requested"} <= actions


def test_delete_account(web, make_user, signin, outbox):
    """B2: soft delete — keys revoked, sessions dead, e-mail anonymised, login impossible."""
    from accounts.models import User
    from core.db import session
    uid, key, email, pw = make_user()
    signin(web, email, pw)
    r = web.request("DELETE", "/v1/me", json={"password": "wrong-password-here"}, headers=JSON)
    assert r.status_code == 401
    r = web.request("DELETE", "/v1/me", json={"password": pw}, headers=JSON)
    assert r.status_code == 200 and r.json()["data"]["status"] == "deleted"
    assert web.get("/v1/me").status_code == 401
    assert web.get("/v1/status", headers={"Authorization": f"Bearer {key}"}).status_code == 401
    with session() as s:
        u = s.get(User, uid)
        assert u.status == "deleted" and u.email == f"deleted-{uid}@deleted.invalid" and u.password_hash is None and u.deleted_at
    r = web.post("/v1/auth/login", json={"email": email, "password": pw}, headers=JSON)
    assert r.status_code == 401
    assert any(m.kind == "account_deleted" and m.to == email for m in outbox)
    # the address can sign up again
    assert web.post("/v1/auth/signup", json={"email": email, "name": "Again", "password": "yet-another-passphrase"}, headers=JSON).status_code == 202


def test_profile_and_quota_alert_opt_out(web, make_user, signin):
    _, _, email, pw = make_user(with_key=False)
    signin(web, email, pw)
    r = web.patch("/v1/me", json={"name": "Ada L.", "quota_alerts": False}, headers=JSON)
    assert r.status_code == 200 and r.json()["data"]["name"] == "Ada L." and r.json()["data"]["quota_alerts"] is False
    assert web.get("/v1/me").json()["data"]["user"]["quota_alerts"] is False


def test_usage_per_key_csv_and_live_limits(web, make_user, signin):
    """B1/B5: per-key usage, CSV export, live remaining quota per key."""
    _, _, email, pw = make_user(with_key=False)
    signin(web, email, pw)
    k1 = web.post("/v1/me/keys", json={"name": "one"}, headers=JSON).json()["data"]
    k2 = web.post("/v1/me/keys", json={"name": "two"}, headers=JSON).json()["data"]
    for _ in range(3):
        assert web.get("/v1/_test/frame?n=100", headers={"Authorization": f"Bearer {k1['key']}"}).status_code == 200
    assert web.get("/v1/_test/frame?n=50", headers={"Authorization": f"Bearer {k2['key']}"}).status_code == 200
    all_ = web.get("/v1/me/usage?range=24h").json()["data"]
    one = web.get(f"/v1/me/usage?range=24h&key_id={k1['id']}").json()["data"]
    assert all_["totals"]["requests"] == 4 and all_["totals"]["rows"] == 350
    assert one["totals"] == {"requests": 3, "rows": 300, "status_429": 0, "bytes": one["totals"]["bytes"], "rows_parquet": 0}
    assert one["key_id"] == k1["id"] and one["principals"] == [f"key:{k1['id']}"]
    assert web.get("/v1/me/usage?range=24h&key_id=999999").status_code == 404
    r = web.get(f"/v1/me/usage.csv?range=7d&key_id={k1['id']}")
    assert r.status_code == 200 and r.headers["content-type"].startswith("text/csv") and r.headers["X-Row-Count"] == "168"
    lines = r.text.strip().splitlines()
    assert lines[0] == "t,requests,rows,status_429,bytes,rows_parquet" and len(lines) == 169
    assert any(l.split(",")[1] == "3" for l in lines[1:])
    lim = web.get("/v1/me/limits").json()["data"]
    assert lim["tier"]["tier"] == "free" and len(lim["keys"]) == 2
    by_id = {x["key_id"]: x for x in lim["keys"]}
    assert by_id[k1["id"]]["requests"]["remaining"] == 117 and by_id[k1["id"]]["rows"]["remaining"] == 1_000_000 - 300
    assert by_id[k2["id"]]["requests"]["remaining"] == 119 and by_id[k1["id"]]["redis"] is True
    only = web.get(f"/v1/me/limits?key_id={k2['id']}").json()["data"]["keys"]
    assert len(only) == 1 and only[0]["name"] == "two"


def test_mailer_templates_never_leak_keys(app):
    from accounts import mailer
    full_key = "hfmd_live_ab12cd34EXAMPLEKEYnotARealOne00"
    for kind in ("verify", "invite", "reset", "email_change"):
        subject, text, html = mailer.render(kind, name="Ada", link="https://x/y?token=abc", new_email="n@example.com")
        assert subject and "HF Market Data" in text and '<a href="https://x/y?token=abc"' in html
    for kind in ("key_created", "key_revoked", "key_rotated"):
        subject, text, html = mailer.render(kind, name="Ada", key_name="k", key_prefix="hfmd_live_ab12cd34", new_prefix="hfmd_live_zz", key=full_key)
        assert "hfmd_live_ab12cd34" in text and full_key not in text and full_key not in html   # prefix only, never the key
    for kind in ("already_registered", "password_changed", "email_changed", "suspicious_login", "account_deleted", "quota_alert"):
        assert mailer.render(kind, name="Ada", new_email="n@example.com", what="Rows quota at 80 %", detail="…")[0]
    d = mailer.send("verify", "ada@example.com", name="Ada", link="https://x/y?token=abc")
    assert d.delivered is False and d.error == "no_provider"


def test_mailer_never_logs_links(app, caplog):
    """A9: the action link must not end up in the logs."""
    import logging

    from accounts import mailer
    with caplog.at_level(logging.INFO, logger="hfmarketdata.mailer"):
        mailer.send("verify", "ada@example.com", name="Ada", link="https://x/verify?token=SECRET-TOKEN")
    assert "SECRET-TOKEN" not in caplog.text and "ada@example.com" not in caplog.text and "a***@example.com" in caplog.text


def test_mailer_posts_to_resend(app, monkeypatch):
    from dataclasses import replace

    import respx
    from httpx import Response

    from accounts import mailer
    monkeypatch.setattr(mailer, "settings", replace(mailer.settings, resend_api_key="re_test_key"))
    with respx.mock(assert_all_called=True) as mock:
        route = mock.post(mailer.RESEND_URL).mock(return_value=Response(200, json={"id": "email_123"}))
        d = mailer.send("invite", "tim@example.com", name="Tim", link="https://x/accept-invite?token=t")
    assert d.delivered and d.provider_id == "email_123"
    sent = route.calls[0].request
    assert sent.headers["authorization"] == "Bearer re_test_key"
    import json
    body = json.loads(sent.content)
    assert body["to"] == ["tim@example.com"] and "accept-invite?token=t" in body["text"] and body["html"]


def test_mail_is_sent_after_commit_and_dropped_on_rollback(app, outbox):
    """A10/A13: the outbox is flushed by after_commit, never inside the transaction, and dropped on rollback."""
    from accounts import mailer, service
    from core.db import session
    with session() as s:
        d = mailer.queue(s, "already_registered", "x@example.com", name="X")
        assert d.queued and outbox == []           # nothing dispatched yet: still inside the transaction
    assert [m.to for m in outbox] == ["x@example.com"]
    outbox.clear()
    try:
        with session() as s:
            mailer.queue(s, "already_registered", "y@example.com", name="Y")
            raise RuntimeError("boom")
    except RuntimeError:
        pass
    assert outbox == []
    # a provider failure after commit stores the link for admins
    with session() as s:
        u = service.create_user(s, fresh_email("late"), "Late", actor="test")
        issued = service.issue_token(s, u, "invite", actor="test")
        uid, tok_id = u.id, issued.token.id
        assert issued.delivery.queued and issued.token.meta is None
    msg = [m for m in outbox if m.kind == "invite"][0]
    msg.on_failure(msg.kind, msg.to, msg.ctx, mailer.Delivery(False, error="resend_500"))
    with session() as s:
        from accounts.models import User
        assert service.pending_link(s, s.get(User, uid), "invite") == issued.link