"""Sign-in: the cookie, the round trip, the uniform refusal, the lockout, the tokens.

Rules pinned here (accounts plan §4.3, §4.4): the cookie is `__Host-dfp_session; Secure` in
production and `dfp_session` without Secure when APP_ENV is anything else, never keyed on
the request scheme; every refused login is the same 401 with no Set-Cookie, whether the
name exists or not; the per-account lock is silent and the address throttle is the only
429; a token never sits in a route template and never reaches a log line; a GET with a
plain session writes nothing. In-process on SQLite, no network; the argon2 cost is real.
"""

import logging
import pathlib
import re

import pytest
from fastapi.routing import APIRoute
from sqlalchemy import event, select

from app.config import settings
from app.main import app
from app.models import Account, AccountCredential, AuthSession
from app.routers import auth as auth_router
from app.services import directory, passwords, sessions
from tests import _accounts as T
from tests.kit import _env

WEB = pathlib.Path(__file__).resolve().parents[1] / "web"


@pytest.fixture
def world(monkeypatch):
    T.fresh(monkeypatch)
    T.person("rian")
    T.person("adam", level="admin")
    T.person("newbie", password=None, status="invited")
    T.person("gone", status="disabled")
    yield


class FakeClock:
    """Both clocks the sign-in reads: the wall clock (locks, tokens) and the monotonic one
    (the throttles), advanced together."""

    def __init__(self, monkeypatch):
        from datetime import UTC, datetime

        self.wall = datetime(2026, 9, 10, 12, 0, tzinfo=UTC)
        self.mono = 1000.0
        monkeypatch.setattr(sessions, "utcnow", lambda: self.wall)
        monkeypatch.setattr(auth_router, "_mono", lambda: self.mono)

    def advance(self, seconds: float) -> None:
        from datetime import timedelta

        self.wall += timedelta(seconds=seconds)
        self.mono += seconds


class TestCookie:
    def test_production_cookie_is_host_prefixed_and_secure(self, world):
        c = T.client()
        r = T.login(c, "rian")
        assert r.status_code == 200 and r.json() == {"ok": True, "username": "rian", "must_change_password": False}
        cookie = r.headers["set-cookie"]
        assert cookie.startswith("__Host-dfp_session=")
        for flag in ("HttpOnly", "Secure", "SameSite=lax", "Path=/", "Max-Age=2592000"):
            assert flag in cookie, flag
        assert "Domain=" not in cookie

    def test_dev_cookie_is_plain_and_not_secure_whatever_the_scheme(self, world, monkeypatch):
        monkeypatch.setattr(settings, "app_env", "development")
        c = T.client()
        r = T.login(c, "rian")  # over https, and still not Secure: the flag follows APP_ENV
        cookie = r.headers["set-cookie"]
        assert cookie.startswith("dfp_session=") and "Secure" not in cookie and "HttpOnly" in cookie

    def test_the_cookie_is_an_opaque_token_hashed_at_rest(self, world):
        c = T.client()
        token = T.login(c, "rian").cookies["__Host-dfp_session"]
        with _env.TestSessionLocal() as db:
            rows = db.scalars(select(AuthSession)).all()
        assert len(rows) == 1 and rows[0].token_hash == sessions.hash_token(token)
        assert token not in rows[0].token_hash and len(token) >= 40


class TestRoundTrip:
    def test_login_heartbeat_logout(self, world):
        c = T.client()
        assert c.post("/api/auth/heartbeat").status_code == 401
        T.login(c, "rian")
        assert c.post("/api/auth/heartbeat").json() == {"ok": True}
        r = c.post("/api/auth/logout")
        assert r.status_code == 200 and "__Host-dfp_session=" in r.headers["set-cookie"]
        assert c.post("/api/auth/heartbeat").status_code == 401
        assert T.audit_actions() == ["login.ok", "logout"]

    def test_email_is_a_login_alias(self, world):
        assert T.login(T.client(), "RIAN@example.com").status_code == 200

    def test_a_revoked_or_expired_session_is_no_session(self, world, monkeypatch):
        clock = FakeClock(monkeypatch)
        c = T.client()
        T.login(c, "rian")
        clock.advance(sessions.SESSION_SECONDS + 1)
        assert c.post("/api/auth/heartbeat").status_code == 401  # absolute expiry, no sliding
        c2 = T.client()
        T.login(c2, "rian")
        sessions.revoke_all(reason="test")
        assert c2.post("/api/auth/heartbeat").status_code == 401

    def test_wrong_password_is_401_with_no_cookie(self, world):
        r = T.login(T.client(), "rian", "not the password")
        assert r.status_code == 401
        assert r.json()["detail"] == {"error_code": "LOGIN_REFUSED", "summary": "Wrong username or password."}
        assert "set-cookie" not in r.headers

    def test_change_password_needs_the_current_one_and_signs_other_sessions_out(self, world):
        c1, c2 = T.client(), T.client()
        T.login(c1, "rian")
        T.login(c2, "rian")
        r = c1.post("/api/auth/change-password", json={"current_password": "wrong wrong wrong", "new_password": "a brand new passphrase"})
        assert r.status_code == 401
        r = c1.post("/api/auth/change-password", json={"current_password": T.PASSWORD, "new_password": "short"})
        assert r.status_code == 422 and r.json()["detail"]["error_code"] == "PASSWORD_TOO_SHORT"
        r = c1.post("/api/auth/change-password", json={"current_password": T.PASSWORD, "new_password": "a brand new passphrase"})
        assert r.status_code == 200
        assert c1.post("/api/auth/heartbeat").status_code == 200  # the session that asked survives
        assert c2.post("/api/auth/heartbeat").status_code == 401  # every other one is out
        assert T.login(T.client(), "rian", "a brand new passphrase").status_code == 200


class TestUniformRefusal:
    def _sequence(self, monkeypatch, username: str, address: str) -> list[tuple]:
        c = T.client(**{"CF-Connecting-IP": address})
        out = []
        for _ in range(12):
            r = T.login(c, username, "wrong every time")
            out.append((r.status_code, r.json(), r.headers.get("set-cookie"), r.headers.get("retry-after")))
        return out

    def test_a_real_and_a_nonexistent_username_answer_identically(self, world, monkeypatch):
        FakeClock(monkeypatch)
        real = self._sequence(monkeypatch, "rian", "10.0.0.1")
        fake = self._sequence(monkeypatch, "nobody-here", "10.0.0.2")
        assert real == fake
        assert [s for s, *_ in real] == [401] * 10 + [429, 429]
        assert real[-1][3] == auth_router.RETRY_AFTER
        assert all(cookie is None for _, _, cookie, _ in real)

    def test_every_non_verifiable_case_spends_a_dummy_verify(self, world, monkeypatch):
        calls = []
        monkeypatch.setattr(passwords, "dummy_verify", lambda offered: calls.append(offered))
        c = T.client()
        for name in ("nobody", "newbie", "gone"):
            assert T.login(c, name).status_code == 401
        assert len(calls) == 3
        # ...and the audit trail names the reason but never the typed unknown name.
        with _env.TestSessionLocal() as db:
            from app.models import AuditLog

            rows = db.scalars(select(AuditLog).order_by(AuditLog.id)).all()
        assert [r.detail["reason"] for r in rows] == ["unknown", "invited", "disabled"]
        assert rows[0].entity_key is None and "nobody" not in str(rows[0].detail)


class TestLockout:
    def test_eight_failures_lock_silently_for_fifteen_minutes(self, world, monkeypatch):
        clock = FakeClock(monkeypatch)
        c = T.client()
        for _ in range(8):
            assert T.login(c, "rian", "wrong").status_code == 401
        r = T.login(c, "rian")  # the right password, while locked: the same refusal
        assert r.status_code == 401 and r.json()["detail"]["error_code"] == "LOGIN_REFUSED"
        assert T.audit_actions().count("login.locked") == 1
        clock.advance(15 * 60 + 1)
        assert T.login(c, "rian").status_code == 200

    def test_failures_outside_the_window_do_not_add_up(self, world, monkeypatch):
        clock = FakeClock(monkeypatch)
        c = T.client(**{"CF-Connecting-IP": "10.9.9.9"})
        for _ in range(7):
            T.login(c, "rian", "wrong")
        clock.advance(16 * 60)
        for _ in range(2):
            T.login(c, "rian", "wrong")  # 1 and 2 of a fresh window
        assert T.login(c, "rian").status_code == 200

    def test_success_and_unlock_reset_the_counter(self, world, monkeypatch):
        FakeClock(monkeypatch)
        c = T.client()
        for _ in range(8):
            T.login(c, "rian", "wrong")
        with _env.TestSessionLocal() as db:
            directory.unlock(db, db.scalar(select(Account).where(Account.username == "rian")))
        assert T.login(c, "rian").status_code == 200
        with _env.TestSessionLocal() as db:
            cred = db.scalar(select(AccountCredential))
            assert cred.failed_logins == 0 and cred.locked_until is None


class TestTokens:
    def _mint(self, username: str, mode: str) -> str:
        with _env.TestSessionLocal() as db:
            account = db.scalar(select(Account).where(Account.username == username))
            token = directory.mint_token(db, account, mode)
            db.commit()
        return token

    def test_welcome_state_machine(self, world, monkeypatch):
        clock = FakeClock(monkeypatch)
        token = self._mint("newbie", "invite")
        c = T.client()
        assert c.post("/api/auth/token-state", json={"token": token}).json() == {"state": "valid"}
        assert c.post("/api/auth/token-state", json={"token": "nope"}).json() == {"state": "unknown"}
        r = c.post("/api/auth/welcome", json={"token": token, "password": "newbie is in the password"})
        assert r.status_code == 422 and r.json()["detail"]["error_code"] == "PASSWORD_CONTAINS_IDENTITY"
        assert c.post("/api/auth/token-state", json={"token": token}).json() == {"state": "valid"}  # a refused rule spends nothing
        r = c.post("/api/auth/welcome", json={"token": token, "password": "a first real passphrase"})
        assert r.status_code == 200 and r.json()["username"] == "newbie"
        assert "__Host-dfp_session=" in r.headers["set-cookie"]
        assert c.post("/api/auth/token-state", json={"token": token}).json() == {"state": "used"}
        r = c.post("/api/auth/welcome", json={"token": token, "password": "another passphrase here"})
        assert r.status_code == 400 and r.json()["detail"]["error_code"] == "TOKEN_USED"
        with _env.TestSessionLocal() as db:
            assert db.scalar(select(Account.status).where(Account.username == "newbie")) == "active"
        assert T.login(T.client(), "newbie", "a first real passphrase").status_code == 200
        # A reset token is not a welcome token, and an expired one says so.
        reset = self._mint("rian", "reset")
        r = c.post("/api/auth/welcome", json={"token": reset, "password": "another passphrase here"})
        assert r.status_code == 400
        clock.advance(directory.TOKEN_SECONDS + 1)
        assert c.post("/api/auth/token-state", json={"token": reset}).json() == {"state": "expired_resendable"}

    def test_reset_signs_every_other_session_out(self, world):
        old = T.client()
        T.login(old, "rian")
        token = self._mint("rian", "reset")
        c = T.client()
        r = c.post("/api/auth/reset", json={"token": token, "password": "a replacement passphrase"})
        assert r.status_code == 200
        assert old.post("/api/auth/heartbeat").status_code == 401
        assert c.post("/api/auth/heartbeat").status_code == 200
        assert "token.redeem" in T.audit_actions() and "password.set" in T.audit_actions()

    def test_forgot_always_answers_the_same_and_mails_only_a_real_address(self, world):
        c = T.client()
        for identifier in ("rian", "RIAN@example.com", "nobody", "gone", "newbie"):
            assert c.post("/api/auth/forgot", json={"identifier": identifier}).json() == {"ok": True}
        # rian twice, newbie once (invited accounts may reset: it re-sends a way in)
        assert [to for to, _, _ in T.RECORDER.sent] == ["rian@example.com", "rian@example.com", "newbie@example.com"]
        for _, _, text in T.RECORDER.sent:
            assert "/reset#" in text and "?" not in text.split("/reset#", 1)[1].split()[0]
        # Five per address per fifteen minutes: the sixth from this address is the one 429.
        r = c.post("/api/auth/forgot", json={"identifier": "rian"})
        assert r.status_code == 429 and r.json()["detail"]["error_code"] == "LOGIN_LOCKED_OUT"
        # Three tokens per account per day, then silence, and still the same answer.
        other = T.client(**{"CF-Connecting-IP": "10.7.7.7"})
        for _ in range(3):
            assert other.post("/api/auth/forgot", json={"identifier": "rian"}).json() == {"ok": True}
        assert len([1 for to, _, _ in T.RECORDER.sent if to == "rian@example.com"]) == 3

    def test_no_auth_route_and_no_page_carries_a_token_segment(self):
        for route in app.routes:
            if isinstance(route, APIRoute) and route.path.startswith("/api/auth"):
                assert "{" not in route.path, route.path
        src = (WEB / "src" / "App.tsx").read_text()
        assert not re.search(r'path="/(welcome|reset)/', src)

    def test_a_redeem_leaves_no_token_in_any_log_line(self, world, caplog):
        token = self._mint("rian", "reset")
        with caplog.at_level(logging.DEBUG):
            r = T.client().post("/api/auth/reset", json={"token": token, "password": "a replacement passphrase"})
        assert r.status_code == 200
        assert not any(token in rec.getMessage() for rec in caplog.records)
        assert not any(token[:12] in rec.getMessage() for rec in caplog.records)


class TestReadsWriteNothing:
    def test_a_get_with_a_plain_session_issues_no_write(self, world):
        c = T.client()
        T.login(c, "rian")
        statements = []

        @event.listens_for(_env.engine, "before_cursor_execute")
        def spy(conn, cursor, statement, parameters, context, executemany):
            statements.append(statement.strip().split(" ", 1)[0].upper())

        try:
            assert c.get("/api/health").status_code == 200
        finally:
            event.remove(_env.engine, "before_cursor_execute", spy)
        assert statements and set(statements) == {"SELECT"}, statements