ElebobaandClaude Opus 5.5
2582c73d12
CI / ruff · pytest (pull_request) Successful in 2m1s
- Patient list sorted red -> yellow -> grey -> green with KPI row, trend arrows, 7-day adherence bars, discontinuation-risk badge with factor tooltip. - Patient card: scales with severity bands from config/scales (unscheduled marked), mood/sleep/alcohol small multiples, dose heat-strip, alerts feed with ack, safety plan, AI summary with disclaimer. - Live alerts: st.fragment(run_every=2.5s) polling, toast + pinned acute banner on every page; full rerun on new alert only on doctor pages. - Connect form -> POST /patients/connect -> code, QR, deep link. - Thin httpx ApiClient over the frozen schemas with graceful "API недоступен". - app.py discovers dashboard/pages/*.py automatically (st.navigation); demo pages of #17 plug in by adding a file. Refs #16 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
127 lines
4.3 KiB
Python
127 lines
4.3 KiB
Python
"""Dashboard HTTP client: paths, parsing via the frozen schemas, graceful errors (no tracebacks to the UI)."""
|
||
|
||
import json
|
||
|
||
import httpx
|
||
import pytest
|
||
|
||
from app.api import fixtures
|
||
from app.api.schemas import ConnectPatientIn
|
||
from dashboard.api_client import ApiClient, ApiError, ApiUnavailable
|
||
|
||
|
||
def _client(handler) -> ApiClient:
|
||
return ApiClient("http://api.test", "doc-smirnova", transport=httpx.MockTransport(handler))
|
||
|
||
|
||
def _json(models) -> list | dict:
|
||
if isinstance(models, list):
|
||
return [m.model_dump(mode="json") for m in models]
|
||
return models.model_dump(mode="json")
|
||
|
||
|
||
def test_base_url_gets_api_prefix_once():
|
||
assert ApiClient("http://x:8000/").base_url == "http://x:8000/api/v1"
|
||
assert ApiClient("http://x:8000/api/v1").base_url == "http://x:8000/api/v1"
|
||
|
||
|
||
def test_doctor_id_from_env(monkeypatch):
|
||
monkeypatch.setenv("DOCTOR_ID", "doc-other")
|
||
assert ApiClient("http://x").doctor_id == "doc-other"
|
||
monkeypatch.delenv("DOCTOR_ID")
|
||
assert ApiClient("http://x").doctor_id == "doc-smirnova"
|
||
|
||
|
||
def test_patients_and_card_are_parsed():
|
||
seen: list[str] = []
|
||
|
||
def handler(req: httpx.Request) -> httpx.Response:
|
||
seen.append(str(req.url))
|
||
if req.url.path.endswith("/patients"):
|
||
return httpx.Response(200, json=_json(fixtures.PATIENTS))
|
||
return httpx.Response(200, json=_json(fixtures.andrey_card()))
|
||
|
||
c = _client(handler)
|
||
patients = c.patients()
|
||
assert patients[0].display_name == "Андрей С."
|
||
card = c.patient_card("tok-andrey")
|
||
assert card.patient.status_color == "red"
|
||
assert seen[0] == "http://api.test/api/v1/doctors/doc-smirnova/patients"
|
||
assert seen[1] == "http://api.test/api/v1/patients/tok-andrey?doctor_id=doc-smirnova"
|
||
|
||
|
||
def test_alerts_params_and_ack_body():
|
||
captured: dict = {}
|
||
|
||
def handler(req: httpx.Request) -> httpx.Response:
|
||
if req.method == "POST":
|
||
captured["body"] = json.loads(req.content)
|
||
return httpx.Response(200, json=_json(fixtures.all_alerts()[0]))
|
||
captured["params"] = dict(req.url.params)
|
||
return httpx.Response(200, json=_json(fixtures.all_alerts()))
|
||
|
||
c = _client(handler)
|
||
assert len(c.alerts(since_id=2, open_only=True)) == 4
|
||
assert captured["params"] == {"since_id": "2", "open_only": "true"}
|
||
c.ack(3)
|
||
assert captured["body"] == {"doctor_id": "doc-smirnova"}
|
||
|
||
|
||
def test_connect_posts_payload():
|
||
def handler(req: httpx.Request) -> httpx.Response:
|
||
body = json.loads(req.content)
|
||
assert body["drug"] == "сертралин" and body["times"] == ["08:00"]
|
||
return httpx.Response(200, json={"token": "t", "code": "DEMO-1", "deep_link": "https://t.me/x?start=DEMO-1"})
|
||
|
||
body = ConnectPatientIn(
|
||
doctor_id="doc-smirnova",
|
||
display_name="Андрей С.",
|
||
birth_year=1992,
|
||
sex="m",
|
||
snils="000-000-000 00",
|
||
drug="сертралин",
|
||
dose="50 мг",
|
||
)
|
||
assert _client(handler).connect(body).code == "DEMO-1"
|
||
|
||
|
||
def test_connection_error_becomes_unavailable():
|
||
def handler(req: httpx.Request) -> httpx.Response:
|
||
raise httpx.ConnectError("refused", request=req)
|
||
|
||
with pytest.raises(ApiUnavailable) as exc:
|
||
_client(handler).patients()
|
||
assert "API недоступен" in exc.value.message
|
||
|
||
|
||
def test_timeout_becomes_unavailable():
|
||
def handler(req: httpx.Request) -> httpx.Response:
|
||
raise httpx.ReadTimeout("slow", request=req)
|
||
|
||
with pytest.raises(ApiUnavailable):
|
||
_client(handler).alerts()
|
||
|
||
|
||
@pytest.mark.parametrize(("status", "fragment"), [(404, "Не найдено"), (500, "500"), (422, "422")])
|
||
def test_http_errors(status, fragment):
|
||
c = _client(lambda req: httpx.Response(status, json={"detail": "x"}))
|
||
with pytest.raises(ApiError) as exc:
|
||
c.patient_card("tok-x")
|
||
assert fragment in exc.value.message
|
||
assert exc.value.status == status
|
||
assert not isinstance(exc.value, ApiUnavailable)
|
||
|
||
|
||
def test_bad_payload_is_api_error():
|
||
c = _client(lambda req: httpx.Response(200, json=[{"token": "only"}]))
|
||
with pytest.raises(ApiError, match="неожиданный"):
|
||
c.patients()
|
||
c = _client(lambda req: httpx.Response(200, content=b"<html>"))
|
||
with pytest.raises(ApiError):
|
||
c.patients()
|
||
|
||
|
||
def test_clock_failure_is_none():
|
||
c = _client(lambda req: httpx.Response(503))
|
||
assert c.clock() is None
|