Arxip222andClaude Opus 5.5
b8a93ce9d4
A second channel to the same dialog engine as the Telegram bot: consent, onboarding, scales and check-ins come from app/dialog; the bot is unchanged. - API: /auth/emias (doctor's one-time connect code via EmiasMock), /auth/emias/demo (synthetic personas), /auth/anonymous (random token, nothing stored in the identity DB), /crisis/resources from crisis.yaml - web/: React + TS + Vite, mobile-first, safe areas, hash routing, tokens and texts isolated for a Capacitor mobile build Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
87 lines
3.8 KiB
Python
87 lines
3.8 KiB
Python
"""Patient web app API: ЕМИАС (demo) and anonymous sign-in, crisis resources, hand-off to the dialog engine."""
|
|
|
|
from collections.abc import AsyncIterator
|
|
|
|
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.main import app
|
|
|
|
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_nothing(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:
|
|
assert await repo.get_profile(s, first["token"]) is None
|
|
|
|
|
|
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})
|
|
assert res.json() == {"token": persona.token, "mode": "emias", "display_name": persona.display_name}
|
|
async with session.get_clinical_session() as s:
|
|
assert (await repo.get_profile(s, persona.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_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
|