ElebobaandClaude Opus 5.5
31585c7f35
CI / ruff · pytest (pull_request) Successful in 2m41s
- Idle chat: all messages go to the agent (greetings, side-effect and substitute questions included); check-in signals are still recorded silently, quit intent keeps its rule reaction. Seen live: «какие будут побочки» was logged as a complaint («Спасибо, отметил»), «привет» got a canned phrase. - Agent prompt: warm short replies to small talk; substitutes → only the doctor changes the drug. - ЕМИАС demo persona button creates a private copy too (starting from consent + questionnaires, subject to the screening switch); the web app drops a stored session of the old shared demo account. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
292 lines
15 KiB
Python
292 lines
15 KiB
Python
"""Patient web app API: ЕМИАС (demo) and anonymous sign-in, crisis resources, hand-off to the dialog engine."""
|
|
|
|
from collections.abc import AsyncIterator
|
|
from datetime import datetime, time, timedelta
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.api import fixtures
|
|
from app.core import clock
|
|
from app.db import identity_repo, repo, session
|
|
from app.db.bootstrap import init_db
|
|
from app.db.clinical import Dose, Prescription
|
|
from app.main import app
|
|
from app.services.anonymous_care import doctor_id as anonymous_doctor
|
|
|
|
MEM = "sqlite+aiosqlite:///:memory:"
|
|
TOKEN = "b" * 32
|
|
|
|
|
|
@pytest.fixture
|
|
async def client() -> AsyncIterator[AsyncClient]:
|
|
await session.configure(identity_url=MEM, clinical_url=MEM)
|
|
await init_db()
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
|
yield c
|
|
await session.configure()
|
|
|
|
|
|
async def _patient_with_code(code: str) -> None:
|
|
async with session.get_identity_session() as s:
|
|
await identity_repo.create_patient(
|
|
s,
|
|
token=TOKEN,
|
|
display_name="Синтетический П.",
|
|
snils="000-000-000 00",
|
|
clinic_id="clinic-1",
|
|
doctor_id="doc-1",
|
|
now=clock.now(),
|
|
)
|
|
await identity_repo.create_connect_code(s, code=code, token=TOKEN, doctor_id="doc-1", now=clock.now())
|
|
|
|
|
|
async def test_anonymous_tokens_are_fresh_and_store_no_personal_data(client: AsyncClient) -> None:
|
|
first = (await client.post("/api/v1/auth/anonymous")).json()
|
|
second = (await client.post("/api/v1/auth/anonymous")).json()
|
|
assert first["mode"] == "anonymous" and first["display_name"] is None
|
|
assert len(first["token"]) == 32 and first["token"] != second["token"]
|
|
async with session.get_identity_session() as s:
|
|
assert await identity_repo.get_patient(s, first["token"]) is None
|
|
async with session.get_clinical_session() as s:
|
|
# The care team sees a pseudonymous profile without an identity record.
|
|
profile = await repo.get_profile(s, first["token"])
|
|
assert profile.doctor_id == anonymous_doctor() and profile.modules == []
|
|
|
|
|
|
async def test_anonymous_focus_does_not_remove_access_to_doctor(client: AsyncClient) -> None:
|
|
token = (await client.post("/api/v1/auth/anonymous", json={"focus": "ptsd"})).json()["token"]
|
|
async with session.get_clinical_session() as s:
|
|
profile = await repo.get_profile(s, token)
|
|
assert profile.modules == ["veteran"] and profile.doctor_id == anonymous_doctor()
|
|
assert (await client.get(f"/api/v1/me/{token}/today")).json()["has_doctor"] is True
|
|
assert (await client.post("/api/v1/auth/anonymous", json={"focus": "flu"})).status_code == 422
|
|
|
|
|
|
async def test_anonymous_token_starts_the_engine_dialog(client: AsyncClient) -> None:
|
|
token = (await client.post("/api/v1/auth/anonymous")).json()["token"]
|
|
replies = (await client.post(f"/api/v1/dialog/{token}/events", json={"type": "start"})).json()["replies"]
|
|
buttons = [b["data"] for r in replies for row in r["buttons"] for b in row]
|
|
assert "consent:yes" in buttons and "consent:no" in buttons
|
|
|
|
|
|
async def test_emias_code_is_one_time_and_attaches_doctor(client: AsyncClient) -> None:
|
|
await _patient_with_code("DEMO-0042")
|
|
ok = await client.post("/api/v1/auth/emias", json={"code": " demo-0042 "})
|
|
assert ok.status_code == 200 and ok.json() == {"token": TOKEN, "mode": "emias", "display_name": None}
|
|
async with session.get_clinical_session() as s:
|
|
assert (await repo.get_profile(s, TOKEN)).doctor_id == "doc-1"
|
|
again = await client.post("/api/v1/auth/emias", json={"code": "DEMO-0042"})
|
|
assert again.status_code == 401
|
|
|
|
|
|
async def test_emias_rejects_unknown_code(client: AsyncClient) -> None:
|
|
assert (await client.post("/api/v1/auth/emias", json={"code": "NOPE"})).status_code == 401
|
|
assert (await client.post("/api/v1/auth/emias", json={"code": ""})).status_code == 422
|
|
|
|
|
|
async def test_emias_demo_persona(client: AsyncClient) -> None:
|
|
persona = fixtures.DEMO_PERSONAS[0]
|
|
res = (await client.post("/api/v1/auth/emias/demo", json={"persona_token": persona.token})).json()
|
|
# Every visitor gets a private copy of the persona, never the shared account.
|
|
assert res["token"].startswith(persona.token + "-") and res["mode"] == "emias"
|
|
assert res["display_name"] == persona.display_name
|
|
again = (await client.post("/api/v1/auth/emias/demo", json={"persona_token": persona.token})).json()
|
|
assert again["token"] != res["token"]
|
|
async with session.get_clinical_session() as s:
|
|
assert (await repo.get_profile(s, res["token"])).doctor_id == fixtures.DOCTOR_ID
|
|
missing = await client.post("/api/v1/auth/emias/demo", json={"persona_token": "tok-nobody"})
|
|
assert missing.status_code == 404
|
|
|
|
|
|
async def test_today_plan_and_week_marks(client: AsyncClient) -> None:
|
|
now = clock.now()
|
|
today = now.date()
|
|
async with session.get_clinical_session() as s:
|
|
await repo.upsert_profile(s, token=TOKEN, doctor_id="doc-1", next_visit=today + timedelta(days=5))
|
|
rx = Prescription(token=TOKEN, drug="sertraline", dose="50 мг", schedule={"times": ["09:00", "21:00"]})
|
|
s.add(rx)
|
|
await s.flush()
|
|
morning = datetime.combine(today, time(9, 0), now.tzinfo)
|
|
s.add_all(
|
|
[
|
|
Dose(token=TOKEN, prescription_id=rx.id, planned_at=morning, status="taken"),
|
|
Dose(token=TOKEN, prescription_id=rx.id, planned_at=morning + timedelta(hours=12)),
|
|
]
|
|
)
|
|
await repo.upsert_checkin(s, token=TOKEN, day=today, now=now, mood=6)
|
|
|
|
body = (await client.get(f"/api/v1/me/{TOKEN}/today")).json()
|
|
assert body["has_doctor"] and body["checkin_done"]
|
|
assert body["next_visit"] == (today + timedelta(days=5)).isoformat()
|
|
assert [(d["time"], d["status"]) for d in body["doses"]] == [("09:00", "taken"), ("21:00", "pending")]
|
|
week = body["prescriptions"][0]["week"]
|
|
assert len(week) == 7 and next(w for w in week if w["day"] == today.isoformat())["status"] == "pending"
|
|
|
|
|
|
async def test_today_for_anonymous_is_empty(client: AsyncClient) -> None:
|
|
token = (await client.post("/api/v1/auth/anonymous")).json()["token"]
|
|
body = (await client.get(f"/api/v1/me/{token}/today")).json()
|
|
assert body["has_doctor"] is True and body["doses"] == [] and body["prescriptions"] == []
|
|
|
|
|
|
async def test_crisis_resources_come_from_config(client: AsyncClient) -> None:
|
|
phones = [r["phone"] for r in (await client.get("/api/v1/crisis/resources")).json()]
|
|
assert "112" in phones
|
|
|
|
|
|
async def test_today_links_prescription_to_reference_and_therapy_day(client: AsyncClient) -> None:
|
|
today = clock.now().date()
|
|
async with session.get_clinical_session() as s:
|
|
await repo.upsert_profile(s, token=TOKEN, doctor_id="doc-1", modules=["veteran"])
|
|
s.add(Prescription(token=TOKEN, drug="Сертралин", dose="50 мг", schedule={}, start_date=today - timedelta(13)))
|
|
rx = (await client.get(f"/api/v1/me/{TOKEN}/today")).json()["prescriptions"][0]
|
|
assert rx["ref_id"] == "sertraline" and rx["indication"] == "ptsd" and rx["therapy_day"] == 14
|
|
|
|
|
|
async def test_demo_persona_week_mirrors_dashboard_story(client: AsyncClient) -> None:
|
|
auth = await client.post("/api/v1/auth/emias/demo", json={"persona_token": fixtures.DEMO_PERSONAS[0].token})
|
|
persona = auth.json()["token"]
|
|
rx = (await client.get(f"/api/v1/me/{persona}/today")).json()["prescriptions"][0]
|
|
assert rx["ref_id"] == "sertraline" and rx["therapy_day"] > 1
|
|
|
|
|
|
async def test_questionnaires_are_off_without_release_or_demo_flag(
|
|
client: AsyncClient, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.delenv("INTAKE_DEMO", raising=False)
|
|
async with session.get_clinical_session() as s:
|
|
await repo.upsert_profile(s, token=TOKEN, doctor_id="doc-1")
|
|
q = (await client.get(f"/api/v1/me/{TOKEN}/today")).json()["questionnaires"]
|
|
assert q == {"enabled": False, "intake": "none", "diary_done": False, "diary_next": "morning"}
|
|
|
|
|
|
async def test_demo_flag_opens_intake_in_the_engine(client: AsyncClient, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("INTAKE_DEMO", "1")
|
|
auth = await client.post("/api/v1/auth/emias/demo", json={"persona_token": fixtures.DEMO_PERSONAS[0].token})
|
|
persona = auth.json()["token"]
|
|
|
|
async def press(button: str | None = None) -> list[str]:
|
|
event = {"type": "button", "button": button} if button else {"type": "start"}
|
|
replies = (await client.post(f"/api/v1/dialog/{persona}/events", json=event)).json()["replies"]
|
|
return [b["data"] for r in replies for row in r["buttons"] for b in row]
|
|
|
|
await press()
|
|
await press("consent:yes")
|
|
buttons = await press("age:adult")
|
|
# Stage 1 starts with the red-flag block, then the order of the MVP questionnaire.
|
|
assert any(b.startswith("iq:") and ":rf_psychosis:" in b for b in buttons)
|
|
assert "intake:help" in buttons and "intake:pause" in buttons
|
|
q = (await client.get(f"/api/v1/me/{persona}/today")).json()["questionnaires"]
|
|
assert q["enabled"] and q["intake"] == "active"
|
|
|
|
|
|
async def test_instrument_catalog_describes_intake_items(client: AsyncClient) -> None:
|
|
body = (await client.get("/api/v1/instruments/catalog")).json()
|
|
phq = body["items"]["phq9_3"]
|
|
assert phq["form"] == "phq9" and phq["index"] == 3 and phq["total"] == 9 and not phq["adaptive"]
|
|
assert body["items"]["cssrs_q1"]["adaptive"] and body["forms"]["cssrs"]["section"] == "safety"
|
|
assert body["items"]["sleep_hours"]["numeric"] and body["items"]["alcohol_portions"]["numeric"]
|
|
assert body["items"]["dphq_9"]["form"] == "diary" and body["items"]["dpcl_18"]["form"] == "diary"
|
|
assert body["forms"]["pcl5"]["recall_period"] and body["forms"]["gad7"]["section"] == "scale"
|
|
drinks = body["standard_drinks"]
|
|
assert len(drinks["header"]) == 4 and len(drinks["rows"]) == 5 and all(len(c) == 4 for _, c in drinks["rows"])
|
|
|
|
|
|
async def test_meds_reference_is_sourced_and_hides_rule_params(client: AsyncClient) -> None:
|
|
body = (await client.get("/api/v1/meds/reference")).json()
|
|
assert "rules" not in body and body["texts"]["dont_stop_alone"]
|
|
drugs = {d["id"]: d for d in body["drugs"]}
|
|
sertraline = drugs["sertraline"]
|
|
assert sertraline["indications"]["depression"]["max_mg"] == 200
|
|
assert (
|
|
sertraline["indications"]["ptsd"]["min_mg"] == 25 and "КР ПТСР" in sertraline["indications"]["ptsd"]["source"]
|
|
)
|
|
assert drugs["diazepam"]["max_course_days"] == 28 and drugs["alprazolam"]["max_course_days"] == 90
|
|
assert all(ind["source"] for d in body["drugs"] for ind in d["indications"].values())
|
|
|
|
|
|
async def test_stage_one_starts_without_modules(client: AsyncClient, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
# Stage 1 is the same for everyone: it does not wait for a module (it sets them itself).
|
|
monkeypatch.setenv("INTAKE_DEMO", "1")
|
|
token = (await client.post("/api/v1/auth/anonymous")).json()["token"]
|
|
|
|
async def press(button: str | None = None) -> list[str]:
|
|
event = {"type": "button", "button": button} if button else {"type": "start"}
|
|
replies = (await client.post(f"/api/v1/dialog/{token}/events", json=event)).json()["replies"]
|
|
return [b["data"] for r in replies for row in r["buttons"] for b in row]
|
|
|
|
assert (await client.get(f"/api/v1/me/{token}/today")).json()["questionnaires"]["enabled"]
|
|
await press()
|
|
await press("consent:yes")
|
|
assert any(":rf_psychosis:" in b for b in await press("age:adult"))
|
|
|
|
|
|
async def test_patient_app_has_its_own_demo_personas(client: AsyncClient) -> None:
|
|
body = (await client.get("/api/v1/patient/demo-personas")).json()
|
|
assert [p["token"] for p in body] == [p.token for p in fixtures.DEMO_PERSONAS]
|
|
|
|
|
|
async def test_anonymous_is_visible_only_to_assigned_doctor(client: AsyncClient) -> None:
|
|
first = (await client.post("/api/v1/auth/anonymous")).json()["token"]
|
|
second = (await client.post("/api/v1/auth/anonymous")).json()["token"]
|
|
items = (await client.get(f"/api/v1/doctors/{anonymous_doctor()}/patients")).json()
|
|
names = {p["token"]: p["display_name"] for p in items}
|
|
assert names[first].startswith("Пациент ") and names[first] != names[second]
|
|
assert first not in names[first]
|
|
own = await client.get(f"/api/v1/patients/{first}", params={"doctor_id": anonymous_doctor()})
|
|
assert own.status_code == 200 and own.json()["patient"]["display_name"] == names[first]
|
|
denied = await client.get(f"/api/v1/patients/{first}", params={"doctor_id": "another-doctor"})
|
|
assert denied.status_code == 403
|
|
assert (await client.get("/api/v1/doctors/another-doctor/patients")).json() == []
|
|
|
|
|
|
async def test_old_anonymous_profile_requires_explicit_care_opt_in(client: AsyncClient) -> None:
|
|
async with session.get_clinical_session() as s:
|
|
await repo.upsert_profile(s, token=TOKEN, doctor_id="", modules=["veteran"])
|
|
# Merely opening Today must not reveal an old private profile to a doctor.
|
|
assert not (await client.get(f"/api/v1/me/{TOKEN}/today")).json()["has_doctor"]
|
|
assert (await client.get(f"/api/v1/doctors/{anonymous_doctor()}/patients")).json() == []
|
|
for _ in range(2):
|
|
assert (await client.post(f"/api/v1/me/{TOKEN}/doctor/connect")).json() == {"has_doctor": True}
|
|
async with session.get_clinical_session() as s:
|
|
p = await repo.get_profile(s, TOKEN)
|
|
assert p.doctor_id == anonymous_doctor() and p.modules == ["veteran"]
|
|
async with session.get_identity_session() as s:
|
|
assert await identity_repo.get_patient(s, TOKEN) is None
|
|
|
|
|
|
async def test_connect_care_does_not_reassign_or_create_patients(client: AsyncClient) -> None:
|
|
assert (await client.post(f"/api/v1/me/{TOKEN}/doctor/connect")).status_code == 404
|
|
async with session.get_clinical_session() as s:
|
|
await repo.upsert_profile(s, token=TOKEN, doctor_id="own-doctor")
|
|
assert (await client.post(f"/api/v1/me/{TOKEN}/doctor/connect")).status_code == 200
|
|
async with session.get_clinical_session() as s:
|
|
assert (await repo.get_profile(s, TOKEN)).doctor_id == "own-doctor"
|
|
|
|
|
|
async def test_screening_capability_respects_release_gate(client: AsyncClient, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("INTAKE_DEMO", "0")
|
|
assert (await client.get("/api/v1/patient/capabilities")).json() == {"primary_screening": False}
|
|
monkeypatch.setenv("INTAKE_DEMO", "1")
|
|
assert (await client.get("/api/v1/patient/capabilities")).json() == {"primary_screening": True}
|
|
|
|
|
|
async def test_anonymous_contact_request_reaches_doctor(client: AsyncClient, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("INTAKE_DEMO", "0")
|
|
token = (await client.post("/api/v1/auth/anonymous")).json()["token"]
|
|
for event in [
|
|
{"type": "start"},
|
|
{"type": "button", "button": "consent:yes"},
|
|
{"type": "button", "button": "age:adult"},
|
|
{"type": "button", "button": "menu:contact_doctor"},
|
|
]:
|
|
response = await client.post(f"/api/v1/dialog/{token}/events", json=event)
|
|
assert response.status_code == 200
|
|
items = (await client.get(f"/api/v1/doctors/{anonymous_doctor()}/patients")).json()
|
|
assert items[0]["last_event"] == "Пациент просит врача связаться"
|
|
card = (await client.get(f"/api/v1/patients/{token}", params={"doctor_id": anonymous_doctor()})).json()
|
|
assert card["patient"]["last_event"] == items[0]["last_event"]
|
|
async with session.get_identity_session() as s:
|
|
assert await identity_repo.get_patient(s, token) is None
|