"""
Per-account sign-in lockout (`login_throttle.py`).

The per-IP limit bounds one address, and IPv6 gives every subscriber a /64 of
them; an online guess targets an account, so the account is what is counted.
The properties pinned here are the ones that make that safe to ship:

  * the Nth wrong passphrase locks, and a locked account is refused **before**
    the passphrase is checked — the right one is refused too
  * an unknown username locks exactly like a real one, so the 429 says nothing
    the 401 did not (M1)
  * a right passphrase clears the count, and failures age out of the window
  * `change_password` checks the same passphrase, so it counts on the same row
  * the two numbers are the admin's to change, and zero turns it off

What one account costs another through this — a stranger locking your name —
is `test_availability_between_members.py`, because it takes two accounts.
"""

import asyncio
from datetime import UTC, datetime, timedelta

import pytest
from meshbay_hub.api.deps import set_admin_usernames
from meshbay_hub.db.models import LoginThrottle
from meshbay_hub.login_throttle import _key
from sqlalchemy import select, update

RIGHT = "r" * 44
WRONG = "w" * 44


async def _register(client, username, auth_key=RIGHT):
    r = await client.post("/v1/users/register", json={
        "username": username, "email": f"{username}@example.com",
        "auth_key": auth_key})
    assert r.status_code == 201, r.text


async def _login(client, username, auth_key):
    return await client.post("/v1/users/login", json={
        "username": username, "auth_key": auth_key})


async def _fail(client, username, times):
    for _ in range(times):
        r = await _login(client, username, WRONG)
        assert r.status_code == 401, r.text


@pytest.mark.asyncio
async def test_the_fourth_failure_locks_and_the_right_passphrase_is_refused(client):
    await _register(client, "alice_test")
    await _fail(client, "alice_test", 4)

    r = await _login(client, "alice_test", RIGHT)
    assert r.status_code == 429, r.text
    assert r.json()["detail"] == "account_locked"
    # An hour, give or take the time the four failures took.
    assert 3500 <= int(r.headers["retry-after"]) <= 3600


@pytest.mark.asyncio
async def test_an_unknown_name_locks_exactly_like_a_real_one(client):
    """M1: the lockout must not become the enumeration oracle `login` avoids."""
    await _register(client, "bob_test")
    await _fail(client, "bob_test", 4)
    await _fail(client, "nobody-by-this-name", 4)

    real = await _login(client, "bob_test", WRONG)
    ghost = await _login(client, "nobody-by-this-name", WRONG)
    assert (real.status_code, real.json()) == (ghost.status_code, ghost.json())
    assert real.status_code == 429


@pytest.mark.asyncio
async def test_the_right_passphrase_clears_the_count(client):
    await _register(client, "carol_test")
    await _fail(client, "carol_test", 3)
    r = await _login(client, "carol_test", RIGHT)
    assert r.status_code == 200, r.text

    # Three more would have been seven in a row without the reset.
    await _fail(client, "carol_test", 3)
    assert (await _login(client, "carol_test", RIGHT)).status_code == 200


@pytest.mark.asyncio
async def test_a_lockout_ends_when_its_window_does(client, db_session):
    await _register(client, "dave_test")
    await _fail(client, "dave_test", 4)
    assert (await _login(client, "dave_test", RIGHT)).status_code == 429

    await db_session.execute(
        update(LoginThrottle).where(LoginThrottle.key == _key("dave_test"))
        .values(last_failure_at=datetime.now(UTC) - timedelta(minutes=61)))
    await db_session.commit()

    assert (await _login(client, "dave_test", RIGHT)).status_code == 200


@pytest.mark.asyncio
async def test_old_failures_do_not_carry_into_a_new_window(client, db_session):
    await _register(client, "erin_test")
    await _fail(client, "erin_test", 3)
    await db_session.execute(
        update(LoginThrottle).where(LoginThrottle.key == _key("erin_test"))
        .values(last_failure_at=datetime.now(UTC) - timedelta(minutes=61)))
    await db_session.commit()

    # One stale window of three, then one fresh failure: a count of one, not four.
    await _fail(client, "erin_test", 1)
    assert (await _login(client, "erin_test", RIGHT)).status_code == 200


@pytest.mark.asyncio
async def test_a_burst_of_concurrent_guesses_gets_no_more_than_the_limit(client):
    """The attempt is taken before the check, in one statement.

    Read-then-write would let every request in a burst read "no failures yet"
    and be checked. On SQLite writes serialise anyway, so this pins the
    behaviour rather than proving the statement under PostgreSQL's concurrency;
    the statement is an `ON CONFLICT DO UPDATE … WHERE`, which both evaluate
    against the row as locked.
    """
    await _register(client, "frank_test")
    results = await asyncio.gather(*[_login(client, "frank_test", WRONG) for _ in range(10)])
    codes = sorted(r.status_code for r in results)
    assert codes.count(401) == 4, codes
    assert codes.count(429) == 6, codes


@pytest.mark.asyncio
async def test_change_password_counts_on_the_same_row(client):
    """It checks the same passphrase, so it is the same oracle."""
    await _register(client, "grace_test")
    token = (await _login(client, "grace_test", RIGHT)).json()["access_token"]
    auth = {"Authorization": f"Bearer {token}"}

    for _ in range(4):
        r = await client.post("/v1/users/password", headers=auth, json={
            "old_auth_key": WRONG, "new_auth_key": "n" * 44})
        assert r.status_code == 403, r.text

    r = await client.post("/v1/users/password", headers=auth, json={
        "old_auth_key": RIGHT, "new_auth_key": "n" * 44})
    assert r.status_code == 429, r.text
    assert (await _login(client, "grace_test", RIGHT)).status_code == 429


@pytest.mark.asyncio
async def test_a_signed_in_session_is_told_its_own_lockout(client):
    """A passphrase change re-wraps every node's bundle before the hub accepts
    it, so the client must know not to start one the hub would then refuse."""
    await _register(client, "olivia_test")
    token = (await _login(client, "olivia_test", RIGHT)).json()["access_token"]
    auth = {"Authorization": f"Bearer {token}"}
    assert (await client.get("/v1/users/me", headers=auth)).json()["passphrase_locked_for"] == 0

    await _fail(client, "olivia_test", 4)
    left = (await client.get("/v1/users/me", headers=auth)).json()["passphrase_locked_for"]
    assert 3500 <= left <= 3600


@pytest.mark.asyncio
async def test_an_attempt_that_checks_no_passphrase_is_not_counted(client, db_session):
    """A legacy account asked to upgrade has been told nothing about its passphrase."""
    from meshbay_hub.db.models import User

    await _register(client, "heidi_test")
    await db_session.execute(
        update(User).where(User.username == "heidi_test").values(pw_version=2))
    await db_session.commit()

    for _ in range(6):
        r = await _login(client, "heidi_test", RIGHT)
        assert r.status_code == 401 and r.json()["detail"] == "auth_upgrade_required"
    failures = await db_session.scalar(
        select(LoginThrottle.failures).where(LoginThrottle.key == _key("heidi_test")))
    assert not failures


@pytest.mark.asyncio
async def test_the_table_never_holds_what_was_typed(client, db_session):
    """People type passphrases into the username field."""
    await _login(client, "my-secret-passphrase-typed-in-the-wrong-box", WRONG)
    keys = (await db_session.execute(select(LoginThrottle.key))).scalars().all()
    assert keys and all("secret" not in k for k in keys)


# ── The admin's two numbers ──────────────────────────────────────────────────

async def _admin_headers(client, username="root_test"):
    await _register(client, username)
    set_admin_usernames([username])
    token = (await _login(client, username, RIGHT)).json()["access_token"]
    return {"Authorization": f"Bearer {token}"}


@pytest.mark.asyncio
async def test_the_admin_sets_the_limit_and_the_hub_applies_it(client):
    admin = await _admin_headers(client)

    r = await client.get("/v1/admin/settings", headers=admin)
    assert r.json()["login"] == {"max_failures": 4, "lockout_minutes": 60}
    assert r.json()["login_defaults"] == {"max_failures": 4, "lockout_minutes": 60}

    r = await client.patch("/v1/admin/settings", headers=admin,
                           json={"login": {"max_failures": 2, "lockout_minutes": 5}})
    assert r.status_code == 200, r.text
    assert r.json()["login"] == {"max_failures": 2, "lockout_minutes": 5}

    await _register(client, "ivan_test")
    await _fail(client, "ivan_test", 2)
    r = await _login(client, "ivan_test", RIGHT)
    assert r.status_code == 429
    assert int(r.headers["retry-after"]) <= 300


@pytest.mark.asyncio
async def test_zero_failures_turns_the_lockout_off(client):
    admin = await _admin_headers(client)
    await client.patch("/v1/admin/settings", headers=admin,
                       json={"login": {"max_failures": 0}})

    await _register(client, "judy_test")
    await _fail(client, "judy_test", 8)
    assert (await _login(client, "judy_test", RIGHT)).status_code == 200


@pytest.mark.asyncio
async def test_values_are_clamped_and_unknown_keys_refused(client):
    admin = await _admin_headers(client)

    r = await client.patch("/v1/admin/settings", headers=admin,
                           json={"login": {"max_failures": -3, "lockout_minutes": 10**9}})
    assert r.status_code == 200
    low, _ = r.json()["login_bounds"]["max_failures"]
    _, high = r.json()["login_bounds"]["lockout_minutes"]
    assert r.json()["login"] == {"max_failures": low, "lockout_minutes": high}

    r = await client.patch("/v1/admin/settings", headers=admin,
                           json={"login": {"lockout_hours": 1}})
    assert r.status_code == 422


@pytest.mark.asyncio
async def test_only_an_admin_changes_them(client):
    await _admin_headers(client)          # an admin exists; this is someone else
    await _register(client, "mallory_test")
    token = (await _login(client, "mallory_test", RIGHT)).json()["access_token"]
    r = await client.patch("/v1/admin/settings",
                           headers={"Authorization": f"Bearer {token}"},
                           json={"login": {"max_failures": 0}})
    assert r.status_code == 403