70 lines
2.9 KiB
Python
70 lines
2.9 KiB
Python
"""Periodic planning reports must not serialize/compress on the ASGI loop."""
|
|
import threading
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from k1link.web import planning_live_api
|
|
from k1link.web.response_compression import ResponseCompressionMiddleware
|
|
|
|
|
|
@pytest.mark.parametrize("encoding", ["gzip", "identity"])
|
|
def test_active_report_keeps_content_and_encodes_off_loop(monkeypatch, encoding):
|
|
document = {"state": "completed", "evidence": [{"label": "Проверка", "x": 1.25}] * 50}
|
|
calls = []
|
|
original_json = planning_live_api.JSONResponse
|
|
original_compress = planning_live_api.gzip.compress
|
|
|
|
class RecordedJSON(original_json):
|
|
def render(self, content):
|
|
calls.append(("json", threading.get_ident()))
|
|
return super().render(content)
|
|
|
|
def compress(*args, **kwargs):
|
|
calls.append(("gzip", threading.get_ident()))
|
|
return original_compress(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(planning_live_api, "JSONResponse", RecordedJSON)
|
|
monkeypatch.setattr(planning_live_api.gzip, "compress", compress)
|
|
app = FastAPI()
|
|
app.include_router(planning_live_api.build_planning_live_router(SimpleNamespace(get=lambda: document)))
|
|
app.add_middleware(ResponseCompressionMiddleware)
|
|
|
|
@app.get("/loop-thread")
|
|
async def loop_thread():
|
|
return threading.get_ident()
|
|
|
|
with TestClient(app) as client:
|
|
loop_id = client.get("/loop-thread").json()
|
|
result = client.get("/api/v1/mission-planner/live-tests/active", headers={"Accept-Encoding": encoding})
|
|
assert result.status_code == 200
|
|
assert result.json() == document # Includes HTTP decompression; no double gzip.
|
|
assert result.headers["cache-control"] == "no-store"
|
|
assert result.headers["content-type"] == "application/json"
|
|
assert [name for name, _ in calls] == (["json", "gzip"] if encoding == "gzip" else ["json"])
|
|
assert all(thread != loop_id for _, thread in calls)
|
|
if encoding == "gzip":
|
|
assert result.headers["content-encoding"] == "gzip"
|
|
assert result.headers["vary"] == "Accept-Encoding"
|
|
else:
|
|
assert "content-encoding" not in result.headers
|
|
|
|
|
|
def test_active_report_retains_empty_and_failure_contract():
|
|
service = SimpleNamespace(get=lambda: None)
|
|
app = FastAPI()
|
|
app.include_router(planning_live_api.build_planning_live_router(service))
|
|
with TestClient(app) as client:
|
|
result = client.get("/api/v1/mission-planner/live-tests/active")
|
|
assert result.status_code == 200 and result.json() is None
|
|
|
|
def fail():
|
|
raise RuntimeError("synthetic unavailable report")
|
|
|
|
service.get = fail
|
|
result = client.get("/api/v1/mission-planner/live-tests/active")
|
|
assert result.status_code == 409
|
|
assert result.json() == {"detail": "synthetic unavailable report"}
|