plm-lottery/tests/unit/test_admin.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, "admin_token", "test-admin-token")
    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 create_async_engine

    from app.db import base as db_base

    db_base.engine = create_async_engine(settings.database_url)
    from sqlalchemy.ext.asyncio import async_sessionmaker

    db_base.AsyncSessionLocal = async_sessionmaker(db_base.engine, expire_on_commit=False)

    # app.db.session did `from app.db.base import AsyncSessionLocal` at its own
    # first import, which only copies the reference as it was at that moment —
    # reassigning db_base.AsyncSessionLocal above doesn't reach it. get_session()
    # looks up its module global at call time, so rebinding it here (every test)
    # keeps it pointed at *this* test's engine instead of whichever ran first.
    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 app.api.routes.admin import router as admin_router
    from fastapi import FastAPI

    app = FastAPI()
    app.include_router(admin_router)

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

    await db_base.engine.dispose()


async def test_admin_requires_token(client):
    resp = await client.get("/admin/config")
    assert resp.status_code == 403


async def test_admin_rejects_wrong_token(client):
    resp = await client.get("/admin/config", headers={"X-Admin-Token": "wrong"})
    assert resp.status_code == 403


async def test_admin_reads_and_updates_config(client):
    headers = {"X-Admin-Token": "test-admin-token"}

    resp = await client.get("/admin/config", headers=headers)
    assert resp.status_code == 200
    assert resp.json()["fee_address"] == ""

    resp = await client.put(
        "/admin/config", headers=headers, json={"fee_address": "plm1qfeeaddress", "bet_amount_sats": 500_000_000}
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["fee_address"] == "plm1qfeeaddress"
    assert body["bet_amount_sats"] == 500_000_000

    resp = await client.get("/admin/config", headers=headers)
    assert resp.json()["fee_address"] == "plm1qfeeaddress"


async def test_admin_can_pause_and_resume_the_lottery(client):
    headers = {"X-Admin-Token": "test-admin-token"}

    resp = await client.get("/admin/config", headers=headers)
    assert resp.json()["paused"] is False

    resp = await client.post("/admin/pause", headers=headers)
    assert resp.status_code == 200
    assert resp.json()["paused"] is True

    resp = await client.get("/admin/config", headers=headers)
    assert resp.json()["paused"] is True

    resp = await client.post("/admin/resume", headers=headers)
    assert resp.status_code == 200
    assert resp.json()["paused"] is False


async def test_admin_pause_requires_token(client):
    resp = await client.post("/admin/pause")
    assert resp.status_code == 403


async def test_admin_lists_users(client):
    from app.db import base as db_base
    from app.db.models import User
    from app.wallet.hd import derive_user_address

    async with db_base.AsyncSessionLocal() as session:
        session.add(
            User(
                username="alice",
                password_hash="unused",
                derivation_index=0,
                address=derive_user_address(0),
                cached_balance_sats=1_000_000_000,
            )
        )
        await session.commit()

    headers = {"X-Admin-Token": "test-admin-token"}
    resp = await client.get("/admin/users", headers=headers)
    assert resp.status_code == 200
    body = resp.json()
    assert len(body) == 1
    assert body[0]["username"] == "alice"
    assert body[0]["balance_sats"] == 1_000_000_000


async def test_admin_exports_user_privkey(client):
    from embit.ec import PrivateKey

    from app.db import base as db_base
    from app.db.models import User
    from app.wallet.hd import derive_user_address, derive_user_key
    from app.wallet.plm_network import PLM_MAINNET

    async with db_base.AsyncSessionLocal() as session:
        user = User(
            username="bob",
            password_hash="unused",
            derivation_index=1,
            address=derive_user_address(1),
        )
        session.add(user)
        await session.commit()
        await session.refresh(user)
        user_id = user.id

    headers = {"X-Admin-Token": "test-admin-token"}
    resp = await client.get(f"/admin/users/{user_id}/privkey", headers=headers)
    assert resp.status_code == 200
    body = resp.json()

    expected_wif = PrivateKey(derive_user_key(1).secret, compressed=True, network=PLM_MAINNET).wif(
        network=PLM_MAINNET
    )
    assert body["wif"] == expected_wif
    assert body["address"] == derive_user_address(1)


async def test_admin_privkey_404_for_unknown_user(client):
    headers = {"X-Admin-Token": "test-admin-token"}
    resp = await client.get("/admin/users/999/privkey", headers=headers)
    assert resp.status_code == 404


async def test_admin_resets_user_password(client):
    from app.auth.security import hash_password, verify_password
    from app.db import base as db_base
    from app.db.models import User
    from app.wallet.hd import derive_user_address

    old_hash = hash_password("original-password")
    async with db_base.AsyncSessionLocal() as session:
        user = User(
            username="carol",
            password_hash=old_hash,
            derivation_index=2,
            address=derive_user_address(2),
        )
        session.add(user)
        await session.commit()
        await session.refresh(user)
        user_id = user.id

    headers = {"X-Admin-Token": "test-admin-token"}
    resp = await client.post(f"/admin/users/{user_id}/reset-password", headers=headers)
    assert resp.status_code == 200
    body = resp.json()
    assert body["username"] == "carol"
    new_password = body["new_password"]
    assert new_password and new_password != "original-password"

    async with db_base.AsyncSessionLocal() as session:
        refreshed = await session.get(User, user_id)
        assert refreshed.password_hash != old_hash
        assert verify_password(new_password, refreshed.password_hash)
        assert not verify_password("original-password", refreshed.password_hash)


async def test_admin_reset_password_requires_token(client):
    resp = await client.post("/admin/users/1/reset-password")
    assert resp.status_code == 403


async def test_admin_reset_password_404_for_unknown_user(client):
    headers = {"X-Admin-Token": "test-admin-token"}
    resp = await client.post("/admin/users/999/reset-password", headers=headers)
    assert resp.status_code == 404