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 # --- B-60: the status change is an admin mutation, so it leaves a trace ---------- async def _audit_entries(event_type: str) -> list: from sqlalchemy import select from app.db.base import AsyncSessionLocal from app.db.models import AuditLog async with AsyncSessionLocal() as session: return ( await session.scalars(select(AuditLog).where(AuditLog.event_type == event_type)) ).all() async def test_status_change_is_audit_logged(client): # B-60 """One shared ADMIN_TOKEN and no per-admin identity means the audit log is the only accountability there is — a report could be silently marked resolved.""" token = await _register(client, username="reporter") resp = await client.post( "/bug-reports", json={"description": "some bug"}, headers={"Authorization": f"Bearer {token}"}, ) report_id = resp.json()["id"] await client.post( f"/admin/bug-reports/{report_id}/status", headers=_ADMIN_HEADERS, json={"status": "resolved"} ) entries = await _audit_entries("bug_report_status_changed") assert len(entries) == 1 import json payload = json.loads(entries[0].payload_json) assert payload == {"report_id": report_id, "from": "open", "to": "resolved"} assert entries[0].user_id is not None # the report's author, so it's traceable both ways async def test_setting_the_status_it_already_has_logs_nothing(client): # B-60 """Same rule as config_updated: an edit that changes nothing isn't an event, or the log fills with noise that hides the real changes.""" resp = await client.post("/bug-reports", json={"description": "some bug"}) report_id = resp.json()["id"] for _ in range(3): await client.post( f"/admin/bug-reports/{report_id}/status", headers=_ADMIN_HEADERS, json={"status": "open"} ) assert await _audit_entries("bug_report_status_changed") == []