From 4c0836a1bebe4138b79ed881d10ffa52b851662b Mon Sep 17 00:00:00 2001 From: Lacky1234union Date: Sun, 4 Oct 2026 03:13:57 +0300 Subject: [PATCH] =?UTF-8?q?feat(backend):=20monitoring=20service=20?= =?UTF-8?q?=E2=80=94=20snapshot,=20rules,=20alerts,=20status=20colour,=20c?= =?UTF-8?q?lock=20persistence?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - app/services/snapshot.py: PatientSnapshot from the clinical DB (field mapping, age, therapy day, recent C-SSRS risk, weapon access from dialog facts) - app/services/monitoring.py: RuleMonitoring registered via register_monitoring; persists alert/crisis actions, crisis_acute audit, enhanced_monitoring flag, cached status_color, notifies the doctor (and the doctor on duty for crises) after commit; a running crisis is not restarted - demo clock offset persisted in clock_state, restored on scheduler start - tests: Andrey's 13 days through the service → G G G Y Y G G G G Y Y Y R Refs #38 Co-Authored-By: Claude Opus 5.5 --- app/scheduler/jobs.py | 6 + app/services/monitoring.py | 157 +++++++++++++++++ app/services/snapshot.py | 93 ++++++++++ tests/test_contracts.py | 16 +- tests/test_services_monitoring.py | 279 ++++++++++++++++++++++++++++++ 5 files changed, 548 insertions(+), 3 deletions(-) create mode 100644 app/services/monitoring.py create mode 100644 app/services/snapshot.py create mode 100644 tests/test_services_monitoring.py diff --git a/app/scheduler/jobs.py b/app/scheduler/jobs.py index ed8cb21..4bccc9c 100644 --- a/app/scheduler/jobs.py +++ b/app/scheduler/jobs.py @@ -12,6 +12,7 @@ from datetime import datetime from zoneinfo import ZoneInfo from sqlalchemy import select, text +from sqlalchemy.exc import SQLAlchemyError from app.core import clock from app.core.intake_runtime import due_scales, intake_enabled, load_runtime @@ -21,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.services import monitoring from app.services import questionnaire as service log = logging.getLogger(__name__) @@ -163,6 +165,10 @@ async def _run(): async def start(): global _worker + try: + 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(): _worker = asyncio.create_task(_run()) diff --git a/app/services/monitoring.py b/app/services/monitoring.py new file mode 100644 index 0000000..4b0e3d8 --- /dev/null +++ b/app/services/monitoring.py @@ -0,0 +1,157 @@ +"""Monitoring service (#38): patient snapshot → deterministic rules → persisted alerts (ADR-007). + +`after_data_change(token)` is called by the dialog engine after it commits clinical data (and by the scheduler +on silence). It builds a `PatientSnapshot`, runs `app.core.rules.evaluate`, then in one transaction: + * writes every `alert`/`crisis` action to `alerts` (a crisis is an `acute` alert) and, for a crisis, + an audit record `crisis_acute` (guardrail #6); + * applies `enhanced_monitoring` to the profile; + * refreshes the cached `status_color` of the profile (ADR-006; the source of truth stays the open alerts). +After the commit the doctor is notified through `integrations.notifier` (a crisis also goes to the doctor on +duty), and only the *dialog* actions are returned to the engine — it never creates alerts itself. + +A crisis rule whose acute alert is still open stays silent: the crisis script is already running and the +doctor already knows, so it is neither restarted nor re-alerted (`rules.evaluate` never suppresses crises). + +The virtual clock offset is persisted in `clock_state` (ADR-005) by `init_clock_persistence()`. +""" + +import asyncio +import logging +from datetime import datetime, timedelta + +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core import clock +from app.core.rules import Action, evaluate, patient_status +from app.core.types import AlertLevel +from app.db import repo +from app.db.audit import audit +from app.db.session import get_clinical_session +from app.integrations.base import integrations +from app.services.contracts import register_monitoring +from app.services.snapshot import build_snapshot + +log = logging.getLogger(__name__) + +SYSTEM_ACTOR = "system" +# Notification address of the doctor on duty: acute alerts must reach someone even when the attending doctor +# is off (SPEC: crisis → immediate escalation). The notifier maps it to the duty channel. +ON_DUTY_DOCTOR = "on-duty" +# Persisted by the service itself, never passed to the dialog engine. +_SERVICE_ACTIONS = frozenset({"alert", "enhanced_monitoring"}) + + +async def refresh_status(session: AsyncSession, token: str) -> str: + """Recompute and cache the dashboard colour from open alerts (also call it after an ack).""" + alerts = await repo.list_alerts(session, token=token, open_only=True) + status = patient_status(AlertLevel[a.level.upper()] for a in alerts) + await repo.upsert_profile(session, token=token, status_color=status.color) + return status.color + + +def _drop_running_crises(actions: list[Action], open_acute: set[str]) -> list[Action]: + """Silence every action of a crisis rule whose acute alert is still open (see module docstring).""" + running = {a.code for a in actions if a.type == "crisis" and a.code in open_acute} + return [a for a in actions if a.code not in running] + + +class RuleMonitoring: + async def after_data_change(self, token: str) -> list[Action]: + now = clock.now() + notify: list[tuple[str, str, str]] = [] # (doctor_id, level, text) — sent after the commit + async with get_clinical_session() as s: + profile = await repo.get_profile(s, token) + snapshot = await build_snapshot(s, token, now, profile=profile) + if profile is None or snapshot is None: + return [] + open_acute = {a.code for a in await repo.list_alerts(s, token=token, open_only=True) if a.level == "acute"} + actions = _drop_running_crises(evaluate(snapshot), open_acute) + for action in actions: + if action.type == "enhanced_monitoring": + profile.enhanced_monitoring = True + if not action.creates_alert or action.level is None: + continue + alert = await repo.add_alert( + s, + token=token, + level=action.level, + code=action.code, + reason=action.reason_ru, + ts=now, + params=action.params, + ) + level = action.level.name.lower() + notify.append((profile.doctor_id, level, action.reason_ru)) + if action.type == "crisis": + await audit( + s, + actor=SYSTEM_ACTOR, + action="crisis_acute", + token=token, + details={"alert_id": alert.id, "code": action.code}, + ts=now, + ) + notify.append((ON_DUTY_DOCTOR, level, action.reason_ru)) + await refresh_status(s, token) + log.info("monitoring token=%s actions=%s", token, [a.code for a in actions]) + for doctor_id, level, text in notify: + await _notify(doctor_id, level, text, token) + return [a for a in actions if a.type not in _SERVICE_ACTIONS] + + async def is_in_crisis(self, token: str) -> bool: + async with get_clinical_session() as s: + alerts = await repo.list_alerts(s, token=token, open_only=True) + return any(a.level == "acute" for a in alerts) + + +async def _notify(doctor_id: str, level: str, text: str, token: str) -> None: + # The alert is already committed and visible on the dashboard; a failing channel must not undo it. + try: + await integrations.notifier.send(doctor_id, level, text, token) # type: ignore[arg-type] + except Exception as exc: # noqa: BLE001 — any channel error + log.warning("notify failed token=%s level=%s error=%s", token, level, type(exc).__name__) + + +register_monitoring(RuleMonitoring) + + +# ---------------------------------------------------------------------------------- demo clock persistence + +_writes: set[asyncio.Task[None]] = set() + + +async def _write_offset() -> None: + # Writes the offset current at write time, so out-of-order tasks still leave the latest value stored. + try: + async with get_clinical_session() as s: + await repo.save_clock_offset(s, clock.offset().total_seconds()) + except (OSError, SQLAlchemyError) as exc: + log.warning("clock offset not saved error=%s", type(exc).__name__) + + +def _save_offset(_: timedelta) -> None: + """Sync hook for `clock.register_persistence`: schedules the DB write on the running loop.""" + try: + loop = asyncio.get_running_loop() + except RuntimeError: + log.warning("clock offset not saved: no running event loop") + return + task = loop.create_task(_write_offset()) + _writes.add(task) + task.add_done_callback(_writes.discard) + + +async def flush_clock_writes() -> None: + """Wait for pending offset writes (tests, shutdown).""" + if _writes: + await asyncio.gather(*list(_writes)) + + +async def init_clock_persistence() -> datetime: + """Restore the virtual clock offset from `clock_state` and save every later change there.""" + async with get_clinical_session() as s: + stored = timedelta(seconds=await repo.load_clock_offset(s)) + clock.register_persistence(lambda: stored, _save_offset) + log.info("clock offset restored seconds=%s", int(stored.total_seconds())) + return clock.now() diff --git a/app/services/snapshot.py b/app/services/snapshot.py new file mode 100644 index 0000000..17008f8 --- /dev/null +++ b/app/services/snapshot.py @@ -0,0 +1,93 @@ +"""Patient snapshot for the rules engine, assembled from the clinical DB (ADR-007). + +Maps storage names to `app.core.rules` names (`sleep_bucket` → `sleep_hours_bucket`, `alcohol_bucket` → +`alcohol`), derives age from `birth_year` and the therapy day from `therapy_start`. Time comes in as `as_of` +(from `app.core.clock`), never from the wall clock. Must not import `app.db.identity` (guardrail #5). +""" + +from datetime import datetime, timedelta +from typing import Any, get_args + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.rules import ( + CssrsRisk, + DailyCheckin, + DoseRecord, + Module, + PatientSnapshot, + ScalePoint, +) +from app.db import repo +from app.db.clinical import Dose, Message, PatientProfile + +# Risk screens are stored as scale results but reach the rules as `cssrs_risk`, not as scale points. +CSSRS_SCALES = frozenset({"cssrs_screen", "cssrs"}) +# A C-SSRS result counts as the current risk for this long (same window as PHQ-9 item 9 and the dialog's +# C-SSRS cooldown): one old "high" must not restart the crisis protocol after the doctor has acted on it. +CSSRS_RECENT = timedelta(hours=24) +# Unknown sex/age fall back to the more cautious option: lower AUDIT-C cut-off, young-adult monitoring. +_UNKNOWN_SEX = "f" +_UNKNOWN_AGE = 0 + +_MODULES = frozenset(get_args(Module)) +_CSSRS_RISKS = frozenset(get_args(CssrsRisk)) + + +def _cssrs_risk(results: list[Any], as_of: datetime) -> CssrsRisk | None: + recent = [r for r in results if r.scale in CSSRS_SCALES and as_of - CSSRS_RECENT < r.ts <= as_of] + if not recent: + return None + severity = recent[-1].severity + return severity if severity in _CSSRS_RISKS else None + + +async def build_snapshot( + session: AsyncSession, token: str, as_of: datetime, *, profile: PatientProfile | None = None +) -> PatientSnapshot | None: + """Snapshot of everything the rules may look at; None if the patient has no clinical profile.""" + profile = profile or await repo.get_profile(session, token) + if profile is None: + return None + + checkins = [ + DailyCheckin( + day=c.day, + mood=c.mood, + med_taken=c.med_taken, + sleep_hours_bucket=c.sleep_bucket, # type: ignore[arg-type] + alcohol=c.alcohol_bucket, # type: ignore[arg-type] + side_effects=list(c.side_effects or []), + quit_intent=bool(c.quit_intent), + nightmares=c.nightmares, + ) + for c in await repo.list_checkins(session, token) + ] + results = [r for r in await repo.list_scale_results(session, token) if r.complete] + doses = (await session.scalars(select(Dose).where(Dose.token == token).order_by(Dose.planned_at))).all() + last_in = await session.scalar( + select(func.max(Message.ts)).where(Message.token == token, Message.direction == "in") + ) + state = await repo.get_dialog_state(session, token) + facts = (state.data.get("facts") or {}) if state is not None else {} + + return PatientSnapshot( + token=token, + as_of=as_of, + age=as_of.year - profile.birth_year if profile.birth_year else _UNKNOWN_AGE, + sex=profile.sex if profile.sex in ("m", "f") else _UNKNOWN_SEX, # type: ignore[arg-type] + modules={m for m in profile.modules or [] if m in _MODULES}, # type: ignore[misc] + therapy_day=(as_of.date() - profile.therapy_start).days if profile.therapy_start else 0, + checkins=checkins, + scale_results=[ + ScalePoint(scale=r.scale, ts=r.ts, total=r.total, item9=r.item9, scheduled=r.scheduled) + for r in results + if r.scale not in CSSRS_SCALES + ], + doses=[DoseRecord(planned_at=d.planned_at, status=d.status) for d in doses], # type: ignore[arg-type] + cssrs_risk=_cssrs_risk(results, as_of), + weapon_access=facts.get("weapon_access"), + last_patient_message_at=last_in, + open_alert_codes=await repo.open_alert_codes(session, token), + ) diff --git a/tests/test_contracts.py b/tests/test_contracts.py index 4063380..cae58c3 100644 --- a/tests/test_contracts.py +++ b/tests/test_contracts.py @@ -1,14 +1,24 @@ from fastapi.testclient import TestClient +from app.db import session +from app.db.bootstrap import init_db from app.dialog.contracts import Event, get_engine from app.main import app from app.services.contracts import get_monitoring +MEM = "sqlite+aiosqlite:///:memory:" + async def test_stubs_are_usable() -> None: - replies = await get_engine().handle("tok", Event(type="text", text="привет")) - assert replies and replies[0].text - assert await get_monitoring().after_data_change("tok") == [] + await session.configure(identity_url=MEM, clinical_url=MEM) + await init_db() + try: + replies = await get_engine().handle("tok", Event(type="text", text="привет")) + assert replies and replies[0].text + assert await get_monitoring().after_data_change("tok") == [] # unknown patient → no actions + assert await get_monitoring().is_in_crisis("tok") is False + finally: + await session.configure() def test_api_fixture_routes() -> None: diff --git a/tests/test_services_monitoring.py b/tests/test_services_monitoring.py new file mode 100644 index 0000000..31627d6 --- /dev/null +++ b/tests/test_services_monitoring.py @@ -0,0 +1,279 @@ +"""MonitoringService (#38): snapshot from the clinical DB → real rules → alerts, colour cache, audit, notify.""" + +from collections.abc import AsyncIterator +from datetime import datetime, timedelta +from typing import Any + +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.audit import AuditLog +from app.db.bootstrap import init_db +from app.db.clinical import Dose, Prescription +from app.integrations.base import MemoryNotifier, integrations +from app.services import contracts, monitoring +from app.services.snapshot import build_snapshot + +TOKEN = "a" * 32 +DOCTOR = "doc-smirnova" +MEM = "sqlite+aiosqlite:///:memory:" +DAY0 = datetime(2026, 10, 5, 8, 0) + +# Persona (docs/product/01_demo_persona.md): day → (dose taken, mood, sleep, nightmares, alcohol, side effects, +# quit intent). Same table as tests/test_dialog_veteran_days.py, written straight to the DB. +TABLE: dict[int, tuple[bool, int, str, bool, str, list[str], bool]] = { + 1: (True, 4, "4-6", True, "none", [], False), + 2: (True, 4, "4-6", True, "none", ["nausea"], False), + 3: (True, 4, "4-6", True, "none", ["nausea"], False), + 4: (True, 3, "4-6", False, "none", ["nausea"], False), + 5: (False, 3, "4-6", False, "none", [], True), + 6: (True, 4, "6-8", False, "none", [], False), + 7: (True, 4, "6-8", False, "none", [], False), + 8: (True, 4, "<4", True, "1-2", [], False), + 9: (True, 4, "<4", True, "more", [], False), + 10: (True, 4, "<4", True, "more", [], False), + 11: (True, 3, "4-6", False, "1-2", [], False), + 12: (True, 3, "4-6", False, "none", [], False), + 13: (True, 2, "<4", True, "none", [], False), +} +COLORS = ["green"] * 3 + ["yellow"] * 2 + ["green"] * 4 + ["yellow"] * 3 + ["red"] + + +class FailingNotifier: + async def send(self, *args: Any, **kwargs: Any) -> None: + raise ConnectionError("channel down") + + +@pytest.fixture +async def db(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[MemoryNotifier]: + monkeypatch.setattr(clock, "get_settings", lambda: Settings(_env_file=None, demo_clock=True)) + notifier = MemoryNotifier() + monkeypatch.setattr(integrations, "notifier", notifier) + clock.set_now(DAY0) + await session.configure(identity_url=MEM, clinical_url=MEM) + await init_db() + async with session.get_clinical_session() as s: + await repo.upsert_profile( + s, + token=TOKEN, + doctor_id=DOCTOR, + modules=["veteran", "depression"], + therapy_start=clock.today(), + 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=clock.now() - timedelta(days=3), + item9=0 if scale == "phq9" else None, + ) + await repo.set_dialog_state(s, token=TOKEN, state="idle", data={}, now=clock.now()) + yield notifier + clock.register_persistence(None, None) + clock.reset() + await session.configure() + + +async def scale( + name: str, total: int, *, item9: int | None = None, severity: str = "-", scheduled: bool = True +) -> None: + async with session.get_clinical_session() as s: + await repo.add_scale_result( + s, + token=TOKEN, + scale=name, + total=total, + severity=severity, + complete=True, + ts=clock.now(), + item9=item9, + scheduled=scheduled, + ) + + +async def color() -> str | None: + async with session.get_clinical_session() as s: + profile = await repo.get_profile(s, TOKEN) + assert profile is not None + return profile.status_color + + +async def ack_all() -> None: + async with session.get_clinical_session() as s: + for a in await repo.list_alerts(s, token=TOKEN, open_only=True): + await repo.ack_alert(s, alert_id=a.id, doctor_id=DOCTOR, ts=clock.now()) + await monitoring.refresh_status(s, TOKEN) + + +async def live_day(day: int) -> None: + """What the dialog writes on one persona day: the dose, an incoming message and the evening check-in.""" + taken, mood, sleep, nightmares, alcohol, effects, quit_intent = TABLE[day] + clock.set_now(DAY0 + timedelta(days=day)) + async with session.get_clinical_session() as s: + rx = await s.scalar(select(Prescription)) + assert rx is not None + s.add(Dose(token=TOKEN, prescription_id=rx.id, planned_at=clock.now(), status="taken" if taken else "skipped")) + await repo.add_message(s, token=TOKEN, direction="in", text=ScrubbedText("..."), ts=clock.now()) + await repo.upsert_checkin( + s, + token=TOKEN, + day=clock.today(), + now=clock.now(), + mood=mood, + med_taken=taken, + sleep_bucket=sleep, + alcohol_bucket=alcohol, + nightmares=nightmares, + side_effects=effects, + quit_intent=quit_intent, + ) + + +async def test_service_is_registered() -> None: + assert isinstance(contracts.get_monitoring(), monitoring.RuleMonitoring) + + +async def test_andrey_13_days(db: MemoryNotifier) -> None: + mon = monitoring.RuleMonitoring() + colors: list[str | None] = [] + returned: dict[int, list[str]] = {} + for day in range(1, 14): + if day == 6: + await ack_all() # the doctor called Andrey after the day-5 alerts + await live_day(day) + if day == 7: + await scale("pcl5", 44) + await scale("phq9", 13, item9=0) + actions = await mon.after_data_change(TOKEN) + if day == 13: + assert [a.code for a in actions] == ["mood_low_3d"] # dialog action: unscheduled PHQ-9 + await scale("cssrs_screen", 0, severity="none", scheduled=False) + await scale("phq9", 18, item9=2, scheduled=False) + actions += await mon.after_data_change(TOKEN) + returned[day] = [a.type for a in actions] + colors.append(await color()) + + assert colors == COLORS + assert "med_info" in returned[5] and "alert" not in returned[5] # alerts are persisted, not returned + assert {"med_info", "suggest_visit"} <= set(returned[10]) + assert {"crisis", "show_safety_plan", "schedule_scale"} <= set(returned[13]) + async with session.get_clinical_session() as s: + alerts = list(await repo.list_alerts(s, token=TOKEN)) + crisis_audit = (await s.scalars(select(AuditLog).where(AuditLog.action == "crisis_acute"))).all() + codes = [a.code for a in reversed(alerts)] + assert codes == [ + "side_effect_3d", + "quit_intent", + "sleep_lt4_3n", + "alcohol_3d", + "phq9_item9", # same evaluation: YAML order + "phq9_rise", + ] + assert len(crisis_audit) == 1 and crisis_audit[0].token == TOKEN and crisis_audit[0].actor == "system" + acute = [(d, lvl) for d, lvl, _text, _tok in db.sent if lvl == "acute"] + assert acute == [(DOCTOR, "acute"), (monitoring.ON_DUTY_DOCTOR, "acute")] + assert all(tok == TOKEN for *_rest, tok in db.sent) + assert await mon.is_in_crisis(TOKEN) + + +async def test_running_crisis_is_not_restarted(db: MemoryNotifier) -> None: + mon = monitoring.RuleMonitoring() + await scale("phq9", 18, item9=1) + first = await mon.after_data_change(TOKEN) + assert any(a.type == "crisis" for a in first) + clock.advance(hours=1) + again = await mon.after_data_change(TOKEN) + assert not any(a.code == "phq9_item9" for a in again) + async with session.get_clinical_session() as s: + assert [a.code for a in await repo.list_alerts(s, token=TOKEN, open_only=True) if a.level == "acute"] == [ + "phq9_item9" + ] + await ack_all() + assert not await mon.is_in_crisis(TOKEN) + assert await color() == "green" + + +async def test_old_cssrs_high_does_not_restart_crisis(db: MemoryNotifier) -> None: + mon = monitoring.RuleMonitoring() + await scale("cssrs_screen", 4, severity="high", scheduled=False) + assert any(a.type == "crisis" for a in await mon.after_data_change(TOKEN)) + await ack_all() + clock.advance(days=2) + assert not any(a.type == "crisis" for a in await mon.after_data_change(TOKEN)) + + +async def test_weapon_access_from_dialog_facts(db: MemoryNotifier) -> None: + async with session.get_clinical_session() as s: + await repo.set_dialog_state( + s, token=TOKEN, state="idle", data={"facts": {"weapon_access": True}}, now=clock.now() + ) + snap = await build_snapshot(s, TOKEN, clock.now()) + assert snap is not None and snap.weapon_access is True + await scale("cssrs_screen", 3, severity="moderate", scheduled=False) + actions = await monitoring.RuleMonitoring().after_data_change(TOKEN) + assert {a.code for a in actions if a.type == "crisis"} == {"weapon_access_risk"} + plan = next(a for a in actions if a.type == "show_safety_plan") + assert plan.params["block"] == "means_restriction" + assert await color() == "red" + + +async def test_snapshot_mapping_and_unknown_patient(db: MemoryNotifier) -> None: + clock.set_now(DAY0 + timedelta(days=4)) + await live_day(4) + async with session.get_clinical_session() as s: + snap = await build_snapshot(s, TOKEN, clock.now()) + assert await build_snapshot(s, "f" * 32, clock.now()) is None + assert snap is not None + assert (snap.age, snap.sex, snap.therapy_day, snap.modules) == (34, "m", 4, {"veteran", "depression"}) + c = snap.checkins[-1] + assert (c.sleep_hours_bucket, c.alcohol, c.side_effects) == ("4-6", "none", ["nausea"]) + assert [p.scale for p in snap.scale_results] == ["pcl5", "phq9", "auditc"] + assert snap.last_patient_message_at is not None and snap.last_patient_message_at.date() == clock.today() + assert await monitoring.RuleMonitoring().after_data_change("f" * 32) == [] + + +async def test_young_adult_enhanced_monitoring(db: MemoryNotifier) -> None: + async with session.get_clinical_session() as s: + await repo.upsert_profile(s, token=TOKEN, birth_year=2004) + actions = await monitoring.RuleMonitoring().after_data_change(TOKEN) + assert not any(a.type == "enhanced_monitoring" for a in actions) # applied by the service itself + async with session.get_clinical_session() as s: + profile = await repo.get_profile(s, TOKEN) + assert profile is not None and profile.enhanced_monitoring is True + assert not await repo.list_alerts(s, token=TOKEN) + assert await color() == "green" + + +async def test_notifier_failure_keeps_alert(db: MemoryNotifier, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(integrations, "notifier", FailingNotifier()) + await scale("pcl5", 55) + await monitoring.RuleMonitoring().after_data_change(TOKEN) + async with session.get_clinical_session() as s: + assert [(a.code, a.level) for a in await repo.list_alerts(s, token=TOKEN)] == [("pcl5_severe", "high")] + assert await color() == "yellow" + + +async def test_clock_offset_persists(db: MemoryNotifier) -> None: + clock.reset() + await monitoring.init_clock_persistence() + clock.advance(days=3) + await monitoring.flush_clock_writes() + async with session.get_clinical_session() as s: + assert await repo.load_clock_offset(s) == pytest.approx(timedelta(days=3).total_seconds()) + + # "Restart": in-memory offset lost, restored from clock_state. + clock.register_persistence(None, None) + clock.reset() + await monitoring.init_clock_persistence() + assert clock.offset() == timedelta(days=3) -- 2.54.0