plm-lottery/tests/unit/test_security.py

import asyncio

from app.auth import security


def test_password_hash_roundtrip():
    hashed = security.hash_password("s3cret!")
    assert security.verify_password("s3cret!", hashed)
    assert not security.verify_password("wrong", hashed)


def test_jwt_roundtrip(monkeypatch):
    monkeypatch.setattr(security.settings, "jwt_secret", "test-secret")
    token = security.create_access_token(user_id=42, token_version=3)
    assert security.decode_access_token(token) == (42, 3)


def test_jwt_decode_defaults_token_version_for_tokens_issued_before_it_existed(monkeypatch):
    """B-34: a token minted before the "tv" claim existed has no such key at
    all. It must still decode — as token_version 0, matching a freshly
    migrated user's starting value — rather than raising or being treated as
    permanently stale."""
    import jwt as pyjwt

    monkeypatch.setattr(security.settings, "jwt_secret", "test-secret")
    payload = {"sub": "42"}
    token = pyjwt.encode(payload, "test-secret", algorithm=security.settings.jwt_algorithm)
    assert security.decode_access_token(token) == (42, 0)


def test_verify_password_returns_false_for_an_unparseable_hash():
    """B-13: only VerifyMismatchError was caught, so a corrupted stored hash raised
    InvalidHashError and became an unhandled 500 on the login endpoint instead of a
    plain "wrong credentials" 401."""
    from app.auth.security import verify_password

    assert verify_password("whatever", "not-an-argon2-hash") is False
    assert verify_password("whatever", "") is False


def test_verify_password_still_rejects_a_wrong_password():
    from app.auth.security import hash_password, verify_password

    stored = hash_password("correct-horse-battery")
    assert verify_password("correct-horse-battery", stored) is True
    assert verify_password("wrong", stored) is False


# --- B-55: Argon2 must not run on the event loop --------------------------------


async def _count_loop_ticks_during(coro) -> tuple[object, int]:
    """Runs `coro` while a heartbeat task tries to run as often as the event loop
    lets it. A blocking call starves the heartbeat completely; a threadpooled one
    leaves the loop free the whole time."""
    ticks = 0

    async def heartbeat() -> None:
        nonlocal ticks
        while True:
            ticks += 1
            await asyncio.sleep(0)

    task = asyncio.create_task(heartbeat())
    await asyncio.sleep(0)  # let the heartbeat reach its loop before timing starts
    try:
        result = await coro
    finally:
        task.cancel()
    return result, ticks


async def test_hash_password_async_keeps_the_event_loop_free():
    """Argon2 costs tens of milliseconds of CPU by design. Run inline from an async
    handler it froze the whole process for that long — every other request plus all
    six background tasks (scheduler, confirmation poller, RBF bumper, listener, both
    reconcilers) — which made a burst of unauthenticated login attempts a cheap way
    to delay draws and confirmations."""
    hashed, ticks = await _count_loop_ticks_during(security.hash_password_async("s3cret-passphrase"))

    assert security.verify_password("s3cret-passphrase", hashed)
    assert ticks > 1  # the loop kept running while the hashing happened


async def test_verify_password_async_keeps_the_event_loop_free():
    stored = security.hash_password("correct-horse-battery")

    ok, ticks = await _count_loop_ticks_during(
        security.verify_password_async("correct-horse-battery", stored)
    )

    assert ok is True
    assert ticks > 1


async def test_verify_password_async_rejects_a_wrong_password():
    """Same answers as the synchronous function it wraps — including the B-13
    unparseable-hash case, which must read as "wrong password", not as an error."""
    stored = security.hash_password("correct-horse-battery")

    assert await security.verify_password_async("wrong", stored) is False
    assert await security.verify_password_async("whatever", "not-an-argon2-hash") is False