"""
Argon2 runs off the event loop, and never two at a time.

One derivation is 256 MB and a quarter to half a second of CPU. On the loop it
stopped the whole hub for that long at every sign-in. In a thread pool it would
have been worse: two concurrent `lanes=4` derivations deadlock inside OpenSSL
and never return (`auth._argon2_executor`). These pin both halves.
"""

import asyncio
import base64
import pathlib
import re
import time

import pytest
from meshbay_hub import auth

SRC = pathlib.Path(__file__).resolve().parents[1] / "src" / "meshbay_hub"


def test_nothing_derives_argon2_on_the_event_loop():
    """Every line of every module, not the first match: a call added above an
    existing one must not hide behind it."""
    direct = re.compile(r"(?<![\w.])(hash_password|verify_password)\s*\(")
    offenders = []
    for path in SRC.rglob("*.py"):
        if path.name == "auth.py":
            continue
        for n, line in enumerate(path.read_text().splitlines(), 1):
            if direct.search(line):
                offenders.append(f"{path.relative_to(SRC)}:{n}: {line.strip()}")
    assert not offenders, (
        "these derive Argon2 on the calling thread; use the *_off_loop versions:\n"
        + "\n".join(offenders))


def test_the_executor_has_exactly_one_worker():
    """More than one lets two `lanes=4` derivations overlap, and they deadlock."""
    assert auth._argon2_executor._max_workers == 1


@pytest.mark.asyncio
async def test_concurrent_derivations_all_return():
    pw_hash, salt = auth.hash_password("k" * 44)
    results = await asyncio.wait_for(asyncio.gather(*[
        auth.verify_password_off_loop("k" * 44, pw_hash, salt, auth.current_pw_version())
        for _ in range(4)]), timeout=30)
    assert results == [True] * 4


@pytest.mark.asyncio
async def test_the_loop_keeps_turning_while_argon2_runs():
    pw_hash, salt = auth.hash_password("k" * 44)
    started = time.perf_counter()
    auth.verify_password("k" * 44, pw_hash, salt, auth.current_pw_version())
    inline = time.perf_counter() - started

    gaps, done = [], asyncio.Event()

    async def ticker():
        last = time.perf_counter()
        while not done.is_set():
            await asyncio.sleep(0.005)
            now = time.perf_counter()
            gaps.append(now - last)
            last = now

    task = asyncio.create_task(ticker())
    await asyncio.sleep(0.02)
    assert await auth.verify_password_off_loop(
        "k" * 44, pw_hash, salt, auth.current_pw_version())
    done.set()
    await task
    # Inline, the loop stalls for the whole derivation; off it, for scheduling noise.
    assert max(gaps) < inline / 2, (max(gaps), inline)


@pytest.mark.asyncio
async def test_concurrent_sign_ins_all_complete(client):
    key = base64.b64encode(b"k" * 32).decode()
    names = [f"concurrent{i}" for i in range(4)]
    for name in names:
        r = await client.post("/v1/users/register", json={
            "username": name, "email": f"{name}@example.test", "auth_key": key})
        assert r.status_code == 201, r.text

    responses = await asyncio.wait_for(asyncio.gather(*[
        client.post("/v1/users/login", json={"username": n, "auth_key": key})
        for n in names]), timeout=60)
    assert [r.status_code for r in responses] == [200] * 4