From 251f9466910fb24082dafc1cd417fe8950dd4268 Mon Sep 17 00:00:00 2001 From: Lacky1234union Date: Sun, 4 Oct 2026 03:25:44 +0300 Subject: [PATCH] =?UTF-8?q?feat(backend):=20demo=20scheduler=20on=20the=20?= =?UTF-8?q?virtual=20clock=20=E2=80=94=20doses,=20missed,=20check-ins,=20s?= =?UTF-8?q?cales,=20silence?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - app/scheduler/demo.py: run_until(now) replays due and fresh events in time order: dose rows + med_reminder, missed after 2 h without an answer → monitoring, daily_checkin, scale_due by the config/modules.yaml schedules, daily monitoring pass (silence, missed doses) → reach_out/scale_due - idempotent via scheduler_events (unique token+key); stale events are not sent; no contact without consent or during an open crisis - config/scheduler.yaml: times and windows; runs while the questionnaire release is off - POST /demo/clock/advance replays the scheduler up to the new time Refs #13 Co-Authored-By: Claude Opus 5.5 --- app/api/routes.py | 4 +- app/db/bootstrap.py | 1 + app/db/scheduler.py | 24 ++++ app/scheduler/demo.py | 257 +++++++++++++++++++++++++++++++++++ app/scheduler/jobs.py | 10 +- config/scheduler.yaml | 25 ++++ tests/test_db.py | 1 + tests/test_scheduler_demo.py | 167 +++++++++++++++++++++++ 8 files changed, 485 insertions(+), 4 deletions(-) create mode 100644 app/db/scheduler.py create mode 100644 app/scheduler/demo.py create mode 100644 config/scheduler.yaml create mode 100644 tests/test_scheduler_demo.py diff --git a/app/api/routes.py b/app/api/routes.py index 5743a9e..3c6d4e2 100644 --- a/app/api/routes.py +++ b/app/api/routes.py @@ -25,6 +25,7 @@ from app.api.schemas import ( from app.core import clock from app.dialog.contracts import get_engine from app.integrations.base import MemoryOutbox, integrations +from app.scheduler import jobs as scheduler router = APIRouter(prefix="/api/v1") router.include_router(questionnaire_router) @@ -99,8 +100,9 @@ async def get_clock() -> ClockOut: @router.post("/demo/clock/advance", response_model=ClockOut, tags=["demo"]) async def advance_clock(body: ClockAdvanceIn) -> ClockOut: - """Moves the virtual clock; #13 makes the scheduler replay all events up to the new time.""" + """Moves the virtual clock and replays the scheduler up to the new time (#13).""" clock.advance(days=body.days, hours=body.hours) + await scheduler.run_until_now() return _clock_out() diff --git a/app/db/bootstrap.py b/app/db/bootstrap.py index 9a76f84..f4f163d 100644 --- a/app/db/bootstrap.py +++ b/app/db/bootstrap.py @@ -5,6 +5,7 @@ The only module (besides registration code) that touches both schemas — and on from app.db import audit as _audit # noqa: F401 (registers audit_log in clinical metadata) from app.db import intake as _intake # noqa: F401 +from app.db import scheduler as _scheduler # noqa: F401 from app.db.clinical import ClinicalBase from app.db.identity import IdentityBase from app.db.session import clinical_engine, identity_engine diff --git a/app/db/scheduler.py b/app/db/scheduler.py new file mode 100644 index 0000000..26d70ef --- /dev/null +++ b/app/db/scheduler.py @@ -0,0 +1,24 @@ +"""Delivery log of the demo scheduler (#13): one row per delivered event, the unique key makes runs idempotent. + +Clinical DB, keyed by token only (guardrail #5). Created by `init_db` (`app.db.bootstrap`). +""" + +from datetime import datetime + +from sqlalchemy import String, UniqueConstraint +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.clinical import TOKEN_LEN, BigId, ClinicalBase +from app.db.types import TZDateTime + + +class SchedulerEvent(ClinicalBase): + __tablename__ = "scheduler_events" + __table_args__ = (UniqueConstraint("token", "key", name="uq_scheduler_events_token_key"),) + + id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) + token: Mapped[str] = mapped_column(String(TOKEN_LEN)) + key: Mapped[str] = mapped_column(String(96)) # e.g. "dose:3:2026-10-06T08:00", "checkin:2026-10-06" + kind: Mapped[str] = mapped_column(String(32)) # med_reminder | missed | daily_checkin | scale_due | evaluation + due_at: Mapped[datetime] = mapped_column(TZDateTime) + done_at: Mapped[datetime] = mapped_column(TZDateTime) diff --git a/app/scheduler/demo.py b/app/scheduler/demo.py new file mode 100644 index 0000000..d36f54a --- /dev/null +++ b/app/scheduler/demo.py @@ -0,0 +1,257 @@ +"""Demo scheduler on the virtual clock (#13, SPEC §8, ADR-005). + +`run_until(now)` replays, in time order, every event that is due by `now` and still fresh +(`config/scheduler.yaml`): + + * med_reminder — a `doses` row (pending) per prescription time, then `system:med_reminder {dose_id, drug, dose}`; + * missed — a reminded dose still pending `missed_after_hours` later → status `missed` → monitoring; + * daily_checkin — `system:daily_checkin` at the check-in time; + * scale_due — scales due by the module schedules (`config/modules.yaml`) → `system:scale_due`; + * evaluation — once a day `MonitoringService.after_data_change()`: catches silence > N days (rule + `no_contact_3d`) and missed doses even when the patient writes nothing. + +The rules decide (ADR-007); the scheduler only turns their dialog actions into system events for the engine +(`reach_out` → `system:reach_out`, `schedule_scale` → `system:scale_due`). Replies go to `integrations.outbox`. + +Idempotent: a delivered event is stored in `scheduler_events` under a unique key and never sent again; a failed +delivery is not stored and is retried on the next run while it is fresh (at least once). Only patients with +consent (and adult) are contacted; while a crisis is open no reminders, check-ins or scales are sent. +Time comes only from the arguments / `app.core.clock` (guardrail #4). +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from datetime import date, datetime, time, timedelta +from functools import lru_cache +from pathlib import Path +from typing import Any + +import yaml +from pydantic import BaseModel, ConfigDict, Field +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import get_settings +from app.core.modules import due_scales, load_modules +from app.core.rules import Action +from app.db import repo +from app.db.clinical import DialogState, Dose, PatientProfile, Prescription +from app.db.scheduler import SchedulerEvent +from app.db.session import get_clinical_session +from app.dialog.contracts import Event, Reply, get_engine +from app.integrations.base import integrations +from app.services.contracts import get_monitoring + +log = logging.getLogger(__name__) + +# ------------------------------------------------------------------------------------------------ config + + +class _Strict(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + +class DosesConfig(_Strict): + late_hours: float = Field(gt=0) + missed_after_hours: float = Field(gt=0) + + +class Window(_Strict): + time: time + late_hours: float = Field(gt=0) + + +class SchedulerConfig(_Strict): + enabled: bool = True + tick_seconds: float = Field(gt=0) + doses: DosesConfig + daily_checkin: Window + scales: Window + evaluation: Window + + +@lru_cache +def load_config(path: Path | None = None) -> SchedulerConfig: + path = path or get_settings().config_dir / "scheduler.yaml" + return SchedulerConfig.model_validate(yaml.safe_load(path.read_text(encoding="utf-8"))) + + +# ------------------------------------------------------------------------------------------------ jobs + +# Same due time → this order: the dose before the check-in that asks about it. +_ORDER = {"evaluation": 0, "missed": 1, "med_reminder": 2, "scale_due": 3, "daily_checkin": 4} + + +@dataclass(order=True) +class Job: + due: datetime + rank: int + token: str + key: str + kind: str = field(compare=False) + params: dict[str, Any] = field(default_factory=dict, compare=False) + + +def _job(due: datetime, token: str, key: str, kind: str, **params: Any) -> Job: + return Job(due=due, rank=_ORDER[kind], token=token, key=key, kind=kind, params=params) + + +def _at(day: date, t: time, now: datetime) -> datetime: + return datetime.combine(day, t, tzinfo=now.tzinfo) + + +def _fresh(due: datetime, now: datetime, late_hours: float) -> bool: + return due <= now < due + timedelta(hours=late_hours) + + +def _window(day: date, w: Window, now: datetime) -> datetime | None: + due = _at(day, w.time, now) + return due if _fresh(due, now, w.late_hours) else None + + +async def _dose_jobs(s: AsyncSession, p: PatientProfile, now: datetime, cfg: SchedulerConfig) -> list[Job]: + jobs: list[Job] = [] + today = now.date() + for rx in (await s.scalars(select(Prescription).where(Prescription.token == p.token))).all(): + start = rx.start_date or p.therapy_start + for day in (today - timedelta(days=1), today): # a fresh dose is never older than a day + if (start and day < start) or (rx.end_date and day > rx.end_date): + continue + for raw in (rx.schedule or {}).get("times", []): + planned = _at(day, time.fromisoformat(raw), now) + if _fresh(planned, now, cfg.doses.late_hours): + key = f"dose:{rx.id}:{planned.isoformat(timespec='minutes')}" + jobs.append(_job(planned, p.token, key, "med_reminder", rx_id=rx.id, planned_at=planned)) + return jobs + + +async def _missed_jobs(s: AsyncSession, token: str, now: datetime, cfg: SchedulerConfig) -> list[Job]: + after = timedelta(hours=cfg.doses.missed_after_hours) + stmt = select(Dose).where(Dose.token == token, Dose.status == "pending", Dose.planned_at <= now - after) + return [_job(d.planned_at + after, token, f"missed:{d.id}", "missed", dose_id=d.id) for d in await s.scalars(stmt)] + + +async def _scale_jobs(s: AsyncSession, p: PatientProfile, now: datetime, cfg: SchedulerConfig) -> list[Job]: + due_at = _window(now.date(), cfg.scales, now) + if due_at is None: + return [] + last: dict[str, date] = {} + for r in await repo.list_scale_results(s, p.token): + if r.complete: + last[r.scale] = r.ts.astimezone(now.tzinfo).date() + if p.therapy_start: + # A scale never taken counts from the visit where therapy started, not as overdue on day 0. + for mod in load_modules().modules.values(): + for entry in mod.schedule: + last.setdefault(entry.scale, p.therapy_start) + therapy_day = (now.date() - p.therapy_start).days if p.therapy_start else 0 + scales = due_scales(set(p.modules or []), therapy_day, last, now.date()) + return [_job(due_at, p.token, f"scale:{sc}:{now.date()}", "scale_due", scale=sc) for sc in scales] + + +async def collect(now: datetime, cfg: SchedulerConfig | None = None) -> list[Job]: + """Due and fresh, not yet delivered events of all contactable patients, in time order.""" + cfg = cfg or load_config() + jobs: list[Job] = [] + async with get_clinical_session() as s: + for p in (await s.scalars(select(PatientProfile).order_by(PatientProfile.token))).all(): + state = await s.get(DialogState, p.token) + data = state.data if state is not None else {} + if data.get("consent") is not True or data.get("adult") is not True: + continue + mine = await _missed_jobs(s, p.token, now, cfg) + if (due := _window(now.date(), cfg.evaluation, now)) is not None: + mine.append(_job(due, p.token, f"eval:{now.date()}", "evaluation")) + if not (data.get("crisis") or {}).get("open"): + mine += await _dose_jobs(s, p, now, cfg) + mine += await _scale_jobs(s, p, now, cfg) + if (due := _window(now.date(), cfg.daily_checkin, now)) is not None: + mine.append(_job(due, p.token, f"checkin:{now.date()}", "daily_checkin")) + done = set((await s.scalars(select(SchedulerEvent.key).where(SchedulerEvent.token == p.token))).all()) + jobs += [j for j in mine if j.key not in done] + return sorted(jobs) + + +# ------------------------------------------------------------------------------------------------ delivery + + +async def _system(token: str, kind: str, **params: Any) -> None: + replies: list[Reply] = await get_engine().handle( + token, + Event(type="system", system=kind, params=params), # type: ignore[arg-type] + ) + if replies: + await integrations.outbox.push(token, replies) + + +async def _after_rules(token: str) -> None: + """Monitoring pass outside a dialog: hand the rules' dialog actions to the engine as system events.""" + actions: list[Action] = await get_monitoring().after_data_change(token) + for a in actions: + if a.type == "reach_out": + await _system(token, "reach_out", **a.params) + elif a.type == "schedule_scale" and a.params.get("scale"): + await _system(token, "scale_due", scale=a.params["scale"], scheduled=bool(a.params.get("scheduled"))) + elif a.type == "crisis": + log.warning("scheduler crisis action token=%s code=%s", token, a.code) # alert already persisted + + +async def _ensure_dose(token: str, rx_id: int, planned_at: datetime) -> tuple[int, Prescription]: + async with get_clinical_session() as s: + rx = await s.get(Prescription, rx_id) + assert rx is not None + dose = await s.scalar( + select(Dose).where(Dose.token == token, Dose.prescription_id == rx_id, Dose.planned_at == planned_at) + ) + if dose is None: + dose = Dose(token=token, prescription_id=rx_id, planned_at=planned_at, status="pending") + s.add(dose) + await s.flush() + return dose.id, rx + + +async def _mark_missed(dose_id: int) -> bool: + async with get_clinical_session() as s: + dose = await s.get(Dose, dose_id) + if dose is None or dose.status != "pending": + return False # answered meanwhile + dose.status = "missed" + return True + + +async def _run(job: Job, now: datetime) -> None: + if job.kind == "med_reminder": + dose_id, rx = await _ensure_dose(job.token, job.params["rx_id"], job.params["planned_at"]) + await _system(job.token, "med_reminder", dose_id=dose_id, drug=rx.drug, dose=rx.dose) + elif job.kind == "missed": + if await _mark_missed(job.params["dose_id"]): + await _after_rules(job.token) + elif job.kind == "daily_checkin": + await _system(job.token, "daily_checkin") + elif job.kind == "scale_due": + await _system(job.token, "scale_due", scale=job.params["scale"], scheduled=True) + elif job.kind == "evaluation": + await _after_rules(job.token) + + +async def _done(job: Job, now: datetime) -> None: + async with get_clinical_session() as s: + s.add(SchedulerEvent(token=job.token, key=job.key, kind=job.kind, due_at=job.due, done_at=now)) + + +async def run_until(now: datetime, cfg: SchedulerConfig | None = None) -> list[Job]: + """Deliver every due event up to `now` in time order; returns the delivered jobs.""" + delivered: list[Job] = [] + for job in await collect(now, cfg): + try: + await _run(job, now) + except Exception as exc: # noqa: BLE001 — one patient's failure must not stop the others + log.error("scheduler job failed token=%s kind=%s error=%s", job.token, job.kind, type(exc).__name__) + continue + await _done(job, now) + delivered.append(job) + if delivered: + log.info("scheduler delivered=%s", [(j.token, j.kind) for j in delivered]) + return delivered diff --git a/app/scheduler/jobs.py b/app/scheduler/jobs.py index 4bccc9c..9a72211 100644 --- a/app/scheduler/jobs.py +++ b/app/scheduler/jobs.py @@ -1,4 +1,4 @@ -"""Questionnaire jobs on the demo clock; clinical release disabled by default. +"""Scheduler on the demo clock: questionnaire jobs when that release is on, else the demo scheduler (#13). One scheduler per database. PostgreSQL's advisory transaction lock serializes workers; an asyncio lock does the same within a SQLite demonstration process. @@ -22,6 +22,7 @@ from app.db.intake import QuestionnaireTask from app.db.session import get_clinical_session from app.dialog.contracts import Event, Reply, get_engine from app.integrations.base import integrations +from app.scheduler import demo from app.services import monitoring from app.services import questionnaire as service @@ -148,6 +149,9 @@ async def run_until_now(): return await service.dispatch_notices() if not intake_enabled(): + # Questionnaire release off → the demo scheduler (doses, check-ins, scales, silence) runs instead. + if demo.load_config().enabled: + await demo.run_until(clock.now()) return await plan() await deliver() @@ -160,7 +164,7 @@ async def _run(): await run_until_now() except Exception as exc: log.error("questionnaire scheduler error=%s", type(exc).__name__) - await asyncio.sleep(30) + await asyncio.sleep(demo.load_config().tick_seconds) async def start(): @@ -169,7 +173,7 @@ async def start(): await monitoring.init_clock_persistence() except (OSError, SQLAlchemyError) as exc: log.error("clock persistence unavailable error=%s", type(exc).__name__) - if _worker is None and intake_enabled(): + if _worker is None and (intake_enabled() or demo.load_config().enabled): _worker = asyncio.create_task(_run()) diff --git a/config/scheduler.yaml b/config/scheduler.yaml new file mode 100644 index 0000000..407b15f --- /dev/null +++ b/config/scheduler.yaml @@ -0,0 +1,25 @@ +# Demo scheduler on the virtual clock (#13, SPEC §8, ADR-005). Local times are in settings.tz. +# Read by app/scheduler/demo.py; used only while the questionnaire release (intake_runtime.yaml) is off. +# +# Every event is delivered once (idempotency key in `scheduler_events`) and only while it is fresh: an event +# older than its late window is not sent at all — after a long jump of the demo clock the patient does not +# receive a pile of yesterday's reminders, and an undelivered reminder never turns into a "missed" dose. + +enabled: true +tick_seconds: 30 + +doses: + late_hours: 2 # reminder not delivered if the dose time is older than this + missed_after_hours: 2 # reminded dose without an answer → status "missed" (SPEC §8) + +daily_checkin: + time: "20:00" + late_hours: 4 + +scales: + time: "10:00" # scales due today (config/modules.yaml schedules) are offered from this hour + late_hours: 10 + +evaluation: + time: "09:00" # daily monitoring pass: silence > N days, missed doses → caring contact + late_hours: 15 diff --git a/tests/test_db.py b/tests/test_db.py index c8fd10f..15d5516 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -70,6 +70,7 @@ async def test_create_all_puts_tables_in_the_right_db(dbs: None) -> None: "questionnaire_records", "questionnaire_notices", "questionnaire_tasks", + "scheduler_events", } == clinical_tables await init_db() # idempotent diff --git a/tests/test_scheduler_demo.py b/tests/test_scheduler_demo.py new file mode 100644 index 0000000..73411e4 --- /dev/null +++ b/tests/test_scheduler_demo.py @@ -0,0 +1,167 @@ +"""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) -- 2.54.0