228 lines
9.7 KiB
Python
228 lines
9.7 KiB
Python
"""Pure dashboard logic: ordering, labels, dose strip, scales config, SNILS, chart builders, page discovery."""
|
||
|
||
import random
|
||
from datetime import date, datetime, timedelta
|
||
from pathlib import Path
|
||
from zoneinfo import ZoneInfo
|
||
|
||
from app.api import fixtures
|
||
from app.api.schemas import AlertOut, DiscontinuationRisk, DoseOut, SeriesPoint
|
||
from dashboard.app import discover_pages, slugify
|
||
from dashboard.components import charts, ui
|
||
from dashboard.components import helpers as h
|
||
|
||
TZ = ZoneInfo("Europe/Moscow")
|
||
|
||
|
||
def _risk(score: int, level: str = "low", factors: list[str] | None = None) -> DiscontinuationRisk:
|
||
return DiscontinuationRisk(score=score, level=level, factors=factors or [])
|
||
|
||
|
||
def test_sort_red_yellow_grey_green_then_risk():
|
||
items = list(reversed(fixtures.PATIENTS))
|
||
items.append(fixtures.PATIENTS[1].model_copy(update={"token": "g", "display_name": "Г", "status_color": "grey"}))
|
||
out = h.sort_patients(items)
|
||
colors = [p.status_color for p in out]
|
||
assert colors == sorted(colors, key=h.STATUS_ORDER.__getitem__)
|
||
assert out[0].token == "tok-andrey"
|
||
yellows = [p for p in out if p.status_color == "yellow"]
|
||
assert yellows[0].display_name == "Сергей П." # highest discontinuation risk among yellow
|
||
assert [p.discontinuation_risk.score for p in yellows] == sorted(
|
||
(p.discontinuation_risk.score for p in yellows), reverse=True
|
||
)
|
||
|
||
|
||
def test_filter_and_kpis():
|
||
items = fixtures.PATIENTS
|
||
assert {p.status_color for p in h.filter_patients(items, "yellow")} == {"yellow"}
|
||
assert [p.display_name for p in h.filter_patients(items, query="андр")] == ["Андрей С."]
|
||
assert len(h.filter_patients(items, query="пароксетин")) == 1
|
||
k = h.kpis(items)
|
||
assert (k.total, k.red, k.yellow, k.green, k.grey, k.high_risk) == (8, 1, 4, 3, 0, 1)
|
||
|
||
|
||
def test_trend_arrows():
|
||
assert h.trend_view("up").text == "↑" and h.trend_view("up").tone == "red"
|
||
assert h.trend_view("down").text == "↓" and h.trend_view("down").tone == "green"
|
||
assert h.trend_view("flat").text == "→"
|
||
assert h.trend_view("unknown").tone == "grey"
|
||
|
||
|
||
def test_risk_badge_and_tooltip():
|
||
v = h.risk_view(_risk(70, "high", ["пропуски 3 из 7", "нет ответа 2 дня"]))
|
||
assert v.text == "Высокий · 70" and v.tone == "red"
|
||
assert "• пропуски 3 из 7\n• нет ответа 2 дня" in v.hint
|
||
assert h.risk_view(_risk(35, "medium")).tone == "yellow"
|
||
assert "значимых факторов нет" in h.risk_view(_risk(5)).hint
|
||
|
||
|
||
def test_adherence_thresholds():
|
||
assert h.adherence_view(None).text == "—"
|
||
assert h.adherence_view(1.0).tone == "green"
|
||
assert h.adherence_view(0.8).tone == "green"
|
||
assert h.adherence_view(0.7).tone == "yellow"
|
||
assert h.adherence_view(0.57) == h.View("57%", "red")
|
||
|
||
|
||
def test_level_tone_matches_adr006():
|
||
assert h.level_tone("acute") == "red"
|
||
assert h.level_tone("high") == h.level_tone("moderate") == "yellow"
|
||
assert h.level_tone("info") == "grey"
|
||
|
||
|
||
def test_severity_bands_come_from_config():
|
||
bands = h.severity_bands("phq9")
|
||
assert [b.lo for b in bands] == [0, 5, 10, 15, 20]
|
||
assert h.severity_label("phq9", 18) == "Умеренно-тяжёлая"
|
||
assert h.severity_label("pcl5", 44) == "Вероятное ПТСР"
|
||
assert h.severity_label("pcl5", None) == ""
|
||
assert h.severity_bands("no_such_scale") == ()
|
||
|
||
|
||
def test_last_and_prev():
|
||
card = fixtures.andrey_card()
|
||
assert h.last_and_prev(card.series, "phq9") == (18, 13)
|
||
assert h.last_and_prev(card.series, "gad7") == (None, None)
|
||
|
||
|
||
def test_dose_grid_fills_gaps_and_counts():
|
||
d0 = datetime(2026, 9, 21, 8, tzinfo=TZ)
|
||
doses = [
|
||
DoseOut(planned_at=d0, drug="сертралин", status="taken"),
|
||
DoseOut(planned_at=d0 + timedelta(hours=12), drug="сертралин", status="missed"),
|
||
DoseOut(planned_at=d0 + timedelta(days=2), drug="сертралин", status="skipped"),
|
||
DoseOut(planned_at=d0 + timedelta(days=3), drug="сертралин", status="pending"),
|
||
]
|
||
g = h.dose_grid(doses)
|
||
assert g.days == [date(2026, 9, 21) + timedelta(days=i) for i in range(4)]
|
||
assert g.slots == ["08:00", "20:00"]
|
||
assert g.cells[("20:00", date(2026, 9, 21))] == "missed"
|
||
assert ("08:00", date(2026, 9, 22)) not in g.cells # no dose planned that day → empty cell
|
||
assert g.counts == {"taken": 1, "missed": 1, "skipped": 1, "pending": 1}
|
||
assert g.adherence == 1 / 3 # pending is not counted
|
||
|
||
|
||
def test_dose_grid_empty():
|
||
g = h.dose_grid([])
|
||
assert g.days == [] and g.adherence is None
|
||
|
||
|
||
def test_dose_grid_converts_to_moscow_time():
|
||
utc = datetime(2026, 9, 21, 5, tzinfo=ZoneInfo("UTC"))
|
||
g = h.dose_grid([DoseOut(planned_at=utc, drug="x", status="taken")])
|
||
assert g.slots == ["08:00"]
|
||
|
||
|
||
def _alert(i: int, level: str = "moderate", acked: bool = False, hours: int = 0) -> AlertOut:
|
||
ts = datetime(2026, 10, 3, 9, tzinfo=TZ) + timedelta(hours=hours)
|
||
return AlertOut(
|
||
id=i,
|
||
token="t",
|
||
patient_name="П",
|
||
ts=ts,
|
||
level=level,
|
||
color=h.level_tone(level),
|
||
code="c",
|
||
reason="r",
|
||
acked_by="d" if acked else None,
|
||
acked_at=ts if acked else None,
|
||
)
|
||
|
||
|
||
def test_alert_ordering_new_and_local_acks():
|
||
alerts = [_alert(1, "moderate", hours=5), _alert(2, "acute", hours=1), _alert(3, "high", acked=True, hours=9)]
|
||
assert [a.id for a in h.sort_alerts(alerts)] == [2, 1, 3]
|
||
assert [a.id for a in h.split_new(alerts, {1})] == [2, 3]
|
||
at = datetime(2026, 10, 3, 22, tzinfo=TZ)
|
||
out = h.apply_local_acks(alerts, {2: ("doc-smirnova", at)})
|
||
assert out[1].acked_by == "doc-smirnova" and out[1].acked_at == at
|
||
assert out[0].acked_at is None
|
||
|
||
|
||
def test_russian_formatting():
|
||
assert [h.age_label(n) for n in (21, 23, 34, 41, 45, 11, 112)] == [
|
||
"21 год",
|
||
"23 года",
|
||
"34 года",
|
||
"41 год",
|
||
"45 лет",
|
||
"11 лет",
|
||
"112 лет",
|
||
]
|
||
assert h.age_label(0) == "возраст не указан"
|
||
assert h.absolute_link("/console?code=X", "http://h:8501/") == "http://h:8501/console?code=X"
|
||
assert h.absolute_link("https://t.me/b?start=X", "http://h") == "https://t.me/b?start=X"
|
||
today = date(2026, 10, 3)
|
||
assert h.visit_label(date(2026, 10, 18), today) == "визит 18 окт · через 15 дней"
|
||
assert h.visit_label(date(2026, 10, 4), today) == "визит 4 окт · через 1 день"
|
||
assert h.visit_label(today, today) == "визит сегодня"
|
||
assert "просрочен" in h.visit_label(date(2026, 10, 1), today)
|
||
assert h.visit_label(None, today) == "визит не назначен"
|
||
assert h.when_label(datetime(2026, 10, 3, 21, tzinfo=TZ), today) == "сегодня, 21:00"
|
||
|
||
|
||
def test_snils_checksum():
|
||
assert h.snils_checksum("112233445") == 95 # well-known valid example 112-233-445 95
|
||
assert h.is_valid_snils("112-233-445 95")
|
||
assert not h.is_valid_snils("112-233-445 96")
|
||
rng = random.Random(1)
|
||
for _ in range(50):
|
||
s = h.synthetic_snils(rng)
|
||
assert s.startswith("000-") and h.is_valid_snils(s)
|
||
|
||
|
||
def test_ui_row_escapes_and_shows_values():
|
||
p = fixtures.PATIENTS[0].model_copy(update={"last_event": "<script>x</script>"})
|
||
html = ui.patient_row(p, detailed=True)
|
||
assert "<script>" not in html and "<script>" in html
|
||
assert "Андрей С." in html and "день терапии 13" in html and "↑" in html
|
||
|
||
|
||
def test_home_fragments_escape_api_values():
|
||
payload = '<img src=x onerror="alert(1)">'
|
||
p = fixtures.PATIENTS[0].model_copy(
|
||
update={
|
||
"display_name": payload,
|
||
"drug": payload,
|
||
"last_event": payload,
|
||
"discontinuation_risk": _risk(70, "high", [payload]),
|
||
}
|
||
)
|
||
for html in (ui.patient_row(p), ui.patient_row(p, detailed=True), ui.workspace_header(payload, payload)):
|
||
assert "<img" not in html
|
||
assert "<img" in html
|
||
|
||
|
||
def test_scale_figure_marks_unscheduled_and_bands():
|
||
card = fixtures.andrey_card()
|
||
fig = charts.scale_figure(card.series, "phq9", "PHQ-9", card.prescriptions[0].start_date)
|
||
assert fig is not None
|
||
trace = fig.data[0]
|
||
assert list(trace.marker.symbol) == ["circle", "circle", "diamond-open"]
|
||
assert len(fig.layout.shapes) == 5 # severity bands from config/scales/phq9.yaml
|
||
assert charts.scale_figure([], "phq9", "PHQ-9") is None
|
||
|
||
|
||
def test_wellbeing_figure_small_multiples():
|
||
card = fixtures.andrey_card()
|
||
fig = charts.wellbeing_figure(card.series)
|
||
assert fig is not None and len(fig.data) == 3
|
||
assert charts.wellbeing_figure([SeriesPoint(day=date(2026, 1, 1), metric="phq9", value=3)]) is None
|
||
|
||
|
||
def test_page_discovery_orders_and_sections(tmp_path: Path):
|
||
for name in ("1_Пациенты.py", "1b_Подключить_пациента.py", "2_Демо-консоль.py", "3_Сплит-экран.py", "helper.py"):
|
||
(tmp_path / name).write_text("", encoding="utf-8")
|
||
(tmp_path / "4_Что-то.py").write_text('NAV_ICON = ":material/star:"\nNAV_SECTION = "Прочее"\n', encoding="utf-8")
|
||
specs = discover_pages(tmp_path)
|
||
assert [s.title for s in specs] == ["Пациенты", "Подключить пациента", "Демо-консоль", "Сплит-экран", "Что-то"]
|
||
assert [s.section for s in specs][:4] == ["Кабинет врача", "Кабинет врача", "Демо", "Демо"]
|
||
assert specs[4].icon == ":material/star:" and specs[4].section == "Прочее"
|
||
assert specs[2].url_path == "demo-konsol"
|
||
assert slugify("Сплит-экран") == "split-ekran"
|
||
|
||
|
||
def test_real_pages_dir_has_doctor_pages():
|
||
titles = [s.title for s in discover_pages()]
|
||
assert titles[:2] == ["Пациенты", "Подключить пациента"]
|