292 lines
13 KiB
Python
292 lines
13 KiB
Python
"""Data model tests on two separate in-memory SQLite databases (identity / clinical).
|
|
|
|
Set TEST_PG_IDENTITY_URL / TEST_PG_CLINICAL_URL to run the same suite against compose Postgres.
|
|
"""
|
|
|
|
import ast
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from collections.abc import AsyncIterator, Callable
|
|
from datetime import UTC, date, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from sqlalchemy import inspect, select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncEngine
|
|
|
|
from app.core.types import AlertLevel, ScrubbedText
|
|
from app.db import identity_repo, repo, session
|
|
from app.db.audit import AuditLog, audit
|
|
from app.db.bootstrap import init_db
|
|
from app.db.clinical import ClinicalBase, DailyCheckin, Dose, Prescription
|
|
from app.db.identity import IdentityBase
|
|
|
|
DB_DIR = Path(__file__).resolve().parent.parent / "app" / "db"
|
|
T0 = datetime(2026, 10, 3, 9, 0, tzinfo=UTC)
|
|
TOKEN = "a" * 32
|
|
TOKEN2 = "b" * 32
|
|
|
|
|
|
@pytest.fixture
|
|
async def dbs() -> AsyncIterator[None]:
|
|
await session.configure(
|
|
identity_url=os.environ.get("TEST_PG_IDENTITY_URL", "sqlite+aiosqlite:///:memory:"),
|
|
clinical_url=os.environ.get("TEST_PG_CLINICAL_URL", "sqlite+aiosqlite:///:memory:"),
|
|
)
|
|
if "TEST_PG_IDENTITY_URL" in os.environ:
|
|
async with session.identity_engine().begin() as conn:
|
|
await conn.run_sync(IdentityBase.metadata.drop_all)
|
|
async with session.clinical_engine().begin() as conn:
|
|
await conn.run_sync(ClinicalBase.metadata.drop_all)
|
|
await init_db()
|
|
yield
|
|
await session.configure()
|
|
|
|
|
|
async def _tables(engine_fn: Callable[[], AsyncEngine]) -> set[str]:
|
|
async with engine_fn().connect() as conn:
|
|
return set(await conn.run_sync(lambda c: inspect(c).get_table_names()))
|
|
|
|
|
|
async def test_create_all_puts_tables_in_the_right_db(dbs: None) -> None:
|
|
identity_tables = await _tables(session.identity_engine)
|
|
clinical_tables = await _tables(session.clinical_engine)
|
|
assert identity_tables == {
|
|
"patients",
|
|
"doctors",
|
|
"connect_codes",
|
|
"max_bindings",
|
|
"max_link_codes",
|
|
"max_webhook_receipts",
|
|
"max_deliveries",
|
|
}
|
|
assert {
|
|
"patient_profiles",
|
|
"item_responses",
|
|
"scale_results",
|
|
"daily_checkins",
|
|
"prescriptions",
|
|
"doses",
|
|
"alerts",
|
|
"messages",
|
|
"dialog_state",
|
|
"safety_plans",
|
|
"clock_state",
|
|
"audit_log",
|
|
"questionnaire_records",
|
|
"questionnaire_notices",
|
|
"questionnaire_tasks",
|
|
"scheduler_events",
|
|
} == clinical_tables
|
|
await init_db() # idempotent
|
|
|
|
|
|
async def test_identity_roundtrip_and_connect_code(dbs: None) -> None:
|
|
async with session.get_identity_session() as s:
|
|
await identity_repo.upsert_doctor(s, doctor_id="d1", full_name="Доктор Тест", clinic_id="c1", on_call=True)
|
|
await identity_repo.create_patient(
|
|
s,
|
|
token=TOKEN,
|
|
display_name="Пациент",
|
|
snils="000-000-000 00",
|
|
clinic_id="c1",
|
|
doctor_id="d1",
|
|
now=T0,
|
|
)
|
|
await identity_repo.create_connect_code(s, code="ABC123", token=TOKEN, doctor_id="d1", now=T0)
|
|
async with session.get_identity_session() as s:
|
|
assert await identity_repo.redeem_connect_code(s, code="ABC123", now=T0) == TOKEN
|
|
assert await identity_repo.redeem_connect_code(s, code="ABC123", now=T0) is None
|
|
assert await identity_repo.redeem_connect_code(s, code="NOPE", now=T0) is None
|
|
await identity_repo.bind_chat(s, token=TOKEN, tg_chat_id=42)
|
|
patient = await identity_repo.record_consent(s, token=TOKEN, version="v1", now=T0)
|
|
assert patient is not None and identity_repo.has_active_consent(patient)
|
|
async with session.get_identity_session() as s:
|
|
patient = await identity_repo.get_patient_by_chat(s, 42)
|
|
assert patient is not None and patient.token == TOKEN
|
|
assert patient.created_at == T0 and patient.created_at.tzinfo is not None
|
|
await identity_repo.revoke_consent(s, token=TOKEN, now=T0 + timedelta(days=1))
|
|
assert not identity_repo.has_active_consent(patient)
|
|
assert [d.doctor_id for d in await identity_repo.doctors_on_call(s)] == ["d1"]
|
|
|
|
|
|
async def test_snils_and_chat_unique(dbs: None) -> None:
|
|
async with session.get_identity_session() as s:
|
|
await identity_repo.create_patient(
|
|
s, token=TOKEN, display_name="A", snils="111", clinic_id="c", doctor_id="d", now=T0
|
|
)
|
|
with pytest.raises(IntegrityError):
|
|
async with session.get_identity_session() as s:
|
|
await identity_repo.create_patient(
|
|
s, token=TOKEN2, display_name="B", snils="111", clinic_id="c", doctor_id="d", now=T0
|
|
)
|
|
async with session.get_identity_session() as s:
|
|
await identity_repo.create_patient(
|
|
s, token=TOKEN2, display_name="B", snils="222", clinic_id="c", doctor_id="d", now=T0
|
|
)
|
|
await identity_repo.bind_chat(s, token=TOKEN, tg_chat_id=7)
|
|
with pytest.raises(IntegrityError):
|
|
async with session.get_identity_session() as s:
|
|
await identity_repo.bind_chat(s, token=TOKEN2, tg_chat_id=7)
|
|
|
|
|
|
async def test_checkin_upsert_merges_and_is_unique_per_day(dbs: None) -> None:
|
|
day = date(2026, 10, 3)
|
|
async with session.get_clinical_session() as s:
|
|
await repo.upsert_checkin(s, token=TOKEN, day=day, now=T0, mood=4)
|
|
await repo.upsert_checkin(s, token=TOKEN, day=day, now=T0, med_taken=True, side_effects=["тошнота"])
|
|
await repo.upsert_checkin(s, token=TOKEN, day=day + timedelta(days=1), now=T0, mood=3)
|
|
async with session.get_clinical_session() as s:
|
|
rows = await repo.list_checkins(s, TOKEN)
|
|
assert len(rows) == 2
|
|
assert (rows[0].mood, rows[0].med_taken, rows[0].side_effects) == (4, True, ["тошнота"])
|
|
assert [r.day for r in await repo.list_checkins(s, TOKEN, since=day + timedelta(days=1))] == [
|
|
day + timedelta(days=1)
|
|
]
|
|
with pytest.raises(ValueError):
|
|
await repo.upsert_checkin(s, token=TOKEN, day=day, now=T0, bogus=1)
|
|
with pytest.raises(IntegrityError):
|
|
async with session.get_clinical_session() as s:
|
|
s.add(DailyCheckin(token=TOKEN, day=day, side_effects=[], updated_at=T0))
|
|
|
|
|
|
async def test_scales_profiles_prescriptions(dbs: None) -> None:
|
|
async with session.get_clinical_session() as s:
|
|
await repo.upsert_profile(s, token=TOKEN, doctor_id="d1", modules=["veteran", "depression"])
|
|
await repo.upsert_profile(s, token=TOKEN2, doctor_id="d2")
|
|
await repo.upsert_profile(s, token=TOKEN, next_visit=date(2026, 10, 20))
|
|
await repo.add_item_response(s, token=TOKEN, scale="phq9", item_id="9", score=1, source="button", ts=T0)
|
|
await repo.add_scale_result(
|
|
s,
|
|
token=TOKEN,
|
|
scale="phq9",
|
|
total=16,
|
|
severity="moderately_severe",
|
|
complete=True,
|
|
ts=T0,
|
|
item9=1,
|
|
scheduled=False,
|
|
flags={"trigger": "low_mood_3d"},
|
|
)
|
|
rx = Prescription(token=TOKEN, drug="sertraline", dose="50 mg", schedule={"times": ["08:00"]})
|
|
s.add(rx)
|
|
await s.flush()
|
|
s.add(Dose(token=TOKEN, prescription_id=rx.id, planned_at=T0))
|
|
async with session.get_clinical_session() as s:
|
|
profiles = await repo.patients_of_doctor(s, "d1")
|
|
assert [(p.token, p.modules, p.next_visit) for p in profiles] == [
|
|
(TOKEN, ["veteran", "depression"], date(2026, 10, 20))
|
|
]
|
|
(result,) = await repo.list_scale_results(s, TOKEN, "phq9")
|
|
assert (result.item9, result.scheduled, result.flags) == (1, False, {"trigger": "low_mood_3d"})
|
|
dose = await s.scalar(select(Dose))
|
|
assert dose is not None and dose.status == "pending" and dose.planned_at == T0
|
|
|
|
|
|
async def test_ack_alert_writes_audit(dbs: None) -> None:
|
|
async with session.get_clinical_session() as s:
|
|
alert = await repo.add_alert(
|
|
s,
|
|
token=TOKEN,
|
|
level=AlertLevel.ACUTE,
|
|
code="phq9_item9",
|
|
reason="п.9 ≥ 1",
|
|
ts=T0,
|
|
params={"item9": 1},
|
|
)
|
|
await repo.add_alert(s, token=TOKEN, level=AlertLevel.INFO, code="silence", reason="молчание", ts=T0)
|
|
alert_id = alert.id
|
|
assert await repo.open_alert_codes(s, TOKEN) == {"phq9_item9", "silence"}
|
|
later = T0 + timedelta(minutes=5)
|
|
async with session.get_clinical_session() as s:
|
|
acked = await repo.ack_alert(s, alert_id=alert_id, doctor_id="d1", ts=later)
|
|
assert acked is not None and acked.acked_by == "d1"
|
|
await repo.ack_alert(s, alert_id=alert_id, doctor_id="d2", ts=later) # no-op
|
|
assert await repo.ack_alert(s, alert_id=9999, doctor_id="d1", ts=later) is None
|
|
async with session.get_clinical_session() as s:
|
|
assert await repo.open_alert_codes(s, TOKEN) == {"silence"}
|
|
(stored,) = [a for a in await repo.list_alerts(s, token=TOKEN) if a.id == alert_id]
|
|
assert (stored.level, stored.acked_by, stored.acked_at) == ("acute", "d1", later)
|
|
logs = (await s.scalars(select(AuditLog))).all()
|
|
assert [(a.actor, a.action, a.token, a.ts) for a in logs] == [("d1", "ack_alert", TOKEN, later)]
|
|
assert logs[0].details == {"alert_id": alert_id, "code": "phq9_item9", "level": "acute"}
|
|
|
|
|
|
async def test_audit_helper(dbs: None) -> None:
|
|
async with session.get_clinical_session() as s:
|
|
await audit(s, actor="system", action="crisis_acute", token=TOKEN, details=None, ts=T0)
|
|
async with session.get_clinical_session() as s:
|
|
entry = await s.scalar(select(AuditLog))
|
|
assert entry is not None and entry.action == "crisis_acute"
|
|
|
|
|
|
async def test_dialog_state_messages_clock(dbs: None) -> None:
|
|
async with session.get_clinical_session() as s:
|
|
assert await repo.get_dialog_state(s, TOKEN) is None
|
|
await repo.set_dialog_state(s, token=TOKEN, state="checkin_mood", data={"step": 1}, now=T0)
|
|
await repo.add_message(s, token=TOKEN, direction="in", text=ScrubbedText("ок"), ts=T0, fsm_state="idle")
|
|
with pytest.raises(ValueError):
|
|
await repo.add_message(s, token=TOKEN, direction="x", text=ScrubbedText("ок"), ts=T0)
|
|
assert await repo.load_clock_offset(s) == 0.0
|
|
await repo.save_clock_offset(s, 86400.0)
|
|
async with session.get_clinical_session() as s:
|
|
await repo.set_dialog_state(s, token=TOKEN, state="checkin_med", data={"step": 2}, now=T0)
|
|
await repo.save_clock_offset(s, 2 * 86400.0)
|
|
async with session.get_clinical_session() as s:
|
|
state = await repo.get_dialog_state(s, TOKEN)
|
|
assert state is not None and (state.state, state.data) == ("checkin_med", {"step": 2})
|
|
assert await repo.load_clock_offset(s) == 2 * 86400.0
|
|
|
|
|
|
async def test_naive_datetime_rejected(dbs: None) -> None:
|
|
with pytest.raises(Exception, match="naive datetime"):
|
|
async with session.get_clinical_session() as s:
|
|
await repo.add_alert(s, token=TOKEN, level=AlertLevel.INFO, code="x", reason="x", ts=datetime(2026, 1, 1))
|
|
|
|
|
|
async def test_session_rolls_back_on_error(dbs: None) -> None:
|
|
with pytest.raises(RuntimeError):
|
|
async with session.get_clinical_session() as s:
|
|
await repo.upsert_profile(s, token=TOKEN, doctor_id="d1")
|
|
raise RuntimeError
|
|
async with session.get_clinical_session() as s:
|
|
assert await repo.get_profile(s, TOKEN) is None
|
|
|
|
|
|
# --- guardrail #5: clinical code never touches identity ----------------------------------------------
|
|
|
|
CLINICAL_MODULES = ["clinical.py", "repo.py", "audit.py", "session.py", "types.py"]
|
|
|
|
|
|
def _imported_modules(path: Path) -> set[str]:
|
|
tree = ast.parse(path.read_text(encoding="utf-8"))
|
|
names: set[str] = set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
names.update(alias.name for alias in node.names)
|
|
elif isinstance(node, ast.ImportFrom):
|
|
names.add(node.module or "")
|
|
names.update(f"{node.module}.{alias.name}" for alias in node.names)
|
|
return names
|
|
|
|
|
|
@pytest.mark.parametrize("filename", CLINICAL_MODULES)
|
|
def test_clinical_code_does_not_import_identity(filename: str) -> None:
|
|
imported = _imported_modules(DB_DIR / filename)
|
|
assert not [m for m in imported if "identity" in m], f"{filename} imports identity: {imported}"
|
|
|
|
|
|
def test_clinical_import_graph_excludes_identity() -> None:
|
|
code = "import sys, app.db.repo, app.db.audit; print('app.db.identity' in sys.modules)"
|
|
out = subprocess.run(
|
|
[sys.executable, "-c", code], capture_output=True, text=True, check=True, cwd=DB_DIR.parents[1]
|
|
)
|
|
assert out.stdout.strip() == "False"
|
|
|
|
|
|
def test_metadata_is_separate() -> None:
|
|
assert not set(IdentityBase.metadata.tables) & set(ClinicalBase.metadata.tables)
|
|
assert "audit_log" in ClinicalBase.metadata.tables
|