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

A derivation is 64 MiB and ~0.1 s at the current version, 256 MB and ~0.45 s
for a v3 hash not yet rewritten. 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, at 64 MiB as at
256 MB (`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():
    # A v3 (256 MB) hash, the heaviest the hub still verifies: at 64 MiB an
    # inline derivation is ~45 ms and half of it is too close to scheduling
    # noise to tell a stalled loop from a busy machine.
    version = 3
    pw_hash, salt = auth.hash_password("k" * 44, version)
    started = time.perf_counter()
    auth.verify_password("k" * 44, pw_hash, salt, 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, 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