peter
b2eaa9e00b
Round 2 of the E22 review, on what round 1 changed. `redact_query` decodes a parameter before the text rules see it, which is what lets `%40` be recognised as an address — but `_SECRET_PARAM_RE` was written for the raw, encoded form, where a value holds no space, `&` or quote. On decoded text its value class stops at the first one, so `?state=a%20b` kept its tail and `?code=%20abc` was not matched at all. The parameter-NAME rule now decides, on the key, before any text rule runs. `forget_user` dropped the account's level elevation without recomputing the root level, so a process whose only elevated account had just deleted itself kept building TRACE records to throw away, until some later request happened to recompute it. Every other path that drops an elevation recomputes. The sink's writer took the erasure lock and THEN waited for a pooled connection, while every request path already holds a connection when it reaches that lock — so under pool pressure the writer held a process-global lock against the very requests that could hand a connection back. Same order on both sides now; the lock still spans the check, the insert and the commit. The storage-budget notice said the rest of the hour was not kept, and then smaller batches kept landing underneath it: the window slides and a refused batch is not charged to it. Says what it does. Tests: the route's half of the erasure rule had no test — only the sink's did — and a source-level contract test now covers the third insert nobody has written yet. `logs._erased` is process-global with an hour's memory while the DB fixture hands serial ids back out from 1, so it is swept before each test like the rate limiters are.
493 lines
20 KiB
Python
493 lines
20 KiB
Python
"""E22 S5 — a user's consented diagnostics level, and the browser log it lets in.
|
|
|
|
Three parts:
|
|
* the level itself: the server computes its expiry, a lapsed one reads as the base, the demo
|
|
account cannot raise it (DB lane, through the real routes);
|
|
* `POST /api/diagnostics/events`: refused at the base level, and nothing in a batch is trusted —
|
|
the account comes from the session, everything is redacted again, events under the consented
|
|
level are dropped, sizes and pace are capped (DB lane);
|
|
* the server side of the elevation: ONE user's requests are logged in detail while everyone
|
|
else's stay at the base level, and the account reaches a handler's own lines at all (pure, an
|
|
in-process app).
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import threading
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from fastapi import Depends, FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
from app import diagnostics, logs, logsink, sysconfig
|
|
from app.config import settings
|
|
from app.models import LogEvent, User
|
|
from app.routes import diagnostics as diagnostics_routes
|
|
|
|
_PW = "correct horse battery"
|
|
|
|
|
|
# --- the level ------------------------------------------------------------------------------
|
|
|
|
|
|
def test_a_raised_level_counts_until_it_lapses_and_not_a_moment_longer():
|
|
now = datetime.now(timezone.utc)
|
|
assert diagnostics.effective_level(User()) == "INFO"
|
|
raised = User(diagnostics_level="TRACE", diagnostics_until=now + timedelta(minutes=1))
|
|
assert diagnostics.effective_level(raised, now) == "TRACE"
|
|
assert diagnostics.effective_level(raised, now + timedelta(minutes=2)) == "INFO"
|
|
# A level without an expiry is not a level that never ends — it is no level at all.
|
|
assert diagnostics.effective_level(User(diagnostics_level="DEBUG")) == "INFO"
|
|
|
|
|
|
def _signed_in(client, make_user, **kw):
|
|
"""A real sign-in: `/api/me` resolves through `optional_current_user`, which the `as_user`
|
|
override does not cover (see test_me_email_flag)."""
|
|
user = make_user(email="diag@example.com", password=_PW, **kw)
|
|
r = client.post("/auth/password-login", json={"email": "diag@example.com", "password": _PW})
|
|
assert r.status_code == 200, r.text
|
|
return user
|
|
|
|
|
|
def test_raising_the_level_sets_the_servers_expiry_whatever_the_client_sends(db, client, make_user):
|
|
user = _signed_in(client, make_user)
|
|
r = client.put("/api/me/diagnostics", json={"level": "DEBUG", "until": "2099-01-01T00:00:00Z"})
|
|
assert r.status_code == 200, r.text
|
|
db.refresh(user)
|
|
expected = datetime.now(timezone.utc) + timedelta(hours=diagnostics.ELEVATION_HOURS)
|
|
assert abs((user.diagnostics_until - expected).total_seconds()) < 60
|
|
assert r.json()["diagnostics"]["level"] == "DEBUG"
|
|
assert client.get("/api/me").json()["diagnostics"]["level"] == "DEBUG"
|
|
|
|
assert client.put("/api/me/diagnostics", json={"level": "INFO"}).status_code == 200
|
|
db.refresh(user)
|
|
assert (user.diagnostics_level, user.diagnostics_until) == (None, None)
|
|
assert client.get("/api/me").json()["diagnostics"] == {
|
|
"level": "INFO",
|
|
"until": None,
|
|
"retention_days": 7,
|
|
}
|
|
|
|
|
|
def test_a_lapsed_level_reads_as_the_base_in_me(db, client, make_user):
|
|
_signed_in(
|
|
client,
|
|
make_user,
|
|
diagnostics_level="TRACE",
|
|
diagnostics_until=datetime.now(timezone.utc) - timedelta(minutes=1),
|
|
)
|
|
assert client.get("/api/me").json()["diagnostics"] == {
|
|
"level": "INFO",
|
|
"until": None,
|
|
"retention_days": 7,
|
|
}
|
|
|
|
|
|
def test_only_the_three_levels_can_be_chosen(db, make_user, as_user):
|
|
client = as_user(make_user())
|
|
for bogus in ("WARNING", "debug", "EVERYTHING"):
|
|
assert client.put("/api/me/diagnostics", json={"level": bogus}).status_code == 422
|
|
|
|
|
|
def test_the_demo_account_can_return_to_the_base_but_never_raise_it(db, make_user, as_user):
|
|
client = as_user(make_user(is_demo=True))
|
|
r = client.put("/api/me/diagnostics", json={"level": "TRACE"})
|
|
assert r.status_code == 403 and r.json()["detail"]["code"] == "demo"
|
|
assert client.put("/api/me/diagnostics", json={"level": "INFO"}).status_code == 200
|
|
|
|
|
|
def test_preferences_must_be_a_json_object(db, make_user, as_user):
|
|
client = as_user(make_user())
|
|
assert client.put("/api/me/preferences", json=["not", "an", "object"]).status_code == 422
|
|
r = client.put("/api/me/preferences", json={"theme": "dark"})
|
|
assert r.status_code == 200 and r.json()["preferences"]["theme"] == "dark"
|
|
|
|
|
|
# --- the browser's batches ------------------------------------------------------------------
|
|
|
|
|
|
def _event(**over) -> dict:
|
|
return {
|
|
"ts": datetime.now(timezone.utc).isoformat(),
|
|
"level": "INFO",
|
|
"event": "nav.view",
|
|
"msg": "view",
|
|
"tab_id": "a1b2c3",
|
|
"version": "0.88.0",
|
|
**over,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def elevated(db, make_user, as_user):
|
|
"""A signed-in client whose user consented to `level` a minute ago, and a reader for its rows."""
|
|
diagnostics_routes._batches.reset()
|
|
|
|
def _make(level: str = "DEBUG"):
|
|
user = make_user(
|
|
diagnostics_level=level,
|
|
diagnostics_until=datetime.now(timezone.utc) + timedelta(hours=23),
|
|
)
|
|
|
|
def rows() -> list[LogEvent]:
|
|
db.expire_all()
|
|
return list(db.execute(select(LogEvent).order_by(LogEvent.id)).scalars())
|
|
|
|
return user, as_user(user), rows
|
|
|
|
yield _make
|
|
diagnostics_routes._batches.reset()
|
|
|
|
|
|
def test_nothing_is_accepted_at_the_base_level(db, make_user, as_user):
|
|
client = as_user(make_user())
|
|
r = client.post("/api/diagnostics/events", json={"events": [_event()]})
|
|
assert r.status_code == 409 and r.json()["detail"]["code"] == "diagnostics_off"
|
|
assert not db.execute(select(LogEvent)).first()
|
|
|
|
|
|
def test_an_anonymous_caller_is_refused(client):
|
|
assert client.post("/api/diagnostics/events", json={"events": [_event()]}).status_code == 401
|
|
|
|
|
|
def test_a_batch_is_stored_as_the_sessions_web_events_and_redacted_again(elevated):
|
|
user, client, rows = elevated("DEBUG")
|
|
r = client.post(
|
|
"/api/diagnostics/events",
|
|
json={
|
|
"events": [
|
|
_event(
|
|
level="WARNING",
|
|
event="ui.notify",
|
|
msg="sent to someone@example.com",
|
|
attrs={"password": "hunter2", "url": "/?token=abc123", "user_id": 999},
|
|
request_id="0123456789abcdef",
|
|
),
|
|
_event(level="DEBUG", event="ui.click", attrs={"control": "feed.refresh"}),
|
|
],
|
|
"user_id": 999, # ignored: the account is the session's
|
|
},
|
|
)
|
|
assert r.json() == {"ok": True, "stored": 2}
|
|
warning, click = rows()
|
|
assert (warning.source, warning.logger, warning.user_id) == ("web", "web", user.id)
|
|
assert warning.msg == "sent to [email]"
|
|
assert warning.attrs["password"] == logs.REDACTED
|
|
assert "abc123" not in json.dumps(warning.attrs)
|
|
assert warning.attrs["tab_id"] == "a1b2c3"
|
|
assert warning.request_id == "0123456789abcdef"
|
|
assert (warning.sev, click.level, click.version) == (13, "DEBUG", "0.88.0")
|
|
|
|
|
|
def test_an_erased_accounts_batch_is_stored_without_the_account(elevated):
|
|
"""The sink's writer is not the only path into `log_events` — this route inserts too, and an
|
|
account erased while its browser was mid-flush must not get its rows back through it. Faked at
|
|
the `forget_user` seam the purge itself calls: once the User row is actually gone the session
|
|
no longer resolves, so the reachable window is exactly a request already past `current_user`.
|
|
(`logs._erased` is cleared before every test by conftest's `_reset_log_erasures`.)"""
|
|
user, client, rows = elevated("DEBUG")
|
|
logs.forget_user(user.id)
|
|
r = client.post("/api/diagnostics/events", json={"events": [_event(level="WARNING")]})
|
|
assert r.json() == {"ok": True, "stored": 1}
|
|
(stored,) = rows()
|
|
assert (stored.source, stored.user_id) == ("web", None)
|
|
|
|
|
|
def test_an_event_under_the_consented_level_is_dropped(elevated):
|
|
_user, client, rows = elevated("DEBUG")
|
|
r = client.post(
|
|
"/api/diagnostics/events",
|
|
json={"events": [_event(level="TRACE", msg="typed: lofi"), _event(level="DEBUG")]},
|
|
)
|
|
assert r.json()["stored"] == 1
|
|
assert [row.level for row in rows()] == ["DEBUG"]
|
|
|
|
|
|
def test_a_timestamp_outside_the_consent_window_is_clamped(elevated):
|
|
_user, client, rows = elevated("DEBUG")
|
|
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)]})
|
|
late, early = rows()
|
|
assert late.ts <= datetime.now(timezone.utc)
|
|
assert early.ts >= now - timedelta(hours=diagnostics.ELEVATION_HOURS, minutes=1)
|
|
|
|
|
|
def test_sizes_are_capped(elevated):
|
|
_user, client, rows = elevated("DEBUG")
|
|
too_many = [_event() for _ in range(diagnostics_routes.EVENTS_MAX + 1)]
|
|
assert client.post("/api/diagnostics/events", json={"events": too_many}).status_code == 422
|
|
|
|
huge = "x" * (diagnostics_routes.BODY_MAX_BYTES + 1)
|
|
r = client.post(
|
|
"/api/diagnostics/events",
|
|
content=json.dumps({"events": [_event(attrs={"blob": huge})]}),
|
|
headers={"content-type": "application/json"},
|
|
)
|
|
assert r.status_code == 413
|
|
|
|
big_attrs = {"blob": "y" * (diagnostics_routes.ATTRS_MAX_BYTES + 10)}
|
|
assert (
|
|
client.post("/api/diagnostics/events", json={"events": [_event(attrs=big_attrs)]}).json()[
|
|
"stored"
|
|
]
|
|
== 1
|
|
)
|
|
(row,) = rows()
|
|
assert "blob" not in row.attrs and row.attrs["attrs_dropped_bytes"] > 0
|
|
|
|
bad_code = client.post("/api/diagnostics/events", json={"events": [_event(event="Nav View")]})
|
|
assert bad_code.status_code == 422
|
|
|
|
|
|
def test_the_pace_is_capped_per_user(elevated):
|
|
_user, client, _rows = elevated("DEBUG")
|
|
codes = [
|
|
client.post("/api/diagnostics/events", json={"events": []}).status_code for _ in range(31)
|
|
]
|
|
assert codes[:30] == [200] * 30 and codes[30] == 429
|
|
|
|
|
|
def test_what_an_account_stores_per_hour_is_capped_and_said_once(elevated, monkeypatch):
|
|
user, client, rows = elevated("DEBUG")
|
|
monkeypatch.setattr(
|
|
diagnostics_routes, "_stored_bytes", diagnostics_routes.RateLimiter(3_000, 3600)
|
|
)
|
|
batch = {"events": [_event(msg="m" * 1_000)]}
|
|
stored = [client.post("/api/diagnostics/events", json=batch).json()["stored"] for _ in range(4)]
|
|
assert stored == [1, 1, 0, 0]
|
|
events = [row.event for row in rows()]
|
|
# Two batches, then ONE notice for the hour, not one per refused batch.
|
|
assert events == ["nav.view", "nav.view", "diagnostics.budget_spent"]
|
|
assert rows()[-1].user_id == user.id
|
|
|
|
|
|
def test_only_a_json_body_is_read(elevated):
|
|
_user, client, rows = elevated("DEBUG")
|
|
# What a same-site `<form enctype="text/plain">` can send without a preflight.
|
|
r = client.post(
|
|
"/api/diagnostics/events",
|
|
content=json.dumps({"events": [_event()]}),
|
|
headers={"content-type": "text/plain"},
|
|
)
|
|
assert r.status_code == 415 and r.json()["detail"]["code"] == "json_required"
|
|
assert rows() == []
|
|
ok = client.post(
|
|
"/api/diagnostics/events",
|
|
content=json.dumps({"events": [_event()]}),
|
|
headers={"content-type": "application/json; charset=utf-8"},
|
|
)
|
|
assert ok.json()["stored"] == 1
|
|
|
|
|
|
# --- the server side of an elevation ---------------------------------------------------------
|
|
|
|
|
|
class _Capture(logging.Handler):
|
|
"""The stdout handler's filters (context + gate), capturing records."""
|
|
|
|
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():
|
|
"""A process configured at INFO with a gated handler on the ROOT (where the real one sits), and
|
|
every piece of module state the elevation touches restored afterwards."""
|
|
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 _app() -> TestClient:
|
|
"""What a real request does: the request middleware, then a SYNC auth dependency that resolves
|
|
the account (run by FastAPI in a threadpool, on a copy of the context), then a handler."""
|
|
app = FastAPI()
|
|
app.middleware("http")(logs.log_requests)
|
|
handler_log = logging.getLogger("siftlode.difftest")
|
|
|
|
def account(user_id: int, level: str = "INFO"):
|
|
until = datetime.now(timezone.utc) + timedelta(hours=1)
|
|
logs.set_user(user_id, level, until)
|
|
|
|
@app.get("/api/thing/{user_id}")
|
|
def _thing(user_id: int, _=Depends(account)):
|
|
handler_log.debug("detail only an elevated request keeps")
|
|
handler_log.info("an ordinary line")
|
|
return {}
|
|
|
|
return TestClient(app)
|
|
|
|
|
|
def test_the_account_reaches_the_handlers_own_lines(gated):
|
|
_app().get("/api/thing/41")
|
|
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")
|
|
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
|
|
# Every line of the elevated request is marked for storage — the sink keeps it at any level.
|
|
ordinary = [r for r in gated.records if r.getMessage() == "an ordinary line"]
|
|
assert [logsink.should_store(r) for r in ordinary] == [True, False]
|
|
# The process's configured level is still what it reports, however low the root went.
|
|
assert logs.current_level() == "INFO" and logging.getLogger().level == logging.DEBUG
|
|
|
|
|
|
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 logging.getLogger().level == logs.TRACE
|
|
client.get("/api/thing/7") # 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
|
|
|
|
|
|
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"})
|
|
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")
|
|
(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]"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"raw",
|
|
["state=a%20b", "code=%20SECRET", "sig=a%22b", "key=a%26b", "reset=a+b"],
|
|
ids=["space", "leading-space", "quote", "ampersand", "plus"],
|
|
)
|
|
def test_a_credential_param_goes_whole_however_its_value_decodes(raw):
|
|
"""Decoding is what lets `_EMAIL_RE` see `%40`, but it also puts spaces and separators into a
|
|
value that `_SECRET_PARAM_RE`'s value class stops at — so the name rule decides, not the text."""
|
|
name = raw.split("=", 1)[0]
|
|
assert logs.redact_query(f"{raw}&q=lofi") == f"{name}={logs.REDACTED}&q=lofi"
|
|
|
|
|
|
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"})
|
|
(line,) = [r for r in gated.records if getattr(r, "event", None) == "http.request"]
|
|
assert "query" not in line.attrs
|
|
|
|
|
|
def test_the_admin_level_applies_to_the_running_process_when_set_and_when_reset(
|
|
gated, db, make_user, as_user
|
|
):
|
|
admin = as_user(make_user(role="admin"))
|
|
r = admin.patch("/api/admin/config/log_level", json={"value": "DEBUG"})
|
|
assert r.status_code == 200, r.text
|
|
assert logs.current_level() == "DEBUG"
|
|
assert admin.delete("/api/admin/config/log_level").status_code == 200
|
|
assert logs.current_level() == settings.log_level
|
|
|
|
|
|
def test_erasing_the_only_elevated_account_puts_the_process_level_back(gated):
|
|
"""The root level is the lowest of the base and every unexpired elevation. `forget_user` drops
|
|
an elevation like any other path that drops one, so it owes the same recompute — or the process
|
|
keeps building every TRACE record for an account that no longer exists, to throw them away."""
|
|
until = datetime.now(timezone.utc) + timedelta(hours=1)
|
|
try:
|
|
logs.set_user(7, "TRACE", until)
|
|
assert logging.getLogger().level == logs.TRACE
|
|
logs.forget_user(7)
|
|
assert logging.getLogger().level == logging.INFO
|
|
finally:
|
|
# `set_user` writes the request scope, and outside a request that scope is this process's
|
|
# for good — a later test would store its INFO lines as an elevated session's.
|
|
logs.set_user(None)
|
|
|
|
|
|
def test_the_stored_admin_level_is_applied_at_boot(gated, db, monkeypatch):
|
|
import asyncio
|
|
|
|
import app.main as app_main
|
|
|
|
sysconfig.set_value(db, "log_level", "DEBUG")
|
|
db.commit()
|
|
monkeypatch.setattr(app_main.state, "is_configured", lambda db: True)
|
|
monkeypatch.setattr(app_main, "start_scheduler", lambda: None)
|
|
monkeypatch.setattr(app_main, "shutdown_scheduler", lambda: None)
|
|
monkeypatch.setattr(app_main.plex_subtitles, "shutdown", lambda: None)
|
|
|
|
async def _no_sweep():
|
|
return None
|
|
|
|
monkeypatch.setattr(app_main, "_sweep_hls_orphans", _no_sweep)
|
|
levels: list[str] = []
|
|
|
|
async def _cycle():
|
|
async with app_main.lifespan(app_main.app):
|
|
levels.append(logs.current_level())
|
|
|
|
asyncio.run(_cycle())
|
|
assert levels == ["DEBUG"]
|
|
|
|
|
|
def test_the_root_level_is_applied_in_the_order_the_elevations_changed(gated, monkeypatch):
|
|
"""Two requests change the elevations at once: A's user leaves (the root may go back to INFO),
|
|
B's user arrives (it must go to DEBUG). B's change is the later one, so DEBUG must be what stays
|
|
— even when A is paused between working out its level and applying it."""
|
|
root = logging.getLogger()
|
|
logs._track_elevation(1, logging.DEBUG, datetime.now(timezone.utc) + timedelta(hours=1))
|
|
real_set = root.setLevel
|
|
other: list[threading.Thread] = []
|
|
|
|
def paused_set(level):
|
|
if level == logging.INFO and not other:
|
|
b = threading.Thread(
|
|
target=logs._track_elevation,
|
|
args=(2, logging.DEBUG, datetime.now(timezone.utc) + timedelta(hours=1)),
|
|
)
|
|
other.append(b)
|
|
b.start()
|
|
b.join(0.3) # B runs to completion here unless A's level change holds it off
|
|
real_set(level)
|
|
|
|
monkeypatch.setattr(root, "setLevel", paused_set)
|
|
logs._track_elevation(1, None, None) # A
|
|
other[0].join(5)
|
|
assert root.level == logging.DEBUG and 2 in logs._elevations
|
|
|
|
|
|
def test_a_real_sign_in_carries_the_consented_level_into_the_process(gated, client, db, make_user):
|
|
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 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
|