Pavel Eliseev
c42ef621e5
CI / ruff · pytest (pull_request) Successful in 55s
Add gated TSQ, Russian C-SSRS Recent, durable responses and notifications, diaries, schedules and audited clinical review. Preserve manual assessment when the Russian follow-up source is unavailable. Refs #52
325 lines
9.7 KiB
Python
325 lines
9.7 KiB
Python
from pathlib import Path
|
||
|
||
import pytest
|
||
import yaml
|
||
from pydantic import ValidationError
|
||
|
||
from app.core.scales import ScaleDef, ScaleResult, delta, load_all, load_scale, score
|
||
|
||
|
||
def answers_for(scale_id: str, total: int) -> dict[str, int]:
|
||
"""Fill items greedily (each up to its max option) so the plain sum equals `total`."""
|
||
scale = load_scale(scale_id)
|
||
out: dict[str, int] = {}
|
||
left = total
|
||
for item in scale.items:
|
||
v = min(left, item.max_value)
|
||
out[item.id] = v
|
||
left -= v
|
||
assert left == 0, f"{scale_id}: total {total} unreachable"
|
||
return out
|
||
|
||
|
||
def severity(scale_id: str, total: int, **kw: object) -> str:
|
||
return score(scale_id, answers_for(scale_id, total), **kw).severity_code
|
||
|
||
|
||
# ---------------------------------------------------------------- loading
|
||
|
||
|
||
def test_all_scales_load_with_expected_item_counts() -> None:
|
||
counts = {k: len(v.items) for k, v in load_all().items()}
|
||
expected = {"phq9": 9, "gad7": 7, "pcl5": 20, "auditc": 3, "pcptsd5": 5, "cssrs_screen": 6, "phq4": 4, "dar5": 5}
|
||
expected.update(rus_audit_s=3, rus_audit=10, tsq=10)
|
||
assert counts == expected
|
||
|
||
|
||
def test_unknown_scale() -> None:
|
||
with pytest.raises(KeyError):
|
||
load_scale("nope")
|
||
|
||
|
||
def test_phq9_item9_buttons_only() -> None:
|
||
phq9 = load_scale("phq9")
|
||
assert [i.id for i in phq9.items if i.buttons_only] == ["phq9_9"]
|
||
assert all(i.buttons_only for i in load_scale("cssrs_screen").items)
|
||
|
||
|
||
def test_clinical_notes_present() -> None:
|
||
for scale in load_all().values():
|
||
assert scale.translation_note, scale.id
|
||
assert scale.license_note, scale.id
|
||
|
||
|
||
# ---------------------------------------------------------------- band boundaries
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("scale_id", "boundaries"),
|
||
[
|
||
(
|
||
"phq9",
|
||
[
|
||
(0, "minimal"),
|
||
(4, "minimal"),
|
||
(5, "mild"),
|
||
(9, "mild"),
|
||
(10, "moderate"),
|
||
(14, "moderate"),
|
||
(15, "moderately_severe"),
|
||
(19, "moderately_severe"),
|
||
(20, "severe"),
|
||
(27, "severe"),
|
||
],
|
||
),
|
||
(
|
||
"gad7",
|
||
[
|
||
(0, "minimal"),
|
||
(4, "minimal"),
|
||
(5, "mild"),
|
||
(9, "mild"),
|
||
(10, "moderate"),
|
||
(14, "moderate"),
|
||
(15, "severe"),
|
||
(21, "severe"),
|
||
],
|
||
),
|
||
(
|
||
"pcl5",
|
||
[
|
||
(0, "below_threshold"),
|
||
(32, "below_threshold"),
|
||
(33, "probable"),
|
||
(49, "probable"),
|
||
(50, "high"),
|
||
(80, "high"),
|
||
],
|
||
),
|
||
("pcptsd5", [(0, "negative"), (2, "negative"), (3, "positive"), (5, "positive")]),
|
||
("phq4", [(0, "normal"), (2, "normal"), (3, "mild"), (5, "mild"), (6, "moderate"), (9, "severe")]),
|
||
],
|
||
)
|
||
def test_band_boundaries(scale_id: str, boundaries: list[tuple[int, str]]) -> None:
|
||
for total, code in boundaries:
|
||
assert severity(scale_id, total) == code, (scale_id, total)
|
||
|
||
|
||
def test_result_fields() -> None:
|
||
r = score("phq9", answers_for("phq9", 12))
|
||
assert r == ScaleResult(
|
||
scale="phq9",
|
||
total=12,
|
||
max_total=27,
|
||
severity_code="moderate",
|
||
severity_label="Умеренная",
|
||
complete=True,
|
||
answered=9,
|
||
flags=[],
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------- flags
|
||
|
||
|
||
def test_phq9_item9_flag() -> None:
|
||
base = {f"phq9_{i}": 0 for i in range(1, 10)}
|
||
assert score("phq9", base).flags == []
|
||
r = score("phq9", base | {"phq9_9": 1})
|
||
assert r.flags == ["phq9_item9_positive"]
|
||
# Item 9 flag fires regardless of the low total.
|
||
assert r.severity_code == "minimal"
|
||
|
||
|
||
def test_phq9_item9_flag_in_incomplete_questionnaire() -> None:
|
||
r = score("phq9", {"phq9_9": 2})
|
||
assert not r.complete
|
||
assert "phq9_item9_positive" in r.flags
|
||
|
||
|
||
def test_pcl5_probable_flag() -> None:
|
||
assert score("pcl5", answers_for("pcl5", 32)).flags == []
|
||
assert score("pcl5", answers_for("pcl5", 33)).flags == ["pcl5_probable_ptsd"]
|
||
|
||
|
||
def test_pcptsd5_flag() -> None:
|
||
assert score("pcptsd5", answers_for("pcptsd5", 2)).flags == []
|
||
assert score("pcptsd5", answers_for("pcptsd5", 3)).flags == ["pcptsd5_positive"]
|
||
|
||
|
||
def test_phq4_subscale_flags() -> None:
|
||
zero = {"phq4_1": 0, "phq4_2": 0, "phq4_3": 0, "phq4_4": 0}
|
||
assert score("phq4", zero | {"phq4_1": 1, "phq4_2": 1, "phq4_3": 1, "phq4_4": 1}).flags == []
|
||
assert score("phq4", zero | {"phq4_1": 2, "phq4_2": 1}).flags == ["phq4_anxiety_positive"]
|
||
assert score("phq4", zero | {"phq4_3": 3}).flags == ["phq4_depression_positive"]
|
||
both = score("phq4", {"phq4_1": 3, "phq4_2": 0, "phq4_3": 0, "phq4_4": 3})
|
||
assert both.flags == ["phq4_anxiety_positive", "phq4_depression_positive"]
|
||
|
||
|
||
# ---------------------------------------------------------------- AUDIT-C by sex
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("total", "sex", "code"),
|
||
[
|
||
(3, "m", "negative"),
|
||
(4, "m", "positive"),
|
||
(2, "f", "negative"),
|
||
(3, "f", "positive"),
|
||
# Unknown sex: the more sensitive (female) cut-off.
|
||
(2, None, "negative"),
|
||
(3, None, "positive"),
|
||
(0, "m", "negative"),
|
||
(12, "f", "positive"),
|
||
],
|
||
)
|
||
def test_auditc_by_sex(total: int, sex: str | None, code: str) -> None:
|
||
r = score("auditc", answers_for("auditc", total), sex=sex) # type: ignore[arg-type]
|
||
assert r.severity_code == code
|
||
assert (r.flags == ["auditc_positive"]) == (code == "positive")
|
||
|
||
|
||
# ---------------------------------------------------------------- C-SSRS triage
|
||
|
||
|
||
def cssrs(*yes: int) -> dict[str, int]:
|
||
return {f"cssrs_{i}": int(i in yes) for i in range(1, 7)}
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("yes", "level"),
|
||
[
|
||
((), "none"),
|
||
((1,), "low"),
|
||
((2,), "low"),
|
||
((1, 2), "low"),
|
||
((1, 2, 3), "moderate"),
|
||
((3,), "moderate"),
|
||
((1, 2, 3, 4), "high"),
|
||
((5,), "high"),
|
||
((6,), "high"),
|
||
((1, 6), "high"),
|
||
],
|
||
)
|
||
def test_cssrs_triage(yes: tuple[int, ...], level: str) -> None:
|
||
r = score("cssrs_screen", cssrs(*yes))
|
||
assert r.severity_code == level
|
||
assert r.total == len(yes)
|
||
assert r.max_total == 6
|
||
|
||
|
||
def test_cssrs_flags() -> None:
|
||
assert score("cssrs_screen", cssrs(4)).flags == ["cssrs_high"]
|
||
assert score("cssrs_screen", cssrs(3)).flags == ["cssrs_moderate"]
|
||
assert score("cssrs_screen", cssrs(1)).flags == []
|
||
|
||
|
||
def test_cssrs_partial_high_answer_still_high() -> None:
|
||
# Screen version skips q3–q5 when q2 = no, so partial answers are normal; risk must still be detected.
|
||
r = score("cssrs_screen", {"cssrs_1": 1, "cssrs_2": 0, "cssrs_6": 1})
|
||
assert r.severity_code == "high"
|
||
assert not r.complete
|
||
|
||
|
||
# ---------------------------------------------------------------- incomplete / invalid
|
||
|
||
|
||
def test_incomplete_no_imputation() -> None:
|
||
r = score("phq9", {"phq9_1": 3, "phq9_2": 3})
|
||
assert r.total == 6
|
||
assert r.answered == 2
|
||
assert r.complete is False
|
||
assert r.severity_code == "mild" # lower bound, not prorated (prorating would give 27)
|
||
|
||
|
||
def test_empty_answers() -> None:
|
||
r = score("gad7", {})
|
||
assert (r.total, r.answered, r.complete, r.severity_code) == (0, 0, False, "minimal")
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"answers",
|
||
[
|
||
{"phq9_1": 4},
|
||
{"phq9_1": -1},
|
||
{"phq9_10": 1},
|
||
{"gad7_1": 1},
|
||
{"phq9_1": True},
|
||
{"phq9_1": "2"},
|
||
],
|
||
)
|
||
def test_invalid_answers(answers: dict[str, int]) -> None:
|
||
with pytest.raises(ValueError):
|
||
score("phq9", answers)
|
||
|
||
|
||
def test_cssrs_rejects_non_binary() -> None:
|
||
with pytest.raises(ValueError):
|
||
score("cssrs_screen", {"cssrs_1": 2})
|
||
|
||
|
||
def test_pcl5_accepts_4_phq9_does_not() -> None:
|
||
assert score("pcl5", {"pcl5_1": 4}).total == 4
|
||
with pytest.raises(ValueError):
|
||
score("phq9", {"phq9_1": 4})
|
||
|
||
|
||
# ---------------------------------------------------------------- delta
|
||
|
||
|
||
def test_delta() -> None:
|
||
prev = score("phq9", answers_for("phq9", 8))
|
||
cur = score("phq9", answers_for("phq9", 14))
|
||
assert delta(prev, cur) == 6
|
||
assert delta(cur, prev) == -6
|
||
|
||
|
||
def test_delta_rejects_mismatch_and_incomplete() -> None:
|
||
phq = score("phq9", answers_for("phq9", 8))
|
||
with pytest.raises(ValueError):
|
||
delta(phq, score("gad7", answers_for("gad7", 8)))
|
||
with pytest.raises(ValueError):
|
||
delta(phq, score("phq9", {"phq9_1": 1}))
|
||
|
||
|
||
# ---------------------------------------------------------------- schema validation
|
||
|
||
|
||
def _raw(scale_id: str) -> dict:
|
||
path = Path(__file__).resolve().parent.parent / "config" / "scales" / f"{scale_id}.yaml"
|
||
return yaml.safe_load(path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def test_schema_rejects_gap_in_bands() -> None:
|
||
raw = _raw("phq9")
|
||
raw["severity"][1]["min"] = 6
|
||
with pytest.raises(ValidationError):
|
||
ScaleDef.model_validate(raw)
|
||
|
||
|
||
def test_schema_rejects_wrong_range() -> None:
|
||
raw = _raw("gad7")
|
||
raw["range"]["max"] = 27
|
||
with pytest.raises(ValidationError):
|
||
ScaleDef.model_validate(raw)
|
||
|
||
|
||
def test_schema_rejects_unknown_item_in_flag() -> None:
|
||
raw = _raw("phq9")
|
||
raw["flags"][0]["when"]["item"] = "phq9_99"
|
||
with pytest.raises(ValidationError):
|
||
ScaleDef.model_validate(raw)
|
||
|
||
|
||
def test_schema_rejects_typo_key() -> None:
|
||
raw = _raw("phq9")
|
||
raw["flagz"] = raw.pop("flags")
|
||
with pytest.raises(ValidationError):
|
||
ScaleDef.model_validate(raw)
|
||
|
||
|
||
def test_load_all_from_custom_dir(tmp_path: Path) -> None:
|
||
(tmp_path / "scales").mkdir()
|
||
(tmp_path / "scales" / "gad7.yaml").write_text(yaml.safe_dump(_raw("gad7"), allow_unicode=True), "utf-8")
|
||
assert list(load_all(tmp_path)) == ["gad7"]
|