"""B-92 (API-37, API-46, API-47, AUD-41, ARC-02, ARC-09, ARC-61, ELC-29): readiness, Electrum status, audit browsing, the registration row and the logs.""" import json import logging import sys from datetime import datetime, timedelta, timezone import pytest from httpx import ASGITransport, AsyncClient from sqlalchemy import select from tests.unit.test_abuse_limits import ADMIN, _register, client # noqa: F401 (fixture + helper) @pytest.fixture def listener(client): from app.main import app return app.state.electrum_listener # --- /health/ready ----------------------------------------------------------------- def _ready_app(listener, *, header_age, heartbeat_age, connected=True): from types import SimpleNamespace from app.main import app listener.client = object() if connected else None listener.last_header_age_seconds = lambda: header_age app.state.scheduler = SimpleNamespace(heartbeat_age_seconds=lambda: heartbeat_age) async def test_ready_only_when_connected_fresh_and_alive(client, listener): _ready_app(listener, header_age=100, heartbeat_age=5) ok = await client.get("/health/ready") assert ok.status_code == 200 and ok.json() == { "status": "ready", "checks": {"electrum_connected": True, "last_header_age_seconds": 100, "scheduler_heartbeat_age_seconds": 5}, } for kwargs in ( {"header_age": 100, "heartbeat_age": 5, "connected": False}, {"header_age": 1800, "heartbeat_age": 5}, {"header_age": None, "heartbeat_age": 5}, {"header_age": 100, "heartbeat_age": 30}, {"header_age": 100, "heartbeat_age": None}, ): _ready_app(listener, **kwargs) bad = await client.get("/health/ready") assert bad.status_code == 503 and bad.json()["status"] == "not_ready", kwargs async def test_a_360_second_gap_between_blocks_is_not_unreadiness(client, listener): """With 120 s blocks, 360 s of silence happens ~5% of the time by chance.""" _ready_app(listener, header_age=400, heartbeat_age=5) assert (await client.get("/health/ready")).status_code == 200 def test_the_scheduler_beats_on_every_tick(monkeypatch): import asyncio from app.rounds.scheduler import RoundScheduler scheduler = RoundScheduler(None, None) assert scheduler.heartbeat_age_seconds() is None async def one_tick(): async def fake_tick(): raise asyncio.CancelledError scheduler._tick = fake_tick with pytest.raises(asyncio.CancelledError): await scheduler.run() asyncio.run(one_tick()) assert scheduler.heartbeat_age_seconds() == 0 # --- /admin/electrum --------------------------------------------------------------- async def test_the_electrum_status_endpoint(client, listener): from app.electrum.client import ElectrumEndpoint listener._endpoints = [ElectrumEndpoint("a.example", 50002, True), ElectrumEndpoint("b.example", 50002, True)] listener.client = object() listener.tip_height = 538900 listener.last_rotation = {"reason": "stalled_tip", "from": "x:1", "at": "2026-10-18T00:00:00+00:00"} body = (await client.get("/admin/electrum", headers=ADMIN)).json() assert body["connected"] is True and body["endpoint"] == str(listener.current_endpoint) assert body["tip_height"] == 538900 and body["fallback_count"] == 1 and body["quorum_enforced"] is True assert body["last_rotation"]["reason"] == "stalled_tip" listener.client = None assert (await client.get("/admin/electrum", headers=ADMIN)).json()["endpoint"] is None assert (await client.get("/admin/electrum")).status_code == 403 async def test_the_listener_records_why_it_moved_to_another_server(): from app.electrum.listener import ServerStalledError, _rotation_reason from app.rounds.draw import HeaderValidationError assert _rotation_reason(ServerStalledError("x")) == "stalled_tip" assert _rotation_reason(HeaderValidationError("x")) == "invalid_header" assert _rotation_reason(RuntimeError("x")) == "error" # --- /admin/audit-log --------------------------------------------------------------- async def _seed_audit(rows): from app.db import base as db_base from app.db.models import AuditLog async with db_base.AsyncSessionLocal() as session: for event, user_id, round_id, at in rows: session.add(AuditLog(event_type=event, payload_json="{}", user_id=user_id, round_id=round_id, created_at=at)) await session.commit() async def test_the_audit_log_is_filtered_and_paged(client): now = datetime(2026, 10, 1, 12, tzinfo=timezone.utc) await _seed_audit( [(f"event_{i % 3}", i % 2, 10 + i % 2, now + timedelta(hours=i)) for i in range(12)] ) async def ids(**params): r = await client.get("/admin/audit-log", headers=ADMIN, params=params) assert r.status_code == 200, r.text return [e["id"] for e in r.json()] everything = await ids(limit=500) assert everything == sorted(everything, reverse=True) and len(everything) == 12 first = await ids(limit=5) assert first == everything[:5] assert await ids(limit=5, before_id=first[-1]) == everything[5:10] r = await client.get("/admin/audit-log", headers=ADMIN, params=[("event_type", "event_0"), ("event_type", "event_1")]) assert {e["event_type"] for e in r.json()} == {"event_0", "event_1"} assert await ids(event_type="nope") == [] assert len(await ids(round_id=10, user_id=0)) == 6 window = await ids(**{"from": "2026-10-01T14:00:00+00:00", "to": "2026-10-01T17:00:00+00:00"}) assert len(window) == 3 # 14:00, 15:00, 16:00 — `to` is exclusive for bad in ({"from": "2026-10-01T14:00:00"}, {"before_id": "x"}, {"to": "yesterday"}): r = await client.get("/admin/audit-log", headers=ADMIN, params=bad) assert r.status_code == 422 and r.json()["detail"]["code"] == "invalid_request" # --- user_registered ------------------------------------------------------------------- async def test_a_registration_writes_user_registered_in_its_transaction(client): from app.db import base as db_base from app.db.models import AuditLog, User await _register(client, username="carol") async with db_base.AsyncSessionLocal() as session: user = (await session.scalars(select(User))).one() row = (await session.scalars(select(AuditLog).where(AuditLog.event_type == "user_registered"))).one() assert row.user_id == user.id and row.round_id is None assert json.loads(row.payload_json) == {"derivation_index": 0, "address": user.address} assert "carol" not in row.payload_json # --- logs and proxy ----------------------------------------------------------------------- def test_the_app_logs_also_go_to_stdout(monkeypatch, tmp_path): from app import logging_config root = logging.getLogger() before = list(root.handlers) monkeypatch.setattr(logging_config, "LOG_DIR", tmp_path) monkeypatch.setattr(logging_config, "LOG_FILE", tmp_path / "app.log") try: logging_config.setup_logging() added = [h for h in root.handlers if h not in before] assert any(isinstance(h, logging.StreamHandler) and h.stream is sys.stdout for h in added) finally: for h in [h for h in root.handlers if h not in before]: root.removeHandler(h) h.close() def test_uvicorn_trusts_only_the_proxy_and_the_container_is_checked(): from pathlib import Path root = Path(__file__).resolve().parents[2] dockerfile, compose = (root / "Dockerfile").read_text(), (root / "docker-compose.yml").read_text() assert "--proxy-headers" in dockerfile and "--forwarded-allow-ips" in dockerfile assert "FORWARDED_ALLOW_IPS=172.28.0.10" in compose and "ipv4_address: 172.28.0.10" in compose assert "subnet: 172.28.0.0/24" in compose and "/health/ready" in compose