plm-lottery/tests/unit/test_bug_reports.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 async_sessionmaker, create_async_engine

    from app.db import base as db_base

    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.admin import router as admin_router
    from app.api.routes.bug_reports import router as bug_reports_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(bug_reports_router)
    app.include_router(admin_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()


_ADMIN_HEADERS = {"X-Admin-Token": "test-admin-token"}


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_anonymous_bug_report_has_no_user(client):
    resp = await client.post("/bug-reports", json={"description": "the bet button does nothing"})
    assert resp.status_code == 201

    resp = await client.get("/admin/bug-reports", headers=_ADMIN_HEADERS)
    assert resp.status_code == 200
    reports = resp.json()
    assert len(reports) == 1
    assert reports[0]["description"] == "the bet button does nothing"
    assert reports[0]["user_id"] is None
    assert reports[0]["username"] is None
    assert reports[0]["status"] == "open"


async def test_logged_in_bug_report_is_attributed_to_the_user(client):
    token = await _register(client)

    resp = await client.post(
        "/bug-reports",
        headers={"Authorization": f"Bearer {token}"},
        json={"description": "withdrawal amount looks wrong", "contact": "alice@example.com"},
    )
    assert resp.status_code == 201

    resp = await client.get("/admin/bug-reports", headers=_ADMIN_HEADERS)
    reports = resp.json()
    assert reports[0]["username"] == "alice"
    assert reports[0]["contact"] == "alice@example.com"


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

    resp = await client.post("/bug-reports", headers=headers, json={"description": "some bug"})
    report_id = resp.json()["id"]

    resp = await client.get("/bug-reports/mine", headers=headers)
    assert resp.status_code == 200
    reports = resp.json()
    assert len(reports) == 1
    assert reports[0]["id"] == report_id
    assert reports[0]["status"] == "open"

    await client.post(
        f"/admin/bug-reports/{report_id}/status", headers=_ADMIN_HEADERS, json={"status": "read"}
    )
    resp = await client.get("/bug-reports/mine", headers=headers)
    assert resp.json()[0]["status"] == "read"


async def test_bug_reports_mine_requires_auth(client):
    resp = await client.get("/bug-reports/mine")
    assert resp.status_code == 401


async def test_bug_reports_mine_only_returns_own_reports(client):
    alice_token = await _register(client, username="alice")
    bob_token = await _register(client, username="bob", password="bob-password")

    await client.post(
        "/bug-reports", headers={"Authorization": f"Bearer {alice_token}"}, json={"description": "alice's bug"}
    )

    resp = await client.get("/bug-reports/mine", headers={"Authorization": f"Bearer {bob_token}"})
    assert resp.json() == []


async def test_blank_description_is_rejected(client):
    resp = await client.post("/bug-reports", json={"description": "   "})
    assert resp.status_code == 422


async def test_admin_bug_reports_requires_token(client):
    resp = await client.get("/admin/bug-reports")
    assert resp.status_code == 403


async def test_admin_can_move_through_open_read_resolved(client):
    resp = await client.post("/bug-reports", json={"description": "some bug"})
    report_id = resp.json()["id"]

    resp = await client.post(
        f"/admin/bug-reports/{report_id}/status", headers=_ADMIN_HEADERS, json={"status": "read"}
    )
    assert resp.status_code == 200
    assert resp.json()["status"] == "read"

    resp = await client.post(
        f"/admin/bug-reports/{report_id}/status", headers=_ADMIN_HEADERS, json={"status": "resolved"}
    )
    assert resp.status_code == 200
    assert resp.json()["status"] == "resolved"

    resp = await client.post(
        f"/admin/bug-reports/{report_id}/status", headers=_ADMIN_HEADERS, json={"status": "open"}
    )
    assert resp.status_code == 200
    assert resp.json()["status"] == "open"


async def test_update_status_rejects_unknown_value(client):
    resp = await client.post("/bug-reports", json={"description": "some bug"})
    report_id = resp.json()["id"]

    resp = await client.post(
        f"/admin/bug-reports/{report_id}/status", headers=_ADMIN_HEADERS, json={"status": "bogus"}
    )
    assert resp.status_code == 422


async def test_update_status_unknown_report_is_404(client):
    resp = await client.post(
        "/admin/bug-reports/999/status", headers=_ADMIN_HEADERS, json={"status": "read"}
    )
    assert resp.status_code == 404