plm-lottery/tests/unit/test_users.py

import pytest
from cryptography.fernet import Fernet
from httpx import ASGITransport, AsyncClient

from app.config import settings


@pytest.fixture
async def client(monkeypatch, tmp_path):
    monkeypatch.setattr(settings, "database_url", f"sqlite+aiosqlite:///{tmp_path}/test.db")
    monkeypatch.setattr(settings, "jwt_secret", "test-jwt-secret")
    monkeypatch.setattr(settings, "xprv_encryption_key", Fernet.generate_key().decode())
    monkeypatch.setattr(settings, "master_key_path", str(tmp_path / "master.xprv.enc"))

    import app.wallet.hd as hd

    hd._account_key = None
    hd.generate_master_key()

    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

    from app.db import base as db_base

    # Import app.db.models before create_all — declarative models only register
    # themselves into Base.metadata when their module is first imported, and
    # nothing else in this fixture happens to trigger that import beforehand.
    import app.db.models  # noqa: F401

    db_base.engine = create_async_engine(settings.database_url)
    db_base.AsyncSessionLocal = async_sessionmaker(db_base.engine, expire_on_commit=False)

    from app.db import session as db_session

    db_session.AsyncSessionLocal = db_base.AsyncSessionLocal

    async with db_base.engine.begin() as conn:
        await conn.run_sync(db_base.Base.metadata.create_all)

    from fastapi import FastAPI

    from app.api.routes.users import router as users_router
    from app.auth.routes import router as auth_router
    from app.electrum.listener import ElectrumListener

    app = FastAPI()
    app.include_router(auth_router)
    app.include_router(users_router)
    app.state.electrum_listener = ElectrumListener(lambda endpoint: None, db_base.AsyncSessionLocal)

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

    await db_base.engine.dispose()


async def _register(client, username="alice", password="original-password"):
    resp = await client.post("/auth/register", json={"username": username, "password": password})
    assert resp.status_code == 201
    return resp.json()["access_token"]


async def test_change_password_requires_current_password(client):
    token = await _register(client)
    headers = {"Authorization": f"Bearer {token}"}

    resp = await client.post(
        "/users/me/change-password",
        headers=headers,
        json={"current_password": "wrong-password", "new_password": "brand-new-password"},
    )
    assert resp.status_code == 401


async def test_change_password_updates_login(client):
    token = await _register(client)
    headers = {"Authorization": f"Bearer {token}"}

    resp = await client.post(
        "/users/me/change-password",
        headers=headers,
        json={"current_password": "original-password", "new_password": "brand-new-password"},
    )
    assert resp.status_code == 200
    assert resp.json()["access_token"]

    resp = await client.post("/auth/login", json={"username": "alice", "password": "original-password"})
    assert resp.status_code == 401

    resp = await client.post("/auth/login", json={"username": "alice", "password": "brand-new-password"})
    assert resp.status_code == 200


async def test_change_password_invalidates_the_old_token_but_not_the_new_one(client):
    """B-34: neither self-service change-password nor the admin reset used to
    invalidate already-issued JWTs, so a stolen token (or an attacker who
    already had the old password) stayed logged in until the token's natural
    24h expiry — even past a password change meant to lock them out."""
    old_token = await _register(client)
    old_headers = {"Authorization": f"Bearer {old_token}"}

    resp = await client.post(
        "/users/me/change-password",
        headers=old_headers,
        json={"current_password": "original-password", "new_password": "brand-new-password"},
    )
    assert resp.status_code == 200
    new_token = resp.json()["access_token"]
    assert new_token != old_token

    # The old token (what an attacker holding the old password would still
    # have) is now rejected...
    resp = await client.get("/users/me", headers=old_headers)
    assert resp.status_code == 401
    assert resp.json()["detail"]["code"] == "session_expired"

    # ...but the freshly issued one keeps this same session working, so the
    # user who just changed their own password isn't logged out too.
    new_headers = {"Authorization": f"Bearer {new_token}"}
    resp = await client.get("/users/me", headers=new_headers)
    assert resp.status_code == 200


async def test_change_password_rejects_too_short(client):
    token = await _register(client)
    headers = {"Authorization": f"Bearer {token}"}

    resp = await client.post(
        "/users/me/change-password",
        headers=headers,
        json={"current_password": "original-password", "new_password": "short"},
    )
    assert resp.status_code == 400


async def test_change_password_requires_auth(client):
    resp = await client.post(
        "/users/me/change-password",
        json={"current_password": "x", "new_password": "brand-new-password"},
    )
    assert resp.status_code in (401, 403)


@pytest.mark.parametrize(
    "payload",
    [
        {"username": "", "password": "longenough1"},
        {"username": "ab", "password": "longenough1"},          # under 3 chars
        {"username": "bad user!", "password": "longenough1"},   # disallowed characters
        {"username": "validname", "password": "short"},         # under MIN_PASSWORD_LENGTH
        {"username": "validname", "password": ""},
    ],
)
async def test_register_rejects_weak_credentials(client, payload):
    """B-12: registration accepted an empty username and a one-character password,
    while /users/me/change-password demanded 8 — an odd place to be lenient on a
    custodial system holding real funds."""
    resp = await client.post("/auth/register", json=payload)
    assert resp.status_code == 422


async def test_register_accepts_valid_credentials(client):
    resp = await client.post("/auth/register", json={"username": "goodname", "password": "longenough1"})
    assert resp.status_code == 201


async def test_me_created_at_is_utc_stamped(client):
    """B-35: SQLite/aiosqlite returns DateTime columns as naive, even though every
    value written is UTC (app.db.models.utcnow). A bare .isoformat() on that naive
    value has no "Z"/offset, and JavaScript's `new Date()` then parses it as local
    time instead of UTC."""
    token = await _register(client)
    headers = {"Authorization": f"Bearer {token}"}

    resp = await client.get("/users/me", headers=headers)
    assert resp.status_code == 200
    created_at = resp.json()["created_at"]
    assert created_at.endswith("+00:00") or created_at.endswith("Z")