peter
9cf177b198
- admin config group "Bug reports": Gitea address, repository, token (secret),
and "Check the connection" — the repository must answer, take issues and
be PRIVATE (a report carries its reporter's log); checked on every send too
- POST /api/debug-sessions/{id}/issue sends the SAVED draft: create the issue,
upload each file on its own request (the measured API), then show the
screenshots inline; progress is kept in the draft, so a send that stops
half-way keeps its issue and the retry sends only what is missing; 5 new
issues an hour per user; audited as issue_create
- the composer: Create issue (only where a tracker is set up, naming the
repository), a partial-send banner with retry, and a Preview — the tracker's
own rendering in a sandboxed frame, or the zip's markdown without one
- the recorder's refusals answer in the localized family (debug_session_*,
screenshot_*, issue(s)_*), not bare codes the UI could only call generic
- a tracker refusal logs the status only: its message can quote the title back
748 lines
29 KiB
Python
748 lines
29 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 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_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)>", "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_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.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 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"})
|
|
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":
|
|
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_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_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"]["reason"] == "issues_rate_limited"
|
|
|
|
|
|
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
|