alexei.dolgolyov
a5effba553
Backend
- snapshot: GET /api/v1/snapshot aggregates targets, devices, sources,
presets and system into one payload for the HA coordinator, collapsing
the prior ~2N+M request fan-out; per-section ?include= gating.
- graph: GET /api/v1/graph{,/schema,/dependents} backed by a pure,
unit-tested graph_schema engine — one authoritative connectable-field
registry so the editor no longer hard-codes topology in two places.
- devices: thread mqtt_source_id through DeviceCreate/Update/Response and
the routes for multi-broker MQTT; shared validate_mqtt_source_exists
(_mqtt_validation.py) reused by device + output-target routes; stop
update_device masking intentional 4xx as 500.
- shutdown: bound uvicorn graceful-shutdown via GRACEFUL_SHUTDOWN_TIMEOUT
(shared by __main__, android_entry, demo) so a lingering events WebSocket
can't strand LED targets or block process exit.
- access log: structured _access_log middleware attributing each request to
its authenticated token label (never the secret); uvicorn access_log off.
Frontend
- graph editor: generic schema-driven port/edge rendering, layout and
connection handling; service-worker refresh.
- device modals: MQTT broker EntitySelect for device_type=mqtt in add-device
and settings, wired into load/save/validate/dirty-check/clone.
- i18n: en/ru/zh keys.
Tests: graph routes + schema, snapshot routes, access log, mqtt_source_id
device regressions, bounded-shutdown entrypoint. 1614 passed.
79 lines
2.4 KiB
Python
79 lines
2.4 KiB
Python
"""Tests for the request access-log middleware (token-label attribution).
|
|
|
|
The middleware emits one structured ``http_request`` line per request, tagged
|
|
with the friendly label of the API token used (from ``auth.api_keys``) so
|
|
traffic can be attributed to a specific client. These tests capture the
|
|
``main.logger`` calls to assert the label is recorded and that no-auth
|
|
endpoints fall back to ``"unauthenticated"``.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def client_and_logs(authenticated_client, monkeypatch):
|
|
"""authenticated_client plus a captured list of (event, kwargs) log calls."""
|
|
import ledgrab.main as main_mod
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
class _FakeLogger:
|
|
def info(self, event, **kw):
|
|
calls.append((event, kw))
|
|
|
|
# main.logger is shared; keep other levels as harmless no-ops.
|
|
def debug(self, *a, **k):
|
|
pass
|
|
|
|
def warning(self, *a, **k):
|
|
pass
|
|
|
|
def error(self, *a, **k):
|
|
pass
|
|
|
|
monkeypatch.setattr(main_mod, "logger", _FakeLogger())
|
|
return authenticated_client, calls
|
|
|
|
|
|
def _http_request_logs(calls):
|
|
return [kw for event, kw in calls if event == "http_request"]
|
|
|
|
|
|
def test_access_log_records_token_label_for_authed_request(client_and_logs):
|
|
client, calls = client_and_logs
|
|
|
|
# /openapi.json requires auth but needs no initialized store, so it returns
|
|
# 200 even without app lifespan — and exercises the auth → label path.
|
|
resp = client.get("/openapi.json")
|
|
assert resp.status_code == 200
|
|
|
|
logs = _http_request_logs(calls)
|
|
assert logs, "no http_request access log emitted"
|
|
entry = logs[-1]
|
|
assert entry["method"] == "GET"
|
|
assert entry["path"] == "/openapi.json"
|
|
assert entry["status"] == 200
|
|
assert entry["token"] == "test" # label of the configured test API key
|
|
assert "duration_ms" in entry
|
|
|
|
|
|
def test_access_log_marks_unauthenticated_for_no_auth_endpoint(client_and_logs):
|
|
client, calls = client_and_logs
|
|
|
|
# /health has no auth dependency, so no label is set on request.state.
|
|
resp = client.get("/health")
|
|
assert resp.status_code == 200
|
|
|
|
logs = _http_request_logs(calls)
|
|
assert logs
|
|
assert logs[-1]["token"] == "unauthenticated"
|
|
|
|
|
|
def test_access_log_never_contains_the_token_secret(client_and_logs):
|
|
client, calls = client_and_logs
|
|
|
|
client.get("/openapi.json")
|
|
|
|
for _event, kw in calls:
|
|
assert "test-api-key-12345" not in repr(kw), "token secret leaked into logs"
|