Pavel Eliseev
3cd7e2fd75
Correct RUS-AUDIT-S composition, preserve C-SSRS lifetime behavior, and prepare clinical review materials. Refs #52
275 lines
10 KiB
Python
275 lines
10 KiB
Python
"""Synthetic boundary, branching and safety regression cases for intake (#52)."""
|
|
|
|
from itertools import product
|
|
|
|
import pytest
|
|
|
|
from app.core.questionnaire import (
|
|
IntakePolicy,
|
|
alcohol_expansion,
|
|
evaluate_intake,
|
|
extend_alcohol,
|
|
load_policy,
|
|
score_alcohol,
|
|
score_cssrs_recent,
|
|
)
|
|
from app.core.scales import load_scale
|
|
|
|
|
|
def approved(mode="graded"):
|
|
data = load_policy().model_dump()
|
|
data.update(cssrs_positive_policy=mode, rus_audit_s_expand_from_zone=2)
|
|
data["approval"] = {
|
|
"reviewer": "SYNTHETIC TEST ONLY",
|
|
"decision_reference": "test-fixture-not-clinical-approval",
|
|
"approved_at": "2026-10-03T12:00:00+03:00",
|
|
"policy_version": data["version"],
|
|
}
|
|
return IntakePolicy.model_validate(data)
|
|
|
|
|
|
def zeros(scale):
|
|
return {key: 0 for key in load_scale(scale).item_ids}
|
|
|
|
|
|
def intake(**changes):
|
|
args = dict(
|
|
phq9={"phq9_1": 0, "phq9_2": 0},
|
|
pcptsd5={},
|
|
pcl5={},
|
|
alcohol=zeros("rus_audit_s"),
|
|
cssrs={"q1": 0, "q2": 0, "q6_lifetime": 0},
|
|
sex="f",
|
|
trauma_exposure=False,
|
|
policy=approved(),
|
|
)
|
|
args.update(changes)
|
|
return evaluate_intake(**args)
|
|
|
|
|
|
@pytest.mark.parametrize("sex", ["f", "m"])
|
|
def test_every_short_alcohol_answer_combination(sex):
|
|
# Exhaustively cover all valid patterns, including sparse 0/2/4 history options.
|
|
for q3, q9, q10 in product(range(5), (0, 2, 4), (0, 2, 4)):
|
|
total = q3 + q9 + q10
|
|
expected = (
|
|
(1 if total < 2 else 2 if total < 5 else 3 if total == 5 else 4)
|
|
if sex == "f"
|
|
else (1 if total < 4 else 2 if total < 7 else 3 if total == 7 else 4)
|
|
)
|
|
result = score_alcohol(
|
|
dict(zip(load_scale("rus_audit_s").item_ids, (q3, q9, q10), strict=True)), short=True, sex=sex
|
|
)
|
|
assert result.complete and result.total == total and result.zone == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"sex,bounds",
|
|
[
|
|
("f", [(0, 1), (4, 1), (5, 2), (9, 2), (10, 3), (11, 4), (40, 4)]),
|
|
("m", [(0, 1), (8, 1), (9, 2), (13, 2), (14, 3), (16, 3), (17, 4), (40, 4)]),
|
|
],
|
|
)
|
|
def test_full_alcohol_boundaries(sex, bounds):
|
|
for total, expected in bounds:
|
|
values = zeros("rus_audit")
|
|
remaining = total
|
|
for item in load_scale("rus_audit").items:
|
|
value = max(o.value for o in item.options if o.value <= remaining)
|
|
values[item.id] = value
|
|
remaining -= value
|
|
assert remaining == 0
|
|
assert score_alcohol(values, short=False, sex=sex).zone == expected
|
|
|
|
|
|
def test_alcohol_missing_unknown_sex_and_no_implicit_approval():
|
|
r = score_alcohol({"rus_audit_3": 0}, short=True, sex="f")
|
|
assert not r.complete and r.total is None and r.zone is None
|
|
r = score_alcohol(zeros("rus_audit_s"), short=True, sex=None)
|
|
assert r.complete and r.total == 0 and r.zone is None
|
|
r = score_alcohol({"rus_audit_3": 4, "rus_audit_9": 4, "rus_audit_10": 4}, short=True, sex="m")
|
|
assert alcohol_expansion(r, load_policy()) is None
|
|
assert alcohol_expansion(r, approved()) is True
|
|
|
|
|
|
def test_full_extension_preserves_history_and_never_counts_twice():
|
|
short = {"rus_audit_3": 0, "rus_audit_9": 2, "rus_audit_10": 4}
|
|
extra = {"rus_audit_1": 0, **{f"rus_audit_{i}": 0 for i in range(4, 9)}, "rus_audit_9": 2}
|
|
merged = extend_alcohol(short, extra)
|
|
result = score_alcohol(merged, short=False, sex="f")
|
|
assert result.complete and result.total == 6
|
|
assert result.derived == ("rus_audit_2",)
|
|
assert "rus_audit_2" not in merged # caller-owned raw answers are not mutated
|
|
with pytest.raises(ValueError, match="conflicting"):
|
|
extend_alcohol(short, {"rus_audit_9": 0})
|
|
with pytest.raises(ValueError, match="conflicts"):
|
|
score_alcohol({"rus_audit_1": 0, "rus_audit_3": 1}, short=False, sex="m")
|
|
|
|
|
|
@pytest.mark.parametrize("bad", [True, "1", 1.0, -1, 5])
|
|
def test_alcohol_rejects_invalid_answers(bad):
|
|
with pytest.raises(ValueError):
|
|
score_alcohol({"rus_audit_3": bad}, short=True, sex="f")
|
|
|
|
|
|
def test_cannot_use_auditc_items_or_sparse_history_values():
|
|
for values in [{"rus_audit_1": 1}, {"auditc_1": 1}, {"rus_audit_9": 1}]:
|
|
with pytest.raises(ValueError):
|
|
score_alcohol(values, short=True, sex="f")
|
|
|
|
|
|
def test_cssrs_negative_branch_is_complete_without_inventing_zeroes():
|
|
a = {"q1": 0, "q2": 0, "q6_lifetime": 0}
|
|
r = score_cssrs_recent(a)
|
|
assert r.complete and r.risk == "none"
|
|
assert r.required == ("q1", "q2", "q6_lifetime")
|
|
assert set(a) == set(r.required)
|
|
|
|
|
|
def test_cssrs_remote_behavior_is_not_negative_and_missing_recency_not_low():
|
|
a = {"q1": 0, "q2": 0, "q6_lifetime": 1}
|
|
r = score_cssrs_recent(a)
|
|
assert not r.complete and r.risk is None and r.risk_lower_bound == "moderate"
|
|
assert r.missing == ("q6_recent",)
|
|
assert score_cssrs_recent({**a, "q6_recent": 0}).risk == "moderate"
|
|
assert score_cssrs_recent({**a, "q6_recent": 1}).risk == "high"
|
|
|
|
|
|
def test_cssrs_ideation_branch_and_partial_high():
|
|
assert score_cssrs_recent({}).risk is None
|
|
assert score_cssrs_recent({"q2": 1}).missing == ("q1", "q3", "q4", "q5", "q6_lifetime")
|
|
r = score_cssrs_recent({"q4": 1})
|
|
assert not r.complete and r.risk is None and r.risk_lower_bound == "high"
|
|
|
|
|
|
def test_cssrs_exhaustive_complete_ideation_patterns():
|
|
for q1, q3, q4, q5, lifetime, recent in product((0, 1), repeat=6):
|
|
a = dict(q1=q1, q2=1, q3=q3, q4=q4, q5=q5, q6_lifetime=lifetime)
|
|
if lifetime:
|
|
a["q6_recent"] = recent
|
|
expected = "high" if q4 or q5 or (lifetime and recent) else "moderate" if q3 or lifetime else "low"
|
|
r = score_cssrs_recent(a)
|
|
assert r.complete and r.risk == expected
|
|
|
|
|
|
def test_cssrs_conflict_preserves_high_signal():
|
|
r = score_cssrs_recent({"q1": 0, "q2": 0, "q4": 1, "q6_lifetime": 0, "q6_recent": 1})
|
|
assert not r.complete and r.risk_lower_bound == "high" and len(r.conflicts) == 2
|
|
|
|
|
|
@pytest.mark.parametrize("bad", [None, True, "Да", 1.0, 2, -1])
|
|
def test_cssrs_requires_explicit_binary_integer(bad):
|
|
with pytest.raises(ValueError):
|
|
score_cssrs_recent({"q1": bad})
|
|
|
|
|
|
def test_unapproved_policy_never_continues_even_a_negative_screen():
|
|
r = intake(policy=load_policy())
|
|
assert r.status == "clinical_review_required" and r.primary_route is None
|
|
assert intake(policy=load_policy(), cssrs={"q1": 1}).status == "clinical_review_required"
|
|
|
|
|
|
def test_immediate_safety_preempts_unfinished_or_invalid_unrelated_scales():
|
|
for overrides in [
|
|
dict(help_now=True, cssrs={"broken": "data"}),
|
|
dict(phq9={"phq9_9": 1}),
|
|
dict(cssrs={"q4": 1}, alcohol={"bad": 50}),
|
|
]:
|
|
assert intake(policy=load_policy(), **overrides).status == "crisis"
|
|
r = intake(phq9={"phq9_9": 1})
|
|
assert "phq9_cssrs_discrepancy" in r.flags
|
|
|
|
|
|
def test_physician_must_choose_between_stop_and_graded():
|
|
low = {"q1": 1, "q2": 0, "q6_lifetime": 0}
|
|
assert intake(cssrs=low, policy=approved("stop_any")).status == "crisis"
|
|
r = intake(cssrs=low, policy=approved("graded"))
|
|
assert r.status == "ready" and "cssrs_low_safety_plan" in r.flags
|
|
data = approved().model_dump()
|
|
data["version"] = "changed"
|
|
with pytest.raises(ValueError, match="matching version"):
|
|
IntakePolicy.model_validate(data)
|
|
|
|
|
|
def test_phq2_and_trauma_expansion_do_not_turn_missing_into_negative():
|
|
assert intake().primary_route == "self_help"
|
|
r = intake(phq9={"phq9_1": 1, "phq9_2": 1})
|
|
assert r.status == "incomplete" and "phq9_9" in r.missing
|
|
assert intake(trauma_exposure=None).status == "incomplete"
|
|
assert intake(trauma_exposure=True, pcptsd5={"pcptsd5_1": 1}).status == "incomplete"
|
|
pc = {f"pcptsd5_{i}": int(i <= 3) for i in range(1, 6)}
|
|
assert intake(trauma_exposure=True, pcptsd5=pc).status == "incomplete"
|
|
assert intake(trauma_exposure=True, pcptsd5=pc, pcl5=zeros("pcl5")).primary_route == "psychologist"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"total,route",
|
|
[
|
|
(30, "self_help"),
|
|
(31, "psychiatrist_or_psychotherapist"),
|
|
(49, "psychiatrist_or_psychotherapist"),
|
|
(50, "priority_psychiatrist"),
|
|
],
|
|
)
|
|
def test_pdf_pcl_boundaries_are_distinct_from_legacy_33(total, route):
|
|
values = zeros("pcl5")
|
|
for key in values:
|
|
values[key] = min(4, total)
|
|
total -= values[key]
|
|
assert intake(pcl5=values).primary_route == route
|
|
|
|
|
|
def test_urgent_route_and_parallel_referrals_do_not_overwrite_each_other():
|
|
alcohol = {i.id: i.max_value for i in load_scale("rus_audit").items}
|
|
r = intake(
|
|
cssrs={"q1": 0, "q2": 0, "q6_lifetime": 1, "q6_recent": 0},
|
|
alcohol=alcohol,
|
|
traumatic_brain_injury=True,
|
|
psychotropics=True,
|
|
)
|
|
assert r.primary_route == "urgent_psychiatrist"
|
|
assert r.parallel_routes == ("narcologist", "neurologist")
|
|
assert "alcohol_psychotropics" in r.flags
|
|
|
|
|
|
def test_alcohol_expansion_and_unknown_sex_block_routine_route():
|
|
assert intake(alcohol={"rus_audit_3": 2, "rus_audit_9": 0, "rus_audit_10": 0}).status == "incomplete"
|
|
r = intake(sex=None)
|
|
assert r.status == "incomplete" and "alcohol_scoring_sex" in r.missing
|
|
|
|
|
|
def test_full_non_drinker_derivation_also_satisfies_short_stage():
|
|
values = zeros("rus_audit")
|
|
del values["rus_audit_2"]
|
|
del values["rus_audit_3"]
|
|
assert intake(alcohol=values).status == "ready"
|
|
|
|
|
|
def test_crisis_keeps_other_flags_and_bad_data_cannot_hide_explicit_signal():
|
|
alcohol = {i.id: i.max_value for i in load_scale("rus_audit").items}
|
|
r = intake(cssrs={"q4": 1}, pcl5={"pcl5_15": 3}, alcohol=alcohol, traumatic_brain_injury=True, psychotropics=True)
|
|
assert r.status == "crisis"
|
|
assert {"pcl5_behavior", "alcohol_zone_iv", "alcohol_psychotropics", "traumatic_brain_injury"} <= set(r.flags)
|
|
assert intake(cssrs={"q4": 1, "invalid": 7}).status == "crisis"
|
|
assert intake(phq9={"phq9_9": 1}, cssrs={"invalid": 7}).status == "crisis"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"total,route",
|
|
[
|
|
(4, "self_help"),
|
|
(5, "psychologist"),
|
|
(9, "psychologist"),
|
|
(10, "psychiatrist_or_psychotherapist"),
|
|
(19, "psychiatrist_or_psychotherapist"),
|
|
(20, "priority_psychiatrist"),
|
|
(24, "priority_psychiatrist"),
|
|
],
|
|
)
|
|
def test_phq_routes_at_each_boundary_without_positive_item9(total, route):
|
|
values = zeros("phq9")
|
|
for key in list(values)[:8]:
|
|
values[key] = min(3, total)
|
|
total -= values[key]
|
|
assert intake(phq9=values).primary_route == route
|