peter
bc03e82687
Creating the issue rebuilt the delivery state from scratch and dropped the claim with it, so for a new issue the claim covered only the creation — a draft, a screenshot or a second send could change the report under the uploads and the body edit that follow, which is exactly what the claim is for. The state is now built on what was read, and a test walks the send's checkpoints to see the claim still held at each one. Found by mutation-checking the round-2 tests: removing the claim's release left them green, because for a new issue there was nothing left to release.
992 lines
40 KiB
Python
992 lines
40 KiB
Python
"""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 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"]
|
|
client.post(f"/api/debug-sessions/{first}/stop")
|
|
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"]
|
|
client.post(
|
|
f"/api/debug-sessions/{sid}/screenshots",
|
|
content=_PNG,
|
|
headers={"content-type": "image/png"},
|
|
)
|
|
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()
|
|
client.get("/api/debug-sessions", headers={"x-debug-session": sid})
|
|
(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"]
|
|
client.post(f"/api/debug-sessions/{own}/stop")
|
|
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()
|
|
client.get("/api/debug-sessions", headers={"x-debug-session": sid})
|
|
(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"
|
|
debugsession.set_delivery(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.
|
|
debugsession.set_delivery(session, {"number": 1, "pending": [], "links": {}})
|
|
db.commit()
|
|
assert client.delete(f"/api/debug-sessions/{sid}/screenshots/{shot}").status_code == 200
|
|
|
|
|
|
def test_a_send_under_way_holds_the_draft_and_the_screenshots_without_blocking_stop(db, recording):
|
|
client, sid = recording
|
|
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"
|
|
r = client.post(f"/api/debug-sessions/{sid}/issue")
|
|
assert r.status_code == 409 and r.json()["detail"]["reason"] == "issue_sending"
|
|
# 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.
|
|
debugsession.set_delivery(
|
|
session,
|
|
{"sending": (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()},
|
|
)
|
|
db.commit()
|
|
assert (
|
|
client.put(f"/api/debug-sessions/{sid}/draft", json={"title": "x", "body": "y"}).status_code
|
|
== 200
|
|
)
|
|
|
|
|
|
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
|
|
client.post(f"/api/debug-sessions/{sid}/stop")
|
|
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
|
|
client.post(f"/api/debug-sessions/{sid}/stop")
|
|
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)
|
|
client.put(
|
|
f"/api/debug-sessions/{sid}/draft",
|
|
json={"title": "Spinner stays", "body": "It spins.", "excluded": [dropped_web.id, 999999]},
|
|
)
|
|
|
|
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
|
|
|
|
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":
|
|
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")
|
|
routes._new_issues.reset()
|
|
yield fake
|
|
routes._new_issues.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 "" 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)
|
|
debugsession.claim_send(session)
|
|
db.commit()
|
|
held: list[bool] = []
|
|
gitea.deliver(
|
|
db,
|
|
session,
|
|
checkpoint=lambda: (db.commit(), held.append(debugsession.sending(session)))[0],
|
|
)
|
|
assert held and all(held) # after the issue was created, and after every upload
|
|
|
|
|
|
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 "" 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_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
|
|
client.post(f"/api/debug-sessions/{first}/stop")
|
|
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_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
|