ElebobaandClaude Opus 5.5
711c646876
CI / ruff · pytest (pull_request) Successful in 18s
- app/core/clock.py: now()/today(), virtual offset (advance/set_now/reset), pluggable offset persistence hook (ADR-005) - config/rules.yaml + app/core/rules.py: YAML declares rules (params, actions, levels, reason_ru), Python evaluators implement predicates by id; dedup of open alerts, crisis never suppressed; ADR-006 colours (ADR-007) - tests: per-rule unit tests, dedup, clock, 13-day demo persona trajectory Refs #5 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
428 lines
16 KiB
Python
428 lines
16 KiB
Python
from datetime import date, datetime, time, timedelta, timezone
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from pydantic import ValidationError
|
||
|
||
from app.core.rules import (
|
||
Action,
|
||
DailyCheckin,
|
||
DoseRecord,
|
||
PatientSnapshot,
|
||
RulesConfig,
|
||
ScalePoint,
|
||
color,
|
||
evaluate,
|
||
load_rules,
|
||
patient_status,
|
||
registered_rule_ids,
|
||
)
|
||
from app.core.types import AlertLevel
|
||
|
||
MSK = timezone(timedelta(hours=3), "MSK")
|
||
DAY0 = date(2026, 9, 21) # visit to the psychiatrist, sertraline prescribed
|
||
|
||
|
||
def at(day: int, hour: int = 21) -> datetime:
|
||
return datetime.combine(DAY0 + timedelta(days=day), time(hour), MSK)
|
||
|
||
|
||
def snap(day: int, **kw: Any) -> PatientSnapshot:
|
||
base: dict[str, Any] = {
|
||
"token": "t" * 32,
|
||
"as_of": at(day),
|
||
"age": 28,
|
||
"sex": "m",
|
||
"modules": {"depression"},
|
||
"therapy_day": day,
|
||
"last_patient_message_at": at(day, 20),
|
||
}
|
||
base.update(kw)
|
||
return PatientSnapshot(**base)
|
||
|
||
|
||
def ci(day: int, **kw: Any) -> DailyCheckin:
|
||
return DailyCheckin(day=DAY0 + timedelta(days=day), **kw)
|
||
|
||
|
||
def codes(actions: list[Action], type_: str | None = None) -> set[str]:
|
||
return {a.code for a in actions if type_ is None or a.type == type_}
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------- config
|
||
|
||
|
||
def test_yaml_loads_and_every_rule_has_evaluator() -> None:
|
||
cfg = load_rules()
|
||
ids = {r.id for r in cfg.rules}
|
||
assert ids == registered_rule_ids()
|
||
|
||
|
||
def test_unknown_rule_id_rejected() -> None:
|
||
with pytest.raises(ValidationError):
|
||
RulesConfig.model_validate(
|
||
{"rules": [{"id": "nope", "actions": [{"type": "alert", "level": "info"}]}]}
|
||
)
|
||
|
||
|
||
def test_alert_without_level_rejected() -> None:
|
||
with pytest.raises(ValidationError):
|
||
RulesConfig.model_validate({"rules": [{"id": "sleep_lt4_3n", "actions": [{"type": "alert"}]}]})
|
||
|
||
|
||
def test_disabled_rule_is_skipped() -> None:
|
||
cfg = load_rules().model_copy(deep=True)
|
||
for r in cfg.rules:
|
||
r.enabled = r.id != "sleep_lt4_3n"
|
||
s = snap(3, checkins=[ci(d, sleep_hours_bucket="<4") for d in (1, 2, 3)])
|
||
assert "sleep_lt4_3n" not in codes(evaluate(s, cfg))
|
||
|
||
|
||
def test_naive_as_of_rejected() -> None:
|
||
with pytest.raises(ValidationError):
|
||
snap(1, as_of=datetime(2026, 9, 22, 21))
|
||
|
||
|
||
def test_quiet_patient_has_no_actions() -> None:
|
||
assert evaluate(snap(3, checkins=[ci(d, mood=6, med_taken=True) for d in (1, 2, 3)])) == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------- crisis rules
|
||
|
||
|
||
def test_phq9_item9_crisis() -> None:
|
||
s = snap(13, scale_results=[ScalePoint(scale="phq9", ts=at(13, 20), total=12, item9=1, scheduled=False)])
|
||
acts = evaluate(s)
|
||
crisis = [a for a in acts if a.code == "phq9_item9"]
|
||
assert [a.type for a in crisis] == ["crisis", "show_safety_plan", "schedule_scale"]
|
||
assert crisis[0].level == AlertLevel.ACUTE
|
||
assert crisis[2].params["scale"] == "cssrs_screen"
|
||
assert "1" in crisis[0].reason_ru
|
||
|
||
|
||
def test_phq9_item9_zero_or_old_is_not_crisis() -> None:
|
||
zero = snap(7, scale_results=[ScalePoint(scale="phq9", ts=at(7, 20), total=13, item9=0)])
|
||
assert "phq9_item9" not in codes(evaluate(zero))
|
||
old = snap(9, scale_results=[ScalePoint(scale="phq9", ts=at(7, 20), total=13, item9=2)])
|
||
assert "phq9_item9" not in codes(evaluate(old))
|
||
|
||
|
||
def test_crisis_never_deduplicated() -> None:
|
||
s = snap(
|
||
13,
|
||
scale_results=[ScalePoint(scale="phq9", ts=at(13, 20), total=12, item9=2)],
|
||
open_alert_codes={"phq9_item9"},
|
||
)
|
||
assert "phq9_item9" in codes(evaluate(s), "crisis")
|
||
|
||
|
||
def test_cssrs_levels() -> None:
|
||
high = evaluate(snap(5, cssrs_risk="high"))
|
||
assert codes(high, "crisis") == {"cssrs_high"}
|
||
moderate = evaluate(snap(5, cssrs_risk="moderate"))
|
||
assert [(a.code, a.level) for a in moderate] == [("cssrs_moderate", AlertLevel.HIGH)]
|
||
assert evaluate(snap(5, cssrs_risk="low")) == []
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("weapon", "cssrs", "item9", "fires"),
|
||
[
|
||
(True, "moderate", None, True),
|
||
(True, None, 1, True),
|
||
(True, "low", 0, False),
|
||
(False, "high", 2, False),
|
||
(None, "moderate", None, False),
|
||
],
|
||
)
|
||
def test_weapon_access_with_risk(
|
||
weapon: bool | None, cssrs: str | None, item9: int | None, fires: bool
|
||
) -> None:
|
||
scales = [] if item9 is None else [ScalePoint(scale="phq9", ts=at(5, 20), total=10, item9=item9)]
|
||
s = snap(5, weapon_access=weapon, cssrs_risk=cssrs, scale_results=scales)
|
||
assert ("weapon_access_risk" in codes(evaluate(s), "crisis")) is fires
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------- scale rules
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("scale", "total", "code", "fires"),
|
||
[
|
||
("phq9", 20, "phq9_severe", True),
|
||
("phq9", 19, "phq9_severe", False),
|
||
("pcl5", 50, "pcl5_severe", True),
|
||
("pcl5", 49, "pcl5_severe", False),
|
||
],
|
||
)
|
||
def test_severe_totals(scale: str, total: int, code: str, fires: bool) -> None:
|
||
s = snap(7, scale_results=[ScalePoint(scale=scale, ts=at(7, 10), total=total, item9=0)])
|
||
acts = evaluate(s)
|
||
assert (code in codes(acts)) is fires
|
||
if fires:
|
||
assert next(a for a in acts if a.code == code).level == AlertLevel.HIGH
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("scale", "prev", "last", "code", "fires"),
|
||
[
|
||
("phq9", 10, 15, "phq9_rise", True),
|
||
("phq9", 10, 14, "phq9_rise", False),
|
||
("pcl5", 30, 40, "pcl5_rise", True),
|
||
("pcl5", 30, 39, "pcl5_rise", False),
|
||
("pcl5", 40, 30, "pcl5_rise", False),
|
||
],
|
||
)
|
||
def test_rise_between_last_two(scale: str, prev: int, last: int, code: str, fires: bool) -> None:
|
||
s = snap(
|
||
14,
|
||
scale_results=[
|
||
ScalePoint(scale=scale, ts=at(14, 10), total=last, item9=0), # order in input does not matter
|
||
ScalePoint(scale=scale, ts=at(0, 10), total=prev, item9=0),
|
||
],
|
||
)
|
||
assert (code in codes(evaluate(s))) is fires
|
||
|
||
|
||
def test_future_scale_results_ignored() -> None:
|
||
s = snap(7, scale_results=[ScalePoint(scale="phq9", ts=at(8), total=25, item9=3)])
|
||
assert evaluate(s) == []
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("sex", "total", "fires"), [("m", 4, True), ("m", 3, False), ("f", 3, True), ("f", 2, False)]
|
||
)
|
||
def test_auditc_by_sex(sex: str, total: int, fires: bool) -> None:
|
||
s = snap(1, sex=sex, scale_results=[ScalePoint(scale="auditc", ts=at(0), total=total)])
|
||
acts = evaluate(s)
|
||
assert ("auditc_positive" in codes(acts)) is fires
|
||
if fires:
|
||
assert acts[0].level == AlertLevel.MODERATE
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------- adherence
|
||
|
||
|
||
def test_quit_intent_voiced() -> None:
|
||
acts = evaluate(snap(5, checkins=[ci(5, med_taken=False, quit_intent=True)]))
|
||
assert [(a.type, a.level) for a in acts if a.code == "quit_intent"] == [
|
||
("alert", AlertLevel.HIGH),
|
||
("med_info", None),
|
||
]
|
||
assert next(a for a in acts if a.type == "med_info").params["topic"] == "delayed_effect"
|
||
|
||
|
||
def test_quit_intent_followed_by_skipped_dose() -> None:
|
||
s = snap(
|
||
7, checkins=[ci(5, med_taken=True, quit_intent=True), ci(6, med_taken=True), ci(7, med_taken=False)]
|
||
)
|
||
assert "quit_intent" in codes(evaluate(s), "alert")
|
||
no_intent = snap(7, checkins=[ci(6, med_taken=True), ci(7, med_taken=False)])
|
||
assert "quit_intent" not in codes(evaluate(no_intent))
|
||
|
||
|
||
def test_missed_doses_3_of_7() -> None:
|
||
doses = [
|
||
DoseRecord(planned_at=at(d, 9), status=st)
|
||
for d, st in [(1, "missed"), (2, "taken"), (3, "skipped"), (4, "taken"), (5, "taken"), (6, "taken")]
|
||
]
|
||
s = snap(7, doses=doses, checkins=[ci(7, med_taken=False)])
|
||
acts = [a for a in evaluate(s) if a.code == "missed_doses_7d"]
|
||
assert [a.type for a in acts] == ["alert", "reach_out"]
|
||
assert acts[0].level == AlertLevel.MODERATE and acts[0].params["count"] == 3
|
||
# Day 1 slides out of the 7-day window on day 8 → only 2 missed days left.
|
||
assert "missed_doses_7d" not in codes(evaluate(snap(8, doses=doses, checkins=[ci(7, med_taken=False)])))
|
||
|
||
|
||
def test_side_effects_three_consecutive_days() -> None:
|
||
two = snap(3, checkins=[ci(1), ci(2, side_effects=["nausea"]), ci(3, side_effects=["nausea"])])
|
||
assert evaluate(two) == []
|
||
gap = snap(
|
||
4,
|
||
checkins=[
|
||
ci(1, side_effects=["nausea"]),
|
||
ci(3, side_effects=["nausea"]),
|
||
ci(4, side_effects=["nausea"]),
|
||
],
|
||
)
|
||
assert evaluate(gap) == []
|
||
three = snap(4, checkins=[ci(d, side_effects=["nausea"]) for d in (2, 3, 4)])
|
||
acts = evaluate(three)
|
||
assert [(a.code, a.level) for a in acts] == [("side_effect_3d", AlertLevel.MODERATE)]
|
||
assert "nausea" in acts[0].reason_ru
|
||
|
||
|
||
def test_streak_seen_in_morning_before_todays_checkin() -> None:
|
||
s = snap(5, as_of=at(5, 8), checkins=[ci(d, side_effects=["nausea"]) for d in (2, 3, 4)])
|
||
assert "side_effect_3d" in codes(evaluate(s))
|
||
stale = snap(6, as_of=at(6, 8), checkins=[ci(d, side_effects=["nausea"]) for d in (2, 3, 4)])
|
||
assert evaluate(stale) == []
|
||
|
||
|
||
# ---------------------------------------------------------------------------------- check-in patterns
|
||
|
||
|
||
def test_sleep_lt4_three_nights() -> None:
|
||
s = snap(10, checkins=[ci(d, sleep_hours_bucket="<4") for d in (8, 9, 10)])
|
||
assert [(a.code, a.level) for a in evaluate(s)] == [("sleep_lt4_3n", AlertLevel.MODERATE)]
|
||
s2 = snap(
|
||
10,
|
||
checkins=[
|
||
ci(8, sleep_hours_bucket="<4"),
|
||
ci(9, sleep_hours_bucket="4-6"),
|
||
ci(10, sleep_hours_bucket="<4"),
|
||
],
|
||
)
|
||
assert evaluate(s2) == []
|
||
|
||
|
||
def test_alcohol_three_days() -> None:
|
||
s = snap(10, checkins=[ci(8, alcohol="1-2"), ci(9, alcohol="more"), ci(10, alcohol="more")])
|
||
acts = evaluate(s)
|
||
assert [a.type for a in acts] == ["alert", "med_info", "suggest_visit"]
|
||
assert acts[0].level == AlertLevel.MODERATE
|
||
assert acts[1].params["topic"] == "alcohol_interaction"
|
||
s2 = snap(10, checkins=[ci(8, alcohol="1-2"), ci(9, alcohol="none"), ci(10, alcohol="more")])
|
||
assert evaluate(s2) == []
|
||
|
||
|
||
def test_low_mood_schedules_unscheduled_phq9_with_cooldown() -> None:
|
||
checkins = [ci(11, mood=3), ci(12, mood=3), ci(13, mood=2)]
|
||
acts = evaluate(snap(13, checkins=checkins))
|
||
assert [(a.type, a.level, a.params) for a in acts] == [
|
||
("schedule_scale", None, {"scale": "phq9", "scheduled": False})
|
||
]
|
||
done = ScalePoint(scale="phq9", ts=at(13, 20), total=12, item9=0, scheduled=False)
|
||
assert evaluate(snap(13, checkins=checkins, scale_results=[done])) == []
|
||
assert evaluate(snap(13, checkins=[ci(11, mood=3), ci(12, mood=4), ci(13, mood=2)])) == []
|
||
|
||
|
||
def test_silence_over_3_days() -> None:
|
||
acts = evaluate(snap(10, last_patient_message_at=at(6, 20)))
|
||
assert [(a.type, a.level) for a in acts] == [("alert", AlertLevel.INFO), ("reach_out", None)]
|
||
assert acts[0].params["silent_days"] == 4
|
||
assert evaluate(snap(10, last_patient_message_at=at(7, 21))) == [] # exactly 3 days
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("age", "day", "fires"), [(23, 8, True), (24, 28, True), (25, 8, False), (23, 29, False)]
|
||
)
|
||
def test_young_adult_enhanced_monitoring(age: int, day: int, fires: bool) -> None:
|
||
acts = evaluate(snap(day, age=age))
|
||
assert (acts == [a for a in acts if a.type == "enhanced_monitoring"] and bool(acts)) is fires
|
||
if fires:
|
||
assert acts[0].level == AlertLevel.INFO
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------- dedup
|
||
|
||
|
||
def test_open_alert_suppresses_whole_rule() -> None:
|
||
checkins = [ci(8, alcohol="1-2"), ci(9, alcohol="more"), ci(10, alcohol="more")]
|
||
assert evaluate(snap(10, checkins=checkins, open_alert_codes={"alcohol_3d"})) == []
|
||
# Other rules are not affected by an unrelated open code.
|
||
assert codes(evaluate(snap(10, checkins=checkins, open_alert_codes={"sleep_lt4_3n"}))) == {"alcohol_3d"}
|
||
|
||
|
||
def test_evaluate_is_deterministic() -> None:
|
||
s = snap(10, checkins=[ci(d, alcohol="more", sleep_hours_bucket="<4") for d in (8, 9, 10)])
|
||
assert evaluate(s) == evaluate(s)
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------- colours
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("levels", "expected"),
|
||
[
|
||
([], "green"),
|
||
([AlertLevel.INFO], "grey"),
|
||
([AlertLevel.INFO, AlertLevel.MODERATE], "yellow"),
|
||
([AlertLevel.HIGH], "yellow"),
|
||
([AlertLevel.MODERATE, AlertLevel.ACUTE], "red"),
|
||
],
|
||
)
|
||
def test_color(levels: list[AlertLevel], expected: str) -> None:
|
||
assert color(levels) == expected
|
||
|
||
|
||
def test_patient_status() -> None:
|
||
st = patient_status([AlertLevel.MODERATE, None, AlertLevel.HIGH])
|
||
assert st.color == "yellow" and st.top_level == AlertLevel.HIGH
|
||
assert patient_status([]).top_level is None
|
||
|
||
|
||
# ---------------------------------------------------------------------------------------------- demo persona
|
||
# docs/product/01_demo_persona.md, scenario B: Alexey, 28, sertraline 50 mg from day 0.
|
||
# (day, med_taken, mood, sleep, alcohol, side effects, quit_intent)
|
||
PERSONA_DAYS: list[tuple[int, bool, int, str, str, list[str], bool]] = [
|
||
(1, True, 4, "6-8", "none", [], False),
|
||
(2, True, 4, "4-6", "none", ["nausea"], False),
|
||
(3, True, 4, "4-6", "none", ["nausea"], False),
|
||
(4, True, 3, "4-6", "none", ["nausea"], False),
|
||
(5, False, 3, "4-6", "none", [], True), # «от таблеток только хуже, брошу»
|
||
(6, True, 4, "6-8", "none", [], False), # doctor called, patient continues
|
||
(7, True, 4, "6-8", "none", [], False),
|
||
(8, True, 4, "<4", "1-2", [], False),
|
||
(9, True, 4, "<4", "more", [], False),
|
||
(10, True, 4, "<4", "more", [], False), # «не могу уснуть без пива»
|
||
(11, True, 3, "4-6", "1-2", [], False),
|
||
(12, True, 3, "4-6", "none", [], False),
|
||
(13, True, 2, "<4", "none", [], False), # «всё бессмысленно, устал от всего»
|
||
]
|
||
PERSONA_SCALES = [
|
||
ScalePoint(scale="phq9", ts=at(-3, 12), total=14, item9=0, scheduled=True), # screening
|
||
ScalePoint(scale="phq9", ts=at(7, 19), total=13, item9=0, scheduled=True), # weekly PHQ-9
|
||
ScalePoint(scale="phq9", ts=at(13, 20), total=18, item9=2, scheduled=False), # unscheduled (mood ≤ 3 × 3)
|
||
]
|
||
EXPECTED = ["G", "G", "G", "Y", "Y", "G", "G", "G", "G", "Y", "Y", "Y", "R"]
|
||
COLOR_LETTER = {"green": "G", "yellow": "Y", "red": "R", "grey": "g"}
|
||
DOCTOR_ACK_DAY = 6 # the doctor called the patient and acked the open alerts
|
||
|
||
|
||
def _persona_snapshot(day: int, open_codes: set[str], include_scales_until: datetime) -> PatientSnapshot:
|
||
rows = [r for r in PERSONA_DAYS if r[0] <= day]
|
||
return snap(
|
||
day,
|
||
checkins=[
|
||
ci(
|
||
d,
|
||
med_taken=med,
|
||
mood=mood,
|
||
sleep_hours_bucket=sleep,
|
||
alcohol=alc,
|
||
side_effects=se,
|
||
quit_intent=qi,
|
||
)
|
||
for d, med, mood, sleep, alc, se, qi in rows
|
||
],
|
||
doses=[DoseRecord(planned_at=at(d, 9), status="taken" if med else "skipped") for d, med, *_ in rows],
|
||
scale_results=[p for p in PERSONA_SCALES if p.ts <= include_scales_until],
|
||
open_alert_codes=set(open_codes),
|
||
)
|
||
|
||
|
||
def test_demo_persona_trajectory() -> None:
|
||
open_alerts: dict[str, AlertLevel] = {}
|
||
colours: list[str] = []
|
||
new_alerts: dict[int, set[str]] = {}
|
||
for day in range(1, 14):
|
||
if day == DOCTOR_ACK_DAY:
|
||
open_alerts.clear() # quit_intent + side_effect_3d acked after the call
|
||
if day == 13:
|
||
# Before the unscheduled PHQ-9: mood ≤ 3 for 3 days triggers it.
|
||
pre = evaluate(_persona_snapshot(day, set(open_alerts), include_scales_until=at(13, 19)))
|
||
assert codes(pre, "schedule_scale") == {"mood_low_3d"}
|
||
acts = evaluate(_persona_snapshot(day, set(open_alerts), include_scales_until=at(day, 23)))
|
||
fresh = {a.code: a.level for a in acts if a.creates_alert and a.level is not None}
|
||
new_alerts[day] = set(fresh)
|
||
open_alerts.update(fresh)
|
||
colours.append(COLOR_LETTER[color(open_alerts.values())])
|
||
|
||
assert colours == EXPECTED
|
||
assert new_alerts[4] == {"side_effect_3d"}
|
||
assert new_alerts[5] == {"quit_intent"}
|
||
assert new_alerts[10] == {"sleep_lt4_3n", "alcohol_3d"}
|
||
assert new_alerts[11] == set() and new_alerts[12] == set() # still open, not re-emitted
|
||
assert new_alerts[13] == {"phq9_item9", "phq9_rise"}
|
||
assert {d for d, c in new_alerts.items() if c} == {4, 5, 10, 13}
|