plm-lottery/tests/unit/test_pause_reason.py

"""B-78 (DAT-69, AUD-11, AUD-12, API-32): why the lottery is paused.

`paused_reason` is `operator` for POST /admin/pause and an automatic reason when the
system pauses itself (`resurrected_tx` so far). An operator pause never overwrites an
automatic reason, and POST /admin/resume — the only way out — clears whichever it is and
logs what it lifted.
"""

import json

from sqlalchemy import select

from app.db.models import AuditLog
from app.rounds.pause import pause_automatically
from tests.unit.test_admin import client  # noqa: F401  (fixture)

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


async def _audit(event_type: str) -> list[dict]:
    from app.db import base as db_base

    async with db_base.AsyncSessionLocal() as session:
        rows = (await session.scalars(select(AuditLog).where(AuditLog.event_type == event_type).order_by(AuditLog.id))).all()
        return [json.loads(r.payload_json) for r in rows]


async def test_an_operator_pause_has_the_reason_operator_and_may_carry_a_note(client):
    assert (await client.get("/admin/config", headers=_HEADERS)).json()["paused_reason"] is None

    response = await client.post("/admin/pause", headers=_HEADERS, json={"note": "migrating the server"})

    assert response.status_code == 200
    assert response.json()["paused"] is True and response.json()["paused_reason"] == "operator"
    assert await _audit("lottery_paused") == [{"reason": "operator", "note": "migrating the server"}]


async def test_pause_still_works_without_a_body_and_rejects_a_note_over_256_characters(client):
    assert (await client.post("/admin/pause", headers=_HEADERS)).status_code == 200
    assert (await client.post("/admin/pause", headers=_HEADERS, json={"note": "x" * 257})).status_code == 422
    assert await _audit("lottery_paused") == [{"reason": "operator"}]


async def test_an_automatic_reason_survives_an_operator_pause_and_resume_clears_it(client):
    from app.db import base as db_base

    async with db_base.AsyncSessionLocal() as session:
        await pause_automatically(session, "resurrected_tx")
        await session.commit()
    assert (await client.get("/admin/config", headers=_HEADERS)).json()["paused_reason"] == "resurrected_tx"

    again = await client.post("/admin/pause", headers=_HEADERS)
    assert again.json()["paused_reason"] == "resurrected_tx"  # the cause that needs a look is not hidden
    assert (await _audit("lottery_paused"))[-1] == {"reason": "operator"}  # but the row says who asked

    resumed = await client.post("/admin/resume", headers=_HEADERS)
    assert resumed.json()["paused"] is False and resumed.json()["paused_reason"] is None
    assert (await _audit("lottery_resumed"))[-1] == {"reason": "resurrected_tx"}  # what was lifted


async def test_resuming_a_lottery_that_was_not_paused_logs_a_null_reason(client):
    await client.post("/admin/resume", headers=_HEADERS)
    assert await _audit("lottery_resumed") == [{"reason": None}]


async def test_an_automatic_pause_overwrites_an_operator_reason(client):
    """DAT-69: the automatic reason is always written, even over an active pause."""
    from app.db import base as db_base

    await client.post("/admin/pause", headers=_HEADERS)
    async with db_base.AsyncSessionLocal() as session:
        await pause_automatically(session, "resurrected_tx")
        await session.commit()
    assert (await client.get("/admin/config", headers=_HEADERS)).json()["paused_reason"] == "resurrected_tx"


async def test_only_automatic_reasons_can_be_written_by_the_system(client):
    import pytest
    from app.db import base as db_base

    async with db_base.AsyncSessionLocal() as session:
        with pytest.raises(ValueError):
            await pause_automatically(session, "operator")


async def test_the_public_round_endpoint_reports_the_reason(client):
    from fastapi import FastAPI
    from httpx import ASGITransport, AsyncClient

    from app.api.routes.rounds import router as rounds_router
    from app.rounds.events import broadcaster  # noqa: F401

    app = FastAPI()
    app.include_router(rounds_router)
    app.state.electrum_listener = type("L", (), {"tip_height": 0, "client": None})()
    await client.post("/admin/pause", headers=_HEADERS)
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t") as public:
        body = (await public.get("/rounds/current")).json()
    assert body["lottery_paused"] is True and body["paused_reason"] == "operator"