"""Demo scheduler (#13) on the virtual clock: real dialog engine, real monitoring, sqlite.""" from collections.abc import AsyncIterator from datetime import datetime, timedelta import pytest from sqlalchemy import select from app.config import Settings from app.core import clock from app.core.types import ScrubbedText from app.db import repo, session from app.db.bootstrap import init_db from app.db.clinical import Dose, Prescription from app.dialog import engine as _engine # noqa: F401 (registers flows in order) from app.dialog.flows.veteran_daily import load_daily_texts from app.integrations.base import MemoryOutbox, integrations from app.scheduler import demo, jobs TOKEN = "b" * 32 MEM = "sqlite+aiosqlite:///:memory:" DAY0 = datetime(2026, 10, 5, 8, 0) # visit day (Europe/Moscow) @pytest.fixture async def outbox(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[MemoryOutbox]: monkeypatch.setattr(clock, "get_settings", lambda: Settings(_env_file=None, demo_clock=True)) box = MemoryOutbox() monkeypatch.setattr(integrations, "outbox", box) clock.set_now(DAY0) await session.configure(identity_url=MEM, clinical_url=MEM) await init_db() now = clock.now() async with session.get_clinical_session() as s: await repo.upsert_profile( s, token=TOKEN, doctor_id="doc-smirnova", modules=["veteran"], therapy_start=now.date(), birth_year=1992, sex="m", ) s.add(Prescription(token=TOKEN, drug="сертралин", dose="50 мг", schedule={"times": ["08:00"]})) for scale, total in (("pcl5", 46), ("phq9", 14), ("auditc", 3)): await repo.add_scale_result( s, token=TOKEN, scale=scale, total=total, severity="-", complete=True, ts=now - timedelta(days=3), item9=0 if scale == "phq9" else None, ) await repo.add_message(s, token=TOKEN, direction="in", text=ScrubbedText("..."), ts=now) await repo.set_dialog_state(s, token=TOKEN, state="idle", data={"consent": True, "adult": True}, now=now) yield box clock.reset() await session.configure() async def at(day: int, hh: int, mm: int = 0) -> list[demo.Job]: clock.set_now(DAY0.replace(hour=hh, minute=mm) + timedelta(days=day)) return await demo.run_until(clock.now()) def kinds(delivered: list[demo.Job], *, evaluation: bool = False) -> list[str]: """Delivered job kinds; the daily monitoring pass (from 09:00) is left out unless asked for.""" return [j.kind for j in delivered if evaluation or j.kind != "evaluation"] async def doses() -> list[Dose]: async with session.get_clinical_session() as s: return list((await s.scalars(select(Dose).order_by(Dose.planned_at))).all()) async def dialog() -> tuple[str, dict]: async with session.get_clinical_session() as s: row = await repo.get_dialog_state(s, TOKEN) assert row is not None return row.state, dict(row.data) async def test_reminder_once_then_missed(outbox: MemoryOutbox) -> None: assert kinds(await at(1, 8, 5)) == ["med_reminder"] [dose] = await doses() assert dose.status == "pending" sent = outbox.drain(TOKEN) assert any(b.data == f"med:taken:{dose.id}" for r in sent for row in r.buttons for b in row) assert kinds(await at(1, 9)) == [] # idempotent: the same reminder is not sent again assert kinds(await at(1, 10, 6)) == ["missed"] assert [d.status for d in await doses()] == ["missed"] assert kinds(await at(1, 11)) == [] async def test_answered_dose_is_not_missed(outbox: MemoryOutbox) -> None: from app.dialog.contracts import Event, get_engine await at(1, 8) [dose] = await doses() await get_engine().handle(TOKEN, Event(type="button", button=f"med:taken:{dose.id}")) assert kinds(await at(1, 10, 30)) == [] assert [d.status for d in await doses()] == ["taken"] async def test_three_missed_days_alert_and_reach_out(outbox: MemoryOutbox) -> None: for day in (1, 2, 3): await at(day, 8) await at(day, 10, 1) outbox.drain(TOKEN) async with session.get_clinical_session() as s: codes = [a.code for a in await repo.list_alerts(s, token=TOKEN)] assert "missed_doses_7d" in codes async def test_stale_events_are_not_sent(outbox: MemoryOutbox) -> None: # The clock jumped straight to the evening: no reminder for 08:00, no dose row, only the check-in. delivered = await at(1, 20, 5) assert kinds(delivered) == ["daily_checkin"] assert await doses() == [] assert outbox.drain(TOKEN)[0].text == load_daily_texts().checkin.start assert (await dialog())[0].startswith("daily.") assert kinds(await at(1, 21)) == [] async def test_scales_by_module_schedule(outbox: MemoryOutbox) -> None: assert "scale_due" not in kinds(await at(7, 10, 5)) # screening on day −3: PCL-5/PHQ-9 every 14 days delivered = await at(11, 10, 5) assert [j.params["scale"] for j in delivered if j.kind == "scale_due"] == ["pcl5", "phq9"] state, data = await dialog() assert state == "scale.pcl5" and "phq9" in str(data.get("pending_scales")) async def test_silence_evaluation(outbox: MemoryOutbox) -> None: # Last message: day 0, 08:00. Day 2 — not yet; day 3, 09:05 — silent for more than 3 days. assert kinds(await at(2, 9, 5), evaluation=True) == ["med_reminder", "evaluation"] async with session.get_clinical_session() as s: assert not await repo.list_alerts(s, token=TOKEN) outbox.drain(TOKEN) assert "evaluation" in kinds(await at(3, 9, 5), evaluation=True) async with session.get_clinical_session() as s: assert [a.code for a in await repo.list_alerts(s, token=TOKEN)] == ["no_contact_3d"] # caring contact pushed to the patient (system:reach_out) assert any("как вы" in r.text.lower() for r in outbox.drain(TOKEN)) async def test_no_contact_without_consent_or_in_crisis(outbox: MemoryOutbox) -> None: async with session.get_clinical_session() as s: await repo.set_dialog_state( s, token=TOKEN, state="crisis", data={"consent": True, "adult": True, "crisis": {"open": True}}, now=clock.now(), ) assert await at(1, 8) == [] # crisis: no reminder async with session.get_clinical_session() as s: await repo.set_dialog_state(s, token=TOKEN, state="idle", data={"consent": False}, now=clock.now()) assert await at(1, 8, 30) == [] async def test_jobs_entrypoint_runs_demo_scheduler(outbox: MemoryOutbox) -> None: clock.set_now(DAY0 + timedelta(days=1)) await jobs.run_until_now() assert [d.status for d in await doses()] == ["pending"] assert outbox.drain(TOKEN)