peter
8d0083d4e4
After Stop (or at the cap, or from Settings → Diagnosztika's past sessions) one dialog drafts the issue: title, description, the steps drafted from the session's markers and page changes (a marker without a label is named by its time), the environment (version, browser, system, window, language). Beside it, what goes with it: the browser and server logs — each reviewable line by line, an unticked line is not sent — and the screenshots, removable, with paste / drop / pick to add more. - it saves as it goes (debug_sessions.draft, migration 0090): the rendered markdown in the user's language, the excluded rows, and its own fields to reopen as it was; a sent session is read-only - the zip export (issue.md, client.log, server.log, screenshots) is built from the saved draft by app/issues/bundle.py — the one definition the Gitea delivery (S4) will share; .log, not .jsonl, which Gitea refuses - the log line's level tags and clock move to components/logLineParts so the review list draws a level the way both log viewers do
546 lines
20 KiB
Python
546 lines
20 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 logging
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from fastapi import BackgroundTasks
|
|
from sqlalchemy import select
|
|
|
|
from app import debugsession, logs
|
|
from app.models import 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_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))
|
|
assert debugsession.level_for(user, session) == ("TRACE", user.diagnostics_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)>", "not_an_image"), (b"", "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": "screenshot_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"]["reason"] == "too_many"
|
|
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 == 422 and r.json()["detail"]["reason"] == "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_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"]["code"] == "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
|