siftlode/backend/tests/test_debug_sessions.py

"""E16 S1 — debug sessions: Start → reproduce → Stop, the tag on every line, the screenshots.

Four parts:
  * the session's life: one open per user, a second Start joins it, a Stop ends it, the cap ends it
    on its own, the demo account cannot record, nobody reaches another user's session (DB lane);
  * the tag: a request naming the caller's own open session is logged at its level with every line
    tagged, while a header naming anything else is ignored (a real sign-in through `_session_user`);
  * the browser's batches: a session counts as consent to its level, and only its own events keep
    its tag (DB lane);
  * screenshots: judged by their bytes, capped in size and count, served back only to their owner.
"""

import contextvars
import json
import logging
from datetime import datetime, timedelta, timezone

import httpx
import pytest
from fastapi import BackgroundTasks
from sqlalchemy import func, select

from app import debugsession, logs
from app.models import AuditLog, DebugScreenshot, DebugSession, LogEvent, User
from app.routes import diagnostics as diagnostics_routes

_PW = "correct horse battery"
_PNG = b"\x89PNG\r\n\x1a\n" + b"\x00" * 32
_JPEG = b"\xff\xd8\xff\xe0" + b"\x00" * 32
_WEBP = b"RIFF\x00\x00\x00\x00WEBPVP8 " + b"\x00" * 16


def _signed_in(client, make_user, email="rec@example.com", **kw) -> User:
    """A real sign-in, so the auth dependency itself reads `X-Debug-Session` (`as_user` bypasses
    it)."""
    user = make_user(email=email, password=_PW, **kw)
    r = client.post("/auth/password-login", json={"email": email, "password": _PW})
    assert r.status_code == 200, r.text
    return user


def _start(client, level="DEBUG") -> dict:
    r = client.post("/api/debug-sessions", json={"level": level})
    assert r.status_code == 200, r.text
    return r.json()["session"]


# --- the session's life ---------------------------------------------------------------------


def test_start_opens_a_capped_session_that_me_reports(db, client, make_user):
    _signed_in(client, make_user)
    session = _start(client, "TRACE")
    assert session["level"] == "TRACE" and session["ended_at"] is None
    started = datetime.fromisoformat(session["started_at"])
    until = datetime.fromisoformat(session["until"])
    assert until - started == timedelta(minutes=debugsession.MAX_MINUTES)
    assert client.get("/api/me").json()["debug_session"]["id"] == session["id"]


def test_a_second_start_joins_the_running_session(db, make_user, as_user):
    client = as_user(make_user())
    first = _start(client, "DEBUG")
    assert _start(client, "TRACE")["id"] == first["id"]
    assert db.scalar(select(DebugSession).where(DebugSession.id == first["id"])).level == "DEBUG"


def test_only_the_two_raised_levels_can_be_recorded(db, make_user, as_user):
    client = as_user(make_user())
    for bogus in ("INFO", "WARNING", "trace"):
        assert client.post("/api/debug-sessions", json={"level": bogus}).status_code == 422


def test_the_demo_account_cannot_record(db, make_user, as_user):
    r = as_user(make_user(is_demo=True)).post("/api/debug-sessions", json={"level": "DEBUG"})
    assert r.status_code == 403 and r.json()["detail"]["code"] == "demo"


def test_stop_ends_it_and_me_stops_reporting_it(db, client, make_user):
    _signed_in(client, make_user)
    sid = _start(client)["id"]
    r = client.post(f"/api/debug-sessions/{sid}/stop")
    assert r.status_code == 200 and r.json()["session"]["ended_reason"] == "stopped"
    assert client.get("/api/me").json()["debug_session"] is None
    # A second Stop changes nothing.
    again = client.post(f"/api/debug-sessions/{sid}/stop").json()["session"]
    assert again["ended_at"] == r.json()["session"]["ended_at"]


def _expire(db, sid: str) -> None:
    session = db.get(DebugSession, sid)
    session.started_at -= timedelta(hours=2)
    session.until = datetime.now(timezone.utc) - timedelta(minutes=1)
    db.commit()


def test_a_session_past_its_cap_is_over_before_anything_closes_it(db, client, make_user):
    _signed_in(client, make_user)
    sid = _start(client)["id"]
    _expire(db, sid)
    assert client.get("/api/me").json()["debug_session"] is None
    # A late Stop records that it ended on its own, at its cap.
    stopped = client.post(f"/api/debug-sessions/{sid}/stop").json()["session"]
    assert stopped["ended_reason"] == "expired"
    assert stopped["ended_at"] == stopped["until"]


def test_a_start_after_an_expired_session_closes_it_and_opens_a_new_one(db, make_user, as_user):
    client = as_user(make_user())
    old = _start(client)["id"]
    _expire(db, old)
    new = _start(client)["id"]
    assert new != old
    db.expire_all()
    assert db.get(DebugSession, old).ended_reason == "expired"


def test_nobody_reaches_another_users_session(db, make_user, as_user):
    owner, stranger = as_user(make_user()), as_user(make_user())
    sid = _start(owner)["id"]
    shot = owner.post(
        f"/api/debug-sessions/{sid}/screenshots",
        content=_PNG,
        headers={"content-type": "image/png"},
    ).json()["screenshot"]["id"]
    for method, path in (
        ("get", f"/api/debug-sessions/{sid}"),
        ("post", f"/api/debug-sessions/{sid}/stop"),
        ("get", f"/api/debug-sessions/{sid}/events"),
        ("get", f"/api/debug-sessions/{sid}/screenshots"),
        ("get", f"/api/debug-sessions/{sid}/screenshots/{shot}"),
        ("delete", f"/api/debug-sessions/{sid}/screenshots/{shot}"),
        ("get", f"/api/debug-sessions/{sid}/export"),
    ):
        r = getattr(stranger, method)(path)
        assert r.status_code == 404, (method, path, r.status_code)
        assert r.json()["detail"] == {"code": "not_found", "reason": "debug_session"}
    r = stranger.put(f"/api/debug-sessions/{sid}/draft", json={"title": "x", "body": "y"})
    assert r.status_code == 404
    r = stranger.post(
        f"/api/debug-sessions/{sid}/screenshots",
        content=_PNG,
        headers={"content-type": "image/png"},
    )
    assert r.status_code == 404
    assert stranger.get("/api/debug-sessions").json()["items"] == []
    db.expire_all()
    assert db.get(DebugSession, sid).ended_at is None


def test_the_list_is_the_users_own_newest_first(db, make_user, as_user):
    client = as_user(make_user())
    first = _start(client)["id"]
    assert client.post(f"/api/debug-sessions/{first}/stop").status_code == 200
    second = _start(client)["id"]
    assert [s["id"] for s in client.get("/api/debug-sessions").json()["items"]] == [second, first]


def test_the_hourly_job_closes_expired_sessions_and_drops_old_ones(db, make_user):
    user = make_user()
    now = datetime.now(timezone.utc)
    lapsed = DebugSession(
        id="b" * 16,
        user_id=user.id,
        level="DEBUG",
        started_at=now - timedelta(hours=2),
        until=now - timedelta(hours=1),
    )
    db.add(lapsed)
    ancient = DebugSession(
        id="c" * 16,
        user_id=user.id,
        level="DEBUG",
        started_at=now - timedelta(days=30),
        until=now - timedelta(days=30) + timedelta(hours=1),
        ended_at=now - timedelta(days=30),
        ended_reason="stopped",
    )
    db.add(ancient)
    db.flush()  # no ORM relationship orders the inserts: the session row first
    db.add(
        DebugScreenshot(session_id=ancient.id, created_at=now, mime="image/png", size=4, data=b"x")
    )
    db.commit()

    assert debugsession.sweep(db) == 1
    assert debugsession.gc(db, 7) == 1
    db.commit()
    db.expire_all()
    assert db.get(DebugSession, lapsed.id).ended_reason == "expired"
    assert db.get(DebugSession, lapsed.id).ended_at == lapsed.until
    assert db.get(DebugSession, ancient.id) is None
    assert db.scalars(select(DebugScreenshot)).all() == []


def test_an_erased_account_takes_its_sessions_with_it(db, make_user, as_user):
    from app.auth import purge_user

    user = make_user()
    client = as_user(user)
    sid = _start(client)["id"]
    assert (
        client.post(
            f"/api/debug-sessions/{sid}/screenshots",
            content=_PNG,
            headers={"content-type": "image/png"},
        ).status_code
        == 200
    )
    purge_user(db, user, BackgroundTasks())
    assert db.get(DebugSession, sid) is None
    assert db.scalars(select(DebugScreenshot)).all() == []


# --- the tag --------------------------------------------------------------------------------


class _Capture(logging.Handler):
    def __init__(self):
        super().__init__(level=logging.NOTSET)
        self.records: list[logging.LogRecord] = []
        self.addFilter(logs.ContextFilter())
        self.addFilter(logs.LevelGate())

    def emit(self, record):
        self.records.append(record)


@pytest.fixture
def gated():
    """The process at INFO with a gated handler on the root, as test_diagnostics' fixture."""
    root = logging.getLogger()
    saved = (logs._base_level, dict(logs._elevations), logs._root_applied, root.level)
    handler = _Capture()
    root.addHandler(handler)
    logs._elevations.clear()
    logs.set_level("INFO")
    try:
        yield handler
    finally:
        root.removeHandler(handler)
        logs._base_level, logs._root_applied = saved[0], saved[2]
        logs._elevations.clear()
        logs._elevations.update(saved[1])
        root.setLevel(saved[3])


def _request_lines(handler: _Capture) -> list[logging.LogRecord]:
    return [r for r in handler.records if getattr(r, "event", None) == "http.request"]


def test_a_request_naming_the_own_open_session_is_tagged_and_raised(gated, db, client, make_user):
    _signed_in(client, make_user)
    sid = _start(client, "DEBUG")["id"]
    gated.records.clear()
    assert client.get("/api/debug-sessions", headers={"x-debug-session": sid}).status_code == 200
    (line,) = _request_lines(gated)
    assert getattr(line, "debug_session_id", None) == sid
    # Raised to the session's level: the line is kept for storage though the user's own is Alap.
    assert getattr(line, "elevated", False) is True


def test_a_header_naming_anything_else_is_ignored_not_refused(gated, db, client, make_user):
    other = make_user(email="other@example.com")
    foreign = debugsession.start(db, other, "TRACE").id
    db.commit()
    _signed_in(client, make_user)
    own = _start(client)["id"]
    assert client.post(f"/api/debug-sessions/{own}/stop").status_code == 200
    for value in (foreign, own, "not-a-session", "f" * 16, "A" * 16):
        gated.records.clear()
        # Not `/api/me`: a polled route logs at DEBUG, and an un-raised request would drop the very
        # line this checks.
        r = client.get("/api/debug-sessions", headers={"x-debug-session": value})
        assert r.status_code == 200, value
        (line,) = _request_lines(gated)
        assert getattr(line, "debug_session_id", None) is None, value
        assert getattr(line, "elevated", False) is False, value


def test_a_header_naming_the_own_session_past_its_cap_is_ignored(gated, db, client, make_user):
    _signed_in(client, make_user)
    sid = _start(client)["id"]
    _expire(db, sid)  # past its cap, not yet swept: still open in the table
    gated.records.clear()
    assert client.get("/api/debug-sessions", headers={"x-debug-session": sid}).status_code == 200
    (line,) = _request_lines(gated)
    assert getattr(line, "debug_session_id", None) is None
    assert getattr(line, "elevated", False) is False


def test_a_request_without_the_header_keeps_the_sessions_floor(gated):
    """The browser's own requests (an <img>, a <video>) carry no header: they must not lower the
    root level a concurrent request of the session still needs."""
    root = logging.getLogger()
    until = datetime.now(timezone.utc) + timedelta(minutes=30)
    request = contextvars.copy_context()  # set_user writes the context: keep it this test's own
    request.run(logs.set_user, 5, "TRACE", until, "a" * 16)
    assert root.level == logs.resolve_level("TRACE")
    request.run(logs.set_user, 5, "INFO", None, None)  # the same user, a request without the header
    assert root.level == logs.resolve_level("TRACE")
    logs.end_session_elevation("a" * 16)  # Stop
    assert root.level == logging.INFO


def test_an_erased_account_takes_its_session_elevation_with_it(gated):
    until = datetime.now(timezone.utc) + timedelta(minutes=30)
    request = contextvars.copy_context()
    request.run(logs.set_user, 11, "TRACE", until, "b" * 16)
    assert logging.getLogger().level == logs.resolve_level("TRACE")
    logs.forget_user(11)
    assert logs._elevations == {} and logging.getLogger().level == logging.INFO


def test_a_session_never_lowers_the_users_own_level():
    now = datetime.now(timezone.utc)
    user = User(diagnostics_level="TRACE", diagnostics_until=now + timedelta(hours=3))
    session = DebugSession(level="DEBUG", until=now + timedelta(minutes=30))
    # The user's own level, but never past the recording: the process tracks this pair under the
    # session, and an expiry outliving it would hold the root logger down after it ended.
    assert debugsession.level_for(user, session) == ("TRACE", session.until)
    assert debugsession.level_for(User(), session) == ("DEBUG", session.until)
    assert debugsession.level_for(User(), None) == ("INFO", None)


# --- the browser's batches -------------------------------------------------------------------


def _event(**over) -> dict:
    return {
        "ts": datetime.now(timezone.utc).isoformat(),
        "level": "DEBUG",
        "event": "ui.click",
        "msg": "click",
        "tab_id": "a1b2c3",
        **over,
    }


@pytest.fixture
def batches():
    diagnostics_routes._batches.reset()
    yield
    diagnostics_routes._batches.reset()


def test_a_session_lets_its_own_events_in_at_its_level_and_tags_them(
    batches, db, make_user, as_user
):
    user = make_user()
    client = as_user(user)
    sid = _start(client, "DEBUG")["id"]
    other = debugsession.start(db, make_user(), "DEBUG").id
    db.commit()
    events = [
        _event(debug_session_id=sid),
        _event(debug_session_id=other),  # not this session's: kept, untagged
        _event(),  # captured before Start: kept, untagged
        _event(level="TRACE", debug_session_id=sid),  # under the session's level: dropped
    ]
    r = client.post(
        "/api/diagnostics/events", json={"events": events}, headers={"x-debug-session": sid}
    )
    assert r.json() == {"ok": True, "stored": 3}
    rows = db.scalars(select(LogEvent).order_by(LogEvent.id)).all()
    assert [row.debug_session_id for row in rows] == [sid, None, None]
    assert {row.user_id for row in rows} == {user.id}


def test_without_the_header_a_base_level_user_is_still_refused(batches, db, make_user, as_user):
    client = as_user(make_user())
    sid = _start(client)["id"]
    r = client.post("/api/diagnostics/events", json={"events": [_event(debug_session_id=sid)]})
    assert r.status_code == 409 and r.json()["detail"]["code"] == "diagnostics_off"


def test_the_evidence_is_the_sessions_rows_in_time_order(db, make_user, as_user):
    user = make_user()
    client = as_user(user)
    sid = _start(client)["id"]
    now = datetime.now(timezone.utc)

    def row(ts, source, session, event="x.y", attrs=None):
        return LogEvent(
            ts=ts,
            level="INFO",
            sev=9,
            source=source,
            logger=source,
            event=event,
            msg=source,
            attrs=attrs,
            user_id=user.id,
            debug_session_id=session,
        )

    db.add_all(
        [
            row(now + timedelta(seconds=2), "api", sid),
            row(now + timedelta(seconds=1), "web", sid),
            row(now, "web", None),
            # The recording's own log upload is plumbing, not evidence…
            row(now, "api", sid, "http.request", {"path": debugsession.INGEST_PATH}),
            # …but a server line with no event code and no attrs is evidence all the same.
            row(now + timedelta(seconds=3), "api", sid, None),
        ]
    )
    db.commit()
    body = client.get(f"/api/debug-sessions/{sid}/events").json()
    assert [(i["source"], i["event"]) for i in body["items"]] == [
        ("web", "x.y"),
        ("api", "x.y"),
        ("api", None),
    ]
    assert body["total"] == 3


# --- screenshots ----------------------------------------------------------------------------


@pytest.fixture
def recording(db, make_user, as_user):
    client = as_user(make_user())
    return client, _start(client)["id"]


def _upload(client, sid, data, content_type="image/png"):
    return client.post(
        f"/api/debug-sessions/{sid}/screenshots",
        content=data,
        headers={"content-type": content_type},
    )


@pytest.mark.parametrize(
    ("data", "mime"), [(_PNG, "image/png"), (_JPEG, "image/jpeg"), (_WEBP, "image/webp")]
)
def test_a_screenshot_is_kept_as_what_its_bytes_are(recording, data, mime):
    client, sid = recording
    # The declared type is not trusted: every one of these claims to be a PNG.
    r = _upload(client, sid, data)
    assert r.status_code == 200 and r.json()["screenshot"]["mime"] == mime
    shot = r.json()["screenshot"]["id"]
    got = client.get(f"/api/debug-sessions/{sid}/screenshots/{shot}")
    assert got.content == data
    assert got.headers["content-type"] == mime
    assert got.headers["x-content-type-options"] == "nosniff"


@pytest.mark.parametrize(
    ("data", "reason"),
    [(b"<svg onload=alert(1)>", "screenshot_not_an_image"), (b"", "screenshot_empty")],
)
def test_what_is_not_an_image_is_refused(recording, data, reason):
    client, sid = recording
    r = _upload(client, sid, data)
    assert r.status_code == 422
    assert r.json()["detail"] == {"code": "refused", "reason": reason}


def test_screenshots_are_capped_in_size_and_count(recording, monkeypatch):
    client, sid = recording
    monkeypatch.setattr(debugsession, "SCREENSHOT_MAX_BYTES", len(_PNG) + 1)
    assert _upload(client, sid, _PNG + b"\x00\x00").status_code == 413
    for _ in range(debugsession.SCREENSHOTS_MAX):
        assert _upload(client, sid, _PNG).status_code == 200
    r = _upload(client, sid, _PNG)
    assert r.status_code == 422 and r.json()["detail"] == {
        "code": "refused",
        "reason": "screenshot_too_many",
        "limit": debugsession.SCREENSHOTS_MAX,
    }
    assert (
        client.get(f"/api/debug-sessions/{sid}").json()["screenshots"]
        == debugsession.SCREENSHOTS_MAX
    )


def test_a_sent_session_takes_no_more_screenshots(db, recording):
    client, sid = recording
    db.get(DebugSession, sid).issue_url = "https://git.example/o/r/issues/1"
    db.commit()
    r = _upload(client, sid, _PNG)
    assert r.status_code == 409 and r.json()["detail"]["reason"] == "screenshot_already_sent"


def test_a_screenshot_can_be_removed_and_is_then_gone(recording):
    client, sid = recording
    shot = _upload(client, sid, _PNG).json()["screenshot"]["id"]
    assert client.get(f"/api/debug-sessions/{sid}/screenshots").json()["items"][0]["id"] == shot
    r = client.delete(f"/api/debug-sessions/{sid}/screenshots/{shot}")
    assert r.json() == {"ok": True, "deleted": shot}
    r = client.get(f"/api/debug-sessions/{sid}/screenshots/{shot}")
    assert r.status_code == 404 and r.json()["detail"]["reason"] == "screenshot"


def test_a_half_sent_session_keeps_its_screenshots_a_finished_one_lets_them_go(db, recording):
    client, sid = recording
    shot = _upload(client, sid, _PNG).json()["screenshot"]["id"]
    session = db.get(DebugSession, sid)
    session.issue_url = "https://git.example/o/r/issues/1"
    _progress(session, {"number": 1, "pending": ["server.log"], "links": {}})
    db.commit()
    # A send that still owes files names its screenshots by their order: removing one renumbers
    # what it has left to send.
    r = client.delete(f"/api/debug-sessions/{sid}/screenshots/{shot}")
    assert r.status_code == 409 and r.json()["detail"]["reason"] == "screenshot_already_sent"
    # Finished: the issue has its own copy, and ours is the user's to drop.
    _progress(session, {"number": 1, "pending": [], "links": {}})
    db.commit()
    assert client.delete(f"/api/debug-sessions/{sid}/screenshots/{shot}").status_code == 200


def _progress(session, state):
    """The delivery a finished-or-abandoned send left on the row, with no claim standing on it.
    `set_delivery` is the live send's writer and refuses without one, which is the point of it."""
    session.draft = {**(session.draft or {}), "delivery": state}


def _age_the_claim(session, hours=1):
    """The claim a send holds, as a request that died an hour ago left it."""
    state = debugsession.delivery(session)
    stale = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()
    session.draft = {
        **(session.draft or {}),
        "delivery": {**state, "sending": {**state["sending"], "at": stale}},
    }


def test_a_send_under_way_holds_the_draft_and_the_screenshots_without_blocking_stop(
    gitea_fake, db, recording
):
    client, sid = recording
    shot = _upload(client, sid, _PNG).json()["screenshot"]["id"]
    _ready(client, sid)
    session = db.get(DebugSession, sid)
    debugsession.claim_send(session)
    db.commit()
    r = client.put(f"/api/debug-sessions/{sid}/draft", json={"title": "x", "body": "y"})
    assert r.status_code == 409 and r.json()["detail"]["reason"] == "issue_sending"
    assert _upload(client, sid, _PNG).json()["detail"]["reason"] == "issue_sending"
    # Nor removed: the send names the screenshots by their order, and this would renumber them
    # under it — even before the issue exists, when nothing else would refuse it yet.
    r = client.delete(f"/api/debug-sessions/{sid}/screenshots/{shot}")
    assert r.status_code == 409 and r.json()["detail"]["reason"] == "issue_sending"
    r = client.post(f"/api/debug-sessions/{sid}/issue")
    assert r.status_code == 409 and r.json()["detail"]["reason"] == "issue_sending"
    assert client.get(f"/api/debug-sessions/{sid}").json()["sending"] is True
    # Stop is NOT held: the recording is the user's to end while their report is going out.
    assert client.post(f"/api/debug-sessions/{sid}/stop").status_code == 200
    # A claim left behind by a request that died is taken over by the next send — which then
    # actually sends. Without the take-over the report could never leave, and nothing in the UI
    # would say why.
    _age_the_claim(session)
    db.commit()
    r = client.post(f"/api/debug-sessions/{sid}/issue")
    assert r.status_code == 200 and r.json()["session"]["issue_url"]
    assert client.get(f"/api/debug-sessions/{sid}").json()["sending"] is False


def test_a_slow_send_keeps_its_own_claim_alive_and_gives_back_only_that_one(db, recording):
    """The lease is renewed by the progress the send reports: a send slower than the lease must not
    expire while it is still working, and then overwrite whoever took its place."""
    client, sid = recording
    session = db.get(DebugSession, sid)
    token = debugsession.claim_send(session)
    _age_the_claim(session)
    db.commit()
    assert debugsession.sending(session) is False
    # One more step reported — proof of life, so the claim stands again.
    debugsession.set_delivery(
        db, session, {"number": 7, "pending": ["server.log"], "links": {}}, token
    )
    db.commit()
    assert debugsession.sending(session) is True
    assert debugsession.not_uploaded(session) == ["server.log"]
    # And once a send that aged out HAS been taken over, the old one writes nothing: its progress
    # report is a whole-column overwrite, so without this it put its own dead token back and took
    # the new send's committed links with it. `expire_on_commit` is off, so the old send still has
    # its own token in memory — only the row can say whose claim it is.
    taken = debugsession.claim_send(session)
    db.commit()
    with pytest.raises(debugsession.SendClaimLost):
        debugsession.set_delivery(db, session, {"number": 9, "pending": [], "links": {}}, token)
    assert debugsession.delivery(session)["number"] == 7
    assert debugsession.not_uploaded(session) == ["server.log"]
    # The claim of a send that took over after ours aged out is not ours to hand back.
    debugsession.release_send(session, token)
    assert debugsession.sending(session) is True
    debugsession.release_send(session, taken)
    assert debugsession.sending(session) is False


def test_a_users_uploads_wait_for_each_other_across_their_sessions(db, recording):
    """The cap below is the user's across ALL their sessions, but a session's row lock only bounds
    that session — two uploads to two sessions both read the total before either commits, and both
    pass. The user's own lock is what makes the cap a cap instead of a guess.

    Driven through `add_screenshot`, not through `lock_user`: the primitive works whether or not
    the upload calls it, so a test that called it here stayed green with the call site deleted."""
    from app.db import SessionLocal

    client, sid = recording
    session = db.get(DebugSession, sid)
    debugsession.add_screenshot(db, session, _PNG)  # not committed: the lock is ours until it is
    other = SessionLocal()
    try:
        # A second upload by this user cannot get in to read the total while this one holds it...
        assert other.scalar(select(func.pg_try_advisory_xact_lock(session.user_id))) is False
        # ...and another user is not queued behind them for it.
        assert other.scalar(select(func.pg_try_advisory_xact_lock(session.user_id + 1))) is True
    finally:
        other.rollback()
        other.close()
    db.rollback()


def test_a_users_screenshots_are_capped_across_their_sessions(recording, monkeypatch):
    client, sid = recording
    monkeypatch.setattr(debugsession, "SCREENSHOT_USER_MAX_BYTES", len(_PNG) * 2)
    assert _upload(client, sid, _PNG).status_code == 200
    assert client.post(f"/api/debug-sessions/{sid}/stop").status_code == 200
    other = _start(client)["id"]
    assert _upload(client, other, _PNG).status_code == 200
    r = _upload(client, other, _PNG)
    assert r.status_code == 422 and r.json()["detail"]["reason"] == "screenshot_storage_full"


def test_new_sessions_are_paced_but_joining_one_is_not(db, make_user, as_user, monkeypatch):
    from app.ratelimit import RateLimiter
    from app.routes import debug_sessions as routes

    monkeypatch.setattr(routes, "_new_sessions", RateLimiter(max_events=1, window_seconds=3600))
    client = as_user(make_user())
    sid = _start(client)["id"]
    assert _start(client)["id"] == sid  # another tab joining
    assert client.post(f"/api/debug-sessions/{sid}/stop").status_code == 200
    r = client.post("/api/debug-sessions", json={"level": "DEBUG"})
    assert r.status_code == 429 and r.json()["detail"]["reason"] == "debug_session_rate_limited"


def test_two_starts_at_once_join_the_one_that_won(db, make_user, as_user, monkeypatch):
    user = make_user()
    client = as_user(user)
    won = _start(client)["id"]

    def raced(db_, user_, level):
        # The loser of the race: past the "is one open?" read, its insert meets the index.
        db_.add(
            DebugSession(
                id="0" * 16,
                user_id=user_.id,
                level=level,
                started_at=datetime.now(timezone.utc),
                until=datetime.now(timezone.utc) + timedelta(minutes=5),
            )
        )
        db_.flush()

    monkeypatch.setattr(debugsession, "start", raced)
    r = client.post("/api/debug-sessions", json={"level": "DEBUG"})
    assert r.status_code == 200, r.text
    assert r.json()["session"]["id"] == won


def test_the_export_is_paced(recording, monkeypatch):
    from app.ratelimit import RateLimiter
    from app.routes import debug_sessions as routes

    client, sid = recording
    monkeypatch.setattr(routes, "_exports", RateLimiter(max_events=1, window_seconds=60))
    assert client.get(f"/api/debug-sessions/{sid}/export").status_code == 200
    r = client.get(f"/api/debug-sessions/{sid}/export")
    assert r.status_code == 429 and r.json()["detail"]["reason"] == "export_rate_limited"


def test_the_upload_path_left_out_of_the_evidence_is_the_real_one():
    assert debugsession.INGEST_PATH in {r.path for r in diagnostics_routes.router.routes}


# --- the draft and the export ---------------------------------------------------------------


def test_a_draft_is_kept_and_comes_back_with_the_session(recording):
    client, sid = recording
    draft = {
        "title": "Spinner stays",
        "body": "## Steps\n1. open",
        "excluded": [3],
        "form": {"a": 1},
    }
    r = client.put(f"/api/debug-sessions/{sid}/draft", json=draft)
    assert r.status_code == 200 and r.json()["session"]["draft"] == draft
    assert client.get(f"/api/debug-sessions/{sid}").json()["draft"] == draft


def test_a_draft_is_bounded(recording, monkeypatch):
    from app.routes import debug_sessions as routes

    client, sid = recording
    monkeypatch.setattr(routes, "FORM_MAX_BYTES", 16)
    r = client.put(
        f"/api/debug-sessions/{sid}/draft",
        json={"title": "t", "body": "b", "form": {"text": "x" * 32}},
    )
    assert r.status_code == 413
    too_long = {"title": "t" * 201, "body": "b"}
    assert client.put(f"/api/debug-sessions/{sid}/draft", json=too_long).status_code == 422


def test_a_sent_session_keeps_its_draft(db, recording):
    client, sid = recording
    db.get(DebugSession, sid).issue_url = "https://git.example/o/r/issues/1"
    db.commit()
    r = client.put(f"/api/debug-sessions/{sid}/draft", json={"title": "t", "body": "b"})
    assert r.status_code == 409 and r.json()["detail"]["reason"] == "issue_already_sent"


def test_the_export_is_the_saved_draft_the_evidence_and_the_screenshots(db, make_user, as_user):
    import io
    import json
    import zipfile

    user = make_user()
    client = as_user(user)
    sid = _start(client)["id"]
    now = datetime.now(timezone.utc)

    def row(ts, source, msg):
        return LogEvent(
            ts=ts,
            level="INFO",
            sev=9,
            source=source,
            logger=source,
            event="x.y",
            msg=msg,
            user_id=user.id,
            debug_session_id=sid,
        )

    kept_web, dropped_web, kept_api = (
        row(now, "web", "kept click"),
        row(now + timedelta(seconds=1), "web", "private search"),
        row(now + timedelta(seconds=2), "api", "GET /api/x 200"),
    )
    db.add_all([kept_web, dropped_web, kept_api])
    db.commit()
    _upload(client, sid, _PNG)
    _upload(client, sid, _JPEG)
    assert (
        client.put(
            f"/api/debug-sessions/{sid}/draft",
            json={
                "title": "Spinner stays",
                "body": "It spins.",
                "excluded": [dropped_web.id, 999999],
            },
        ).status_code
        == 200
    )

    r = client.get(f"/api/debug-sessions/{sid}/export")
    assert r.status_code == 200 and r.headers["content-type"] == "application/zip"
    assert "attachment;" in r.headers["content-disposition"]
    z = zipfile.ZipFile(io.BytesIO(r.content))
    assert z.namelist() == [
        "issue.md",
        "client.log",
        "server.log",
        "screenshot-1.png",
        "screenshot-2.jpg",
    ]
    assert z.read("issue.md").decode() == "# Spinner stays\n\nIt spins.\n"
    web = [json.loads(line)["msg"] for line in z.read("client.log").decode().splitlines()]
    api = [json.loads(line)["msg"] for line in z.read("server.log").decode().splitlines()]
    assert "kept click" in web and "private search" not in web
    assert "GET /api/x 200" in api and all(m != "kept click" for m in api)
    assert z.read("screenshot-1.png") == _PNG


# --- the issue tracker (S4) -----------------------------------------------------------------


class FakeGitea:
    """Gitea 1.27 as measured (plans/e16 S0): a private repository, one file per asset upload, and
    a way to make one upload fail."""

    def __init__(self):
        self.calls: list[tuple[str, str]] = []
        self.private = True
        self.has_issues = True
        self.token_ok = True
        self.fail_asset: str | None = None
        self.fail_patch = False
        self.redirect = False  # an http:// address behind a proxy that sends it to https
        self.garbage_asset = False  # a 2xx that is not the API's answer (a proxy's page)
        self.patched_body: str | None = None
        self.markdown_payloads: list[dict] = []  # what the preview asked the tracker to render

    def handler(self, request: httpx.Request) -> httpx.Response:
        path = request.url.path.removeprefix("/api/v1")
        self.calls.append((request.method, path))
        if request.headers.get("authorization") != "token t0ken" or not self.token_ok:
            return httpx.Response(401, json={"message": "invalid username, password or token"})
        if self.redirect:
            return httpx.Response(301, headers={"location": "https://git.example/"}, text="moved")
        if request.method == "GET" and path == "/repos/o/r":
            return httpx.Response(
                200,
                json={
                    "full_name": "o/r",
                    "private": self.private,
                    "has_issues": self.has_issues,
                    "html_url": "https://git.example/o/r",
                },
            )
        if request.method == "POST" and path == "/repos/o/r/issues":
            return httpx.Response(
                201, json={"number": 7, "html_url": "https://git.example/o/r/issues/7"}
            )
        if request.method == "POST" and path == "/repos/o/r/issues/7/assets":
            name = request.url.params["name"]
            if name == self.fail_asset:
                self.fail_asset = None
                return httpx.Response(500, json={"message": "boom"})
            if self.garbage_asset:
                self.garbage_asset = False
                return httpx.Response(200, text="<html>sign in</html>")
            return httpx.Response(
                201, json={"name": name, "browser_download_url": f"https://git.example/a/{name}"}
            )
        if request.method == "PATCH" and path == "/repos/o/r/issues/7":
            if self.fail_patch:
                self.fail_patch = False
                return httpx.Response(502, json={"message": "bad gateway"})
            self.patched_body = json.loads(request.content)["body"]
            return httpx.Response(201, json={"number": 7})
        if request.method == "POST" and path == "/markdown":
            self.markdown_payloads.append(json.loads(request.content))
            return httpx.Response(200, text="<h3>Steps</h3>")
        return httpx.Response(404, json={"message": "not found"})

    def uploads(self) -> list[str]:
        return [p for m, p in self.calls if p.endswith("/assets")]


@pytest.fixture
def gitea_fake(monkeypatch):
    from app.config import settings
    from app.issues import gitea
    from app.routes import debug_sessions as routes

    fake = FakeGitea()
    real = httpx.Client
    monkeypatch.setattr(
        gitea.httpx, "Client", lambda **kw: real(transport=httpx.MockTransport(fake.handler), **kw)
    )
    monkeypatch.setattr(settings, "issues_gitea_url", "https://git.example/")
    monkeypatch.setattr(settings, "issues_gitea_repo", "o/r")
    monkeypatch.setattr(settings, "issues_gitea_token", "t0ken")
    for limiter in (routes._new_issues, routes._sends, routes._previews):
        limiter.reset()
    yield fake
    for limiter in (routes._new_issues, routes._sends, routes._previews):
        limiter.reset()


def _ready(client, sid, title="Spinner stays"):
    r = client.put(f"/api/debug-sessions/{sid}/draft", json={"title": title, "body": "It spins."})
    assert r.status_code == 200


def test_without_a_tracker_the_report_can_only_be_downloaded(db, client, make_user):
    _signed_in(client, make_user)
    assert client.get("/api/me").json()["issues"] is None
    sid = _start(client)["id"]
    _ready(client, sid)
    r = client.post(f"/api/debug-sessions/{sid}/issue")
    assert r.status_code == 400 and r.json()["detail"]["reason"] == "issues_not_configured"


def test_an_issue_is_created_with_every_file_and_the_screenshots_shown(
    gitea_fake, db, client, make_user
):
    _signed_in(client, make_user)
    assert client.get("/api/me").json()["issues"] == {"repo": "o/r"}
    sid = _start(client)["id"]
    _upload(client, sid, _PNG)
    _ready(client, sid)
    r = client.post(f"/api/debug-sessions/{sid}/issue")
    assert r.status_code == 200, r.text
    session = r.json()["session"]
    assert session["issue_url"] == "https://git.example/o/r/issues/7"
    assert session["pending_files"] == []
    # One file per request (the measured API), each by its name.
    assert gitea_fake.uploads() == ["/repos/o/r/issues/7/assets"] * 3
    assert "![screenshot-1.png](https://git.example/a/screenshot-1.png)" in gitea_fake.patched_body
    assert gitea_fake.patched_body.startswith("It spins.")
    row = db.scalar(select(AuditLog).where(AuditLog.action == "issue_create"))
    assert (row.target_id, row.summary) == (sid, "https://git.example/o/r/issues/7")
    # Sent: the draft is final, and sending again sends nothing.
    assert (
        client.put(f"/api/debug-sessions/{sid}/draft", json={"title": "x", "body": "y"}).status_code
        == 409
    )
    calls = len(gitea_fake.calls)
    assert client.post(f"/api/debug-sessions/{sid}/issue").status_code == 200
    assert len(gitea_fake.calls) == calls


def test_the_claim_is_held_through_every_step_of_a_send(gitea_fake, db, make_user, as_user):
    """The claim is what keeps a draft, a screenshot or a second send out of the way — so it has to
    outlast the issue's creation, not just reach it: most of a slow send is the uploads after."""
    from app.issues import gitea

    user = make_user()
    client = as_user(user)
    sid = _start(client)["id"]
    _upload(client, sid, _PNG)
    _ready(client, sid)
    session = db.get(DebugSession, sid)
    token = debugsession.claim_send(session)
    db.commit()
    held: list[tuple[bool, int]] = []
    gitea.deliver(
        db,
        session,
        token=token,
        checkpoint=lambda: (
            db.commit(),
            held.append((debugsession.sending(session), len(gitea_fake.calls))),
        )[0],
    )
    # The first checkpoint comes before the tracker is touched at all — that is what keeps the
    # bundle's read snapshot, and its connection, out of the calls that follow.
    assert held[0] == (True, 0)
    # ...and the claim outlasts the issue's creation and every upload after it.
    assert len(held) > 1 and all(alive for alive, _ in held)


def test_the_issues_trail_is_written_by_the_step_that_writes_the_link(
    gitea_fake, db, make_user, as_user
):
    """An issue whose link is committed without its trail can never get one: the next send reads
    the link, knows the issue is not new, and records nothing. So the two go in one step."""
    from app.issues import gitea
    from app.routes import debug_sessions as routes

    user = make_user()
    client = as_user(user)
    sid = _start(client)["id"]
    _upload(client, sid, _PNG)
    _ready(client, sid)
    session = db.get(DebugSession, sid)
    token = debugsession.claim_send(session)
    db.commit()
    seen: list[tuple[bool, int]] = []

    def checkpoint() -> None:
        db.commit()
        rows = db.scalar(
            select(func.count()).select_from(AuditLog).where(AuditLog.action == "issue_create")
        )
        seen.append((session.issue_url is not None, rows or 0))

    gitea.deliver(
        db,
        session,
        token=token,
        checkpoint=checkpoint,
        on_created=lambda: routes._record_created(db, user, session),
    )
    # Every commit that can be seen with the link can be seen with the trail beside it — and there
    # has to BE one, or the filter selects nothing and `all()` is vacuously true. This assertion
    # used to go green over a `deliver` that never wrote the link at all.
    linked_rows = [rows for linked, rows in seen if linked]
    assert linked_rows and all(rows == 1 for rows in linked_rows)


def test_a_send_that_cannot_give_its_claim_back_still_reports_its_own_failure(
    gitea_fake, db, make_user, as_user, monkeypatch, caplog
):
    """The rescue in `_end_send` is there so a dead transaction cannot replace the news: what the
    user sees must be why the SEND failed, not why the tidying up did."""
    user = make_user()
    client = as_user(user)
    sid = _start(client)["id"]
    _upload(client, sid, _PNG)
    _ready(client, sid)
    gitea_fake.fail_asset = "client.log"
    locks = {"n": 0}
    real_lock = debugsession.lock

    def flaky(db_session, session):
        locks["n"] += 1
        if locks["n"] > 1:  # the route's own lock works; the one in `_end_send` does not
            raise RuntimeError("the database went away")
        return real_lock(db_session, session)

    monkeypatch.setattr(debugsession, "lock", flaky)
    with caplog.at_level(logging.WARNING):
        r = client.post(f"/api/debug-sessions/{sid}/issue")
    assert r.status_code == 422 and r.json()["detail"]["reason"] == "issues_unreachable"
    events = [getattr(x, "event", "") for x in caplog.records]
    assert "issues.claim.stuck" in events


def test_a_public_repository_gets_nothing(gitea_fake, db, client, make_user):
    gitea_fake.private = False
    _signed_in(client, make_user)
    sid = _start(client)["id"]
    _ready(client, sid)
    r = client.post(f"/api/debug-sessions/{sid}/issue")
    assert r.status_code == 422 and r.json()["detail"]["reason"] == "issues_public_repo"
    assert [m for m, _ in gitea_fake.calls] == ["GET"]


def test_a_send_that_stops_half_way_keeps_the_issue_and_the_retry_finishes_it(
    gitea_fake, db, client, make_user
):
    gitea_fake.fail_asset = "server.log"
    _signed_in(client, make_user)
    sid = _start(client)["id"]
    _ready(client, sid)
    r = client.post(f"/api/debug-sessions/{sid}/issue")
    assert r.status_code == 422 and r.json()["detail"]["reason"] == "issues_unreachable"
    session = client.get(f"/api/debug-sessions/{sid}").json()
    assert session["issue_url"] == "https://git.example/o/r/issues/7"
    assert session["pending_files"] == ["server.log"]
    r = client.post(f"/api/debug-sessions/{sid}/issue")
    assert r.status_code == 200 and r.json()["session"]["pending_files"] == []
    # One issue, and the retry sent only what was missing.
    assert [p for m, p in gitea_fake.calls if (m, p) == ("POST", "/repos/o/r/issues")] == [
        "/repos/o/r/issues"
    ]
    assert len(gitea_fake.uploads()) == 3  # client.log, the failed server.log, then server.log


def test_a_failed_edit_that_shows_the_screenshots_is_finished_by_the_retry(
    gitea_fake, db, client, make_user
):
    gitea_fake.fail_patch = True
    _signed_in(client, make_user)
    sid = _start(client)["id"]
    _upload(client, sid, _PNG)
    _ready(client, sid)
    r = client.post(f"/api/debug-sessions/{sid}/issue")
    assert r.status_code == 422 and r.json()["detail"]["reason"] == "issues_unreachable"
    # Every file is up, but the body does not show the screenshot yet: the send is not finished —
    # and nothing is named as an attachment that did not go, because they all did.
    session = client.get(f"/api/debug-sessions/{sid}").json()
    assert (session["pending_files"], session["images_pending"]) == ([], True)
    r = client.post(f"/api/debug-sessions/{sid}/issue")
    assert r.status_code == 200 and r.json()["session"]["images_pending"] is False
    assert "![screenshot-1.png](https://git.example/a/screenshot-1.png)" in gitea_fake.patched_body
    assert len(gitea_fake.uploads()) == 3  # nothing uploaded twice


def test_an_answer_that_is_not_the_api_is_refused_and_the_issue_stays_linked(
    gitea_fake, db, client, make_user
):
    gitea_fake.garbage_asset = True
    _signed_in(client, make_user)
    sid = _start(client)["id"]
    _ready(client, sid)
    r = client.post(f"/api/debug-sessions/{sid}/issue")
    assert r.status_code == 422 and r.json()["detail"]["reason"] == "issues_rejected"
    assert client.get(f"/api/debug-sessions/{sid}").json()["issue_url"] is not None
    assert client.post(f"/api/debug-sessions/{sid}/issue").status_code == 200
    # Still the one issue.
    assert [m for m, p in gitea_fake.calls if p == "/repos/o/r/issues"] == ["POST"]


def test_the_preview_has_a_pace_of_its_own(gitea_fake, db, make_user, as_user):
    """It is one authenticated call to the operator's tracker each, with the operator's token."""
    from app.routes import debug_sessions as routes

    client = as_user(make_user())
    sid = _start(client)["id"]
    monkeypatch_max = routes._previews.max_events
    routes._previews.max_events = 2
    try:
        for _ in range(2):
            assert (
                client.post(f"/api/debug-sessions/{sid}/preview", json={"text": "x"}).status_code
                == 200
            )
        r = client.post(f"/api/debug-sessions/{sid}/preview", json={"text": "x"})
        assert r.status_code == 429 and r.json()["detail"]["reason"] == "preview_rate_limited"
    finally:
        routes._previews.max_events = monkeypatch_max


def test_a_retry_that_never_finishes_is_still_paced(gitea_fake, db, make_user, as_user):
    """The budget that guards OPENING an issue must never stand between a user and finishing their
    own report — so the retry has a separate, far looser one, that only a loop reaches."""
    from app.routes import debug_sessions as routes

    client = as_user(make_user())
    sid = _start(client)["id"]
    _upload(client, sid, _PNG)
    _ready(client, sid)
    was = routes._sends.max_events
    routes._sends.max_events = 2
    try:
        gitea_fake.fail_asset = "client.log"
        assert client.post(f"/api/debug-sessions/{sid}/issue").status_code == 422
        gitea_fake.fail_asset = "client.log"
        assert client.post(f"/api/debug-sessions/{sid}/issue").status_code == 422
        r = client.post(f"/api/debug-sessions/{sid}/issue")
        assert r.status_code == 429 and r.json()["detail"]["reason"] == "issue_rate_limited"
    finally:
        routes._sends.max_events = was


def test_a_redirecting_tracker_is_refused_not_a_crash(gitea_fake, db, make_user, as_user, caplog):
    gitea_fake.redirect = True
    admin = as_user(make_user(role="admin"))
    with caplog.at_level(logging.WARNING):
        r = admin.post("/api/admin/config/test-issues")
    assert r.status_code == 422 and r.json()["detail"]["reason"] == "issues_rejected"
    # Refused as an ANSWER — only the API's own 2xx is one. A 3xx read as success would be handed to
    # the parser instead, and the tracker's address would look merely unparseable.
    events = [e for x in caplog.records if (e := getattr(x, "event", "")).startswith("issues.")]
    assert events == ["issues.gitea.refused"]


def test_the_delivery_survives_a_save_of_the_draft():
    session = DebugSession(draft={"title": "t", "delivery": {"number": 7, "pending": ["a.log"]}})
    debugsession.save_draft(session, {"title": "t2", "body": "b", "excluded": [], "form": {}})
    assert session.draft["title"] == "t2"
    assert debugsession.not_uploaded(session) == ["a.log"]


def test_a_report_needs_a_title_and_a_wrong_token_says_so(gitea_fake, db, client, make_user):
    _signed_in(client, make_user)
    sid = _start(client)["id"]
    r = client.post(f"/api/debug-sessions/{sid}/issue")
    assert r.status_code == 422 and r.json()["detail"]["reason"] == "issue_title_required"
    _ready(client, sid)
    gitea_fake.token_ok = False
    r = client.post(f"/api/debug-sessions/{sid}/issue")
    assert r.status_code == 422 and r.json()["detail"]["reason"] == "issues_bad_token"
    assert client.get(f"/api/debug-sessions/{sid}").json()["issue_url"] is None


def test_new_issues_are_paced_per_user(gitea_fake, db, make_user, as_user, monkeypatch):
    from app.ratelimit import RateLimiter
    from app.routes import debug_sessions as routes

    monkeypatch.setattr(routes, "_new_issues", RateLimiter(max_events=1, window_seconds=3600))
    client = as_user(make_user())
    first = _start(client)["id"]
    _ready(client, first)
    assert client.post(f"/api/debug-sessions/{first}/issue").status_code == 200
    assert client.post(f"/api/debug-sessions/{first}/stop").status_code == 200
    second = _start(client)["id"]
    _ready(client, second)
    r = client.post(f"/api/debug-sessions/{second}/issue")
    assert r.status_code == 429 and r.json()["detail"] == {
        "code": "refused",
        "reason": "issues_rate_limited",
        "limit": 1,
    }


def test_an_attempt_refused_before_the_issue_opened_is_not_counted(
    gitea_fake, db, make_user, as_user, monkeypatch
):
    from app.ratelimit import RateLimiter
    from app.routes import debug_sessions as routes

    monkeypatch.setattr(routes, "_new_issues", RateLimiter(max_events=1, window_seconds=3600))
    client = as_user(make_user())
    sid = _start(client)["id"]
    _ready(client, sid)
    gitea_fake.token_ok = False
    assert client.post(f"/api/debug-sessions/{sid}/issue").status_code == 422
    gitea_fake.token_ok = True
    assert client.post(f"/api/debug-sessions/{sid}/issue").status_code == 200


def test_the_preview_is_the_trackers_own_rendering(gitea_fake, db, make_user, as_user):
    owner, stranger = as_user(make_user()), as_user(make_user())
    sid = _start(owner)["id"]
    r = owner.post(f"/api/debug-sessions/{sid}/preview", json={"text": "### Steps"})
    assert r.json() == {"ok": True, "html": "<h3>Steps</h3>"}
    assert (
        stranger.post(f"/api/debug-sessions/{sid}/preview", json={"text": "x"}).status_code == 404
    )
    assert stranger.post(f"/api/debug-sessions/{sid}/issue").status_code == 404


def test_the_preview_never_asks_the_tracker_for_a_repository_context(
    gitea_fake, db, make_user, as_user
):
    """A `Context` makes Gitea hunt for commit links, and on the API's render path that hunt
    dereferences a nil checker: the tracker's whole process segfaults. Every report carries a
    hex-shaped id, so the payload — not the text — is what has to stay safe (measured against
    1.26.4 / 1.27.1, fixed upstream in 1.27.3; a self-hoster's version is not ours to pick)."""
    owner = as_user(make_user())
    sid = _start(owner)["id"]
    text = "| tab | 5645b1c0ae01 |"
    assert owner.post(f"/api/debug-sessions/{sid}/preview", json={"text": text}).status_code == 200
    assert gitea_fake.markdown_payloads == [{"Text": text, "Mode": "gfm"}]


def test_the_admin_connection_test_names_what_is_wrong(gitea_fake, db, make_user, as_user):
    admin = as_user(make_user(role="admin"))
    r = admin.post("/api/admin/config/test-issues")
    assert r.json() == {"ok": True, "repo": "o/r", "url": "https://git.example/o/r"}
    gitea_fake.has_issues = False
    r = admin.post("/api/admin/config/test-issues")
    assert r.status_code == 422 and r.json()["detail"]["reason"] == "issues_disabled"
    assert as_user(make_user()).post("/api/admin/config/test-issues").status_code == 403