test: no test throws a response away; the stream session is narrowed, not baselined

The 7 baselined Optional accesses in plex/stream.py were one fact the type did not carry: the
initial spawn (replace_only=None) always supersedes, so only the fallback respawn can get None.
An assert at the call site says so; the pyright baseline drops 195 -> 188.

test_abuse_limits: the public-URL and YouTube enqueue tests stopped at the profile lookup
(400 no_profile) and passed on that unrelated refusal. They now send an inline spec and assert
the job is queued. The same shape — a request whose response is discarded — sat at 53 more
places across the suite (all returned what they should, measured with a TestClient probe);
each now asserts its status, and test_response_checked keeps it that way (seen red).
This commit is contained in:
2026-10-03 19:06:47 +02:00
parent 614d728561
commit ca9ca85b18
19 changed files with 246 additions and 129 deletions
+26 -11
View File
@@ -212,7 +212,12 @@ def test_a_timestamp_outside_the_consent_window_is_clamped(elevated):
now = datetime.now(timezone.utc)
future = (now + timedelta(days=365)).isoformat()
past = (now - timedelta(days=30)).isoformat()
client.post("/api/diagnostics/events", json={"events": [_event(ts=future), _event(ts=past)]})
assert (
client.post(
"/api/diagnostics/events", json={"events": [_event(ts=future), _event(ts=past)]}
).status_code
== 200
)
late, early = rows()
assert late.ts <= datetime.now(timezone.utc)
assert early.ts >= now - timedelta(hours=diagnostics.ELEVATION_HOURS, minutes=1)
@@ -342,15 +347,15 @@ def _app() -> TestClient:
def test_the_account_reaches_the_handlers_own_lines(gated):
_app().get("/api/thing/41")
assert _app().get("/api/thing/41").status_code == 200
line = next(r for r in gated.records if r.getMessage() == "an ordinary line")
assert getattr(line, "user_id", None) == 41
def test_one_users_elevation_logs_their_requests_in_detail_and_nobody_elses(gated):
client = _app()
client.get("/api/thing/7", params={"level": "DEBUG"})
client.get("/api/thing/8")
assert client.get("/api/thing/7", params={"level": "DEBUG"}).status_code == 200
assert client.get("/api/thing/8").status_code == 200
detail = [r for r in gated.records if r.getMessage().startswith("detail only")]
assert [r.user_id for r in detail] == [7]
assert getattr(detail[0], "elevated", False) is True
@@ -363,9 +368,9 @@ def test_one_users_elevation_logs_their_requests_in_detail_and_nobody_elses(gate
def test_the_root_level_returns_to_the_base_when_the_elevation_ends(gated):
client = _app()
client.get("/api/thing/7", params={"level": "TRACE"})
assert client.get("/api/thing/7", params={"level": "TRACE"}).status_code == 200
assert logging.getLogger().level == logs.TRACE
client.get("/api/thing/7") # the same user, back at the base
assert client.get("/api/thing/7").status_code == 200 # the same user, back at the base
assert logging.getLogger().level == logging.INFO
logs._track_elevation(9, logging.DEBUG, datetime.now(timezone.utc) - timedelta(seconds=1))
assert logging.getLogger().level == logging.INFO # an already-expired one is pruned at once
@@ -373,15 +378,25 @@ def test_the_root_level_returns_to_the_base_when_the_elevation_ends(gated):
def test_the_query_string_is_logged_only_for_a_traced_request_and_redacted(gated):
client = _app()
client.get("/api/thing/7", params={"level": "DEBUG", "q": "lofi"})
client.get("/api/thing/7", params={"level": "TRACE", "q": "lofi", "token": "SECRET"})
assert client.get("/api/thing/7", params={"level": "DEBUG", "q": "lofi"}).status_code == 200
assert (
client.get(
"/api/thing/7", params={"level": "TRACE", "q": "lofi", "token": "SECRET"}
).status_code
== 200
)
lines = [r for r in gated.records if getattr(r, "event", None) == "http.request"]
assert "query" not in lines[0].attrs
assert "q=lofi" in lines[1].attrs["query"] and "SECRET" not in lines[1].attrs["query"]
def test_an_address_typed_into_a_search_is_redacted_though_the_browser_encodes_it(gated):
_app().get("/api/thing/7?level=TRACE&q=someone%40example.com&new_password=hunter2")
assert (
_app()
.get("/api/thing/7?level=TRACE&q=someone%40example.com&new_password=hunter2")
.status_code
== 200
)
(line,) = [r for r in gated.records if getattr(r, "event", None) == "http.request"]
assert line.attrs["query"] == "level=TRACE&q=[email]&new_password=[redacted]"
@@ -400,7 +415,7 @@ def test_a_credential_param_goes_whole_however_its_value_decodes(raw):
def test_an_admin_level_of_trace_does_not_log_other_users_query_strings(gated):
logs.set_level("TRACE")
_app().get("/api/thing/7", params={"q": "something private"})
assert _app().get("/api/thing/7", params={"q": "something private"}).status_code == 200
(line,) = [r for r in gated.records if getattr(r, "event", None) == "http.request"]
assert "query" not in line.attrs
@@ -488,7 +503,7 @@ def test_a_real_sign_in_carries_the_consented_level_into_the_process(gated, clie
user = _signed_in(client, make_user)
assert client.put("/api/me/diagnostics", json={"level": "DEBUG"}).status_code == 200
logs._elevations.clear() # only a request resolving the account through auth may put it back
client.get("/api/me")
assert client.get("/api/me").status_code == 200
assert logs._elevations[user.id][0] == logging.DEBUG
assert client.put("/api/me/diagnostics", json={"level": "INFO"}).status_code == 200
assert user.id not in logs._elevations