229 lines
10 KiB
Python
229 lines
10 KiB
Python
"""Streamlit AppTest smoke tests of the doctor pages against an in-memory API (fixtures over httpx.MockTransport)."""
|
||
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
import pytest
|
||
from streamlit.testing.v1 import AppTest
|
||
|
||
from app.api import fixtures
|
||
from dashboard.api_client import ApiClient
|
||
from dashboard.components import state
|
||
|
||
PAGES = Path(__file__).resolve().parents[1] / "dashboard" / "pages"
|
||
LIST_PAGE = str(PAGES / "1_Пациенты.py")
|
||
CONNECT_PAGE = str(PAGES / "1b_Подключить_пациента.py")
|
||
|
||
|
||
def _dump(obj):
|
||
return [o.model_dump(mode="json") for o in obj] if isinstance(obj, list) else obj.model_dump(mode="json")
|
||
|
||
|
||
class FakeApi:
|
||
def __init__(self) -> None:
|
||
self.calls: list[tuple[str, str]] = []
|
||
self.extra_alerts: list = []
|
||
self.patients = list(fixtures.PATIENTS)
|
||
|
||
def __call__(self, req: httpx.Request) -> httpx.Response:
|
||
path = req.url.path.removeprefix("/api/v1")
|
||
self.calls.append((req.method, path))
|
||
if path == "/doctors/doc-smirnova/patients":
|
||
return httpx.Response(200, json=_dump(self.patients))
|
||
if path.startswith("/patients/tok-"):
|
||
card = fixtures.card(path.rsplit("/", 1)[1])
|
||
return httpx.Response(200, json=_dump(card)) if card else httpx.Response(404)
|
||
if path == "/doctors/doc-smirnova/alerts":
|
||
return httpx.Response(200, json=_dump(self.extra_alerts + fixtures.all_alerts()))
|
||
if path.startswith("/alerts/") and path.endswith("/ack"):
|
||
alert = next(a for a in fixtures.all_alerts() if a.id == int(path.split("/")[2]))
|
||
acked = alert.model_copy(update={"acked_by": "doc-smirnova", "acked_at": alert.ts})
|
||
return httpx.Response(200, json=_dump(acked))
|
||
if path == "/patients/connect":
|
||
return httpx.Response(
|
||
200, json={"token": "tok-new", "code": "DEMO-0001", "deep_link": "https://t.me/b?start=DEMO-0001"}
|
||
)
|
||
if path == "/demo/clock":
|
||
return httpx.Response(200, json={"now": "2026-10-03T21:05:00+03:00", "virtual": True, "offset_days": 0})
|
||
return httpx.Response(404)
|
||
|
||
|
||
@pytest.fixture
|
||
def api(monkeypatch) -> FakeApi:
|
||
fake = FakeApi()
|
||
client = ApiClient("http://api.test", "doc-smirnova", transport=httpx.MockTransport(fake))
|
||
monkeypatch.setattr(state, "client", lambda: client)
|
||
return fake
|
||
|
||
|
||
def _html(at: AppTest) -> str:
|
||
return "\n".join(str(el.proto.body) for el in at.get("html"))
|
||
|
||
|
||
def test_list_page_renders_sorted_patients(api):
|
||
at = AppTest.from_file(LIST_PAGE, default_timeout=20).run()
|
||
assert not at.exception
|
||
html = _html(at)
|
||
assert "Кабинет д-ра Смирновой Е. А." in html
|
||
assert "Пациенты под наблюдением" in html
|
||
assert html.index("Андрей С.") < html.index("Сергей П.") < html.index("Павел Р.")
|
||
assert "Высокий · 70" in html # discontinuation-risk badge
|
||
open_buttons = [b for b in at.button if b.label == "Открыть"]
|
||
assert len(open_buttons) == len(fixtures.PATIENTS)
|
||
|
||
|
||
def test_list_filter_and_open_card(api):
|
||
at = AppTest.from_file(LIST_PAGE, default_timeout=20).run()
|
||
at.segmented_control(key="status_filter").set_value("Срочно").run()
|
||
at.text_input(key="patient_query").set_value("андр").run()
|
||
at.toggle(key="patient_details").set_value(True).run()
|
||
assert len([b for b in at.button if b.label == "Открыть"]) == 1
|
||
at.button(key="open-tok-andrey").click().run()
|
||
assert not at.exception
|
||
assert at.query_params["patient"] in ("tok-andrey", ["tok-andrey"])
|
||
html = _html(at)
|
||
assert "PHQ-9 п.9 = 2" in html and "План безопасности" in html
|
||
assert "Решения принимает врач" in html
|
||
assert ("GET", "/patients/tok-andrey") in api.calls
|
||
next(b for b in at.button if b.label == "Все пациенты").click().run()
|
||
assert not at.exception
|
||
assert not at.query_params.get("patient")
|
||
assert len([b for b in at.button if b.label == "Открыть"]) == 1 # the selected filter is retained
|
||
assert at.text_input(key="patient_query").value == "андр"
|
||
assert at.toggle(key="patient_details").value is True
|
||
|
||
|
||
@pytest.mark.parametrize("choice,expected", [("Внимание", 4), ("Без уведомлений", 3), ("Все пациенты", 8)])
|
||
def test_home_status_filters(api, choice, expected):
|
||
at = AppTest.from_file(LIST_PAGE, default_timeout=20).run()
|
||
at.segmented_control(key="status_filter").set_value(choice).run()
|
||
assert not at.exception
|
||
assert len([b for b in at.button if b.label == "Открыть"]) == expected
|
||
assert f"Показано {expected} из 8 пациентов" in _html(at)
|
||
|
||
|
||
def test_home_search_empty_result_and_reset(api):
|
||
at = AppTest.from_file(LIST_PAGE, default_timeout=20).run()
|
||
at.text_input(key="patient_query").set_value(" ПАРОКСЕТИН ").run()
|
||
assert len([b for b in at.button if b.label == "Открыть"]) == 1
|
||
assert "Сергей П." in _html(at)
|
||
at.segmented_control(key="status_filter").set_value("Без уведомлений").run()
|
||
assert "Пациенты не найдены" in _html(at)
|
||
next(b for b in at.button if b.label == "Сбросить фильтры").click().run()
|
||
assert not at.exception
|
||
assert len([b for b in at.button if b.label == "Открыть"]) == 8
|
||
assert at.text_input(key="patient_query").value == ""
|
||
|
||
|
||
def test_home_empty_cabinet_keeps_connect_action(api):
|
||
api.patients = []
|
||
at = AppTest.from_file(LIST_PAGE, default_timeout=20).run()
|
||
assert not at.exception
|
||
assert "Пациентов пока нет" in _html(at)
|
||
assert "Показано 0 из 0 пациентов" in _html(at)
|
||
assert any(b.label == "Подключить пациента" for b in at.button)
|
||
assert not any(b.label == "Открыть" for b in at.button)
|
||
|
||
|
||
def test_home_details_preserve_clinical_information(api):
|
||
at = AppTest.from_file(LIST_PAGE, default_timeout=20).run()
|
||
assert "PHQ-9" in _html(at) and "PCL-5" in _html(at)
|
||
at.toggle(key="patient_details").set_value(True).run()
|
||
assert not at.exception
|
||
html = _html(at)
|
||
assert "сертралин 50 мг" in html
|
||
assert "PHQ-9 п.9 = 2 — кризисный протокол" in html
|
||
assert "пропуски 3 из 7" in html
|
||
|
||
|
||
def test_home_info_only_patient_and_missing_scores(api):
|
||
api.patients = [
|
||
fixtures.PATIENTS[0].model_copy(
|
||
update={
|
||
"status_color": "grey",
|
||
"top_alert_level": "info",
|
||
"phq9_last": None,
|
||
"pcl5_last": None,
|
||
"adherence_7d": None,
|
||
"open_alerts": 1,
|
||
}
|
||
)
|
||
]
|
||
at = AppTest.from_file(LIST_PAGE, default_timeout=20).run()
|
||
at.segmented_control(key="status_filter").set_value("Нет ответа").run()
|
||
assert not at.exception
|
||
assert len([b for b in at.button if b.label == "Открыть"]) == 1
|
||
assert "Из них только инфо: 1" in _html(at)
|
||
assert 'mt-dim">—' in _html(at)
|
||
|
||
|
||
def test_card_ack_posts_and_marks_alert(api):
|
||
at = AppTest.from_file(LIST_PAGE, default_timeout=20)
|
||
at.query_params["patient"] = "tok-andrey"
|
||
at.run()
|
||
assert not at.exception
|
||
assert at.button(key="ack-3").label == "Подтвердить"
|
||
at.button(key="ack-3").click().run()
|
||
assert ("POST", "/alerts/3/ack") in api.calls
|
||
assert not [b for b in at.button if b.key == "ack-3"] # acked alert has no button any more
|
||
|
||
|
||
def test_api_down_shows_banner(monkeypatch):
|
||
def refuse(req: httpx.Request) -> httpx.Response:
|
||
raise httpx.ConnectError("refused", request=req)
|
||
|
||
client = ApiClient("http://api.test", transport=httpx.MockTransport(refuse))
|
||
monkeypatch.setattr(state, "client", lambda: client)
|
||
at = AppTest.from_file(LIST_PAGE, default_timeout=20).run()
|
||
assert not at.exception
|
||
assert "API недоступен" in _html(at)
|
||
assert any(b.label == "Повторить загрузку" for b in at.button)
|
||
assert "Всего пациентов" not in _html(at) # failed loading is not a zero-patient cabinet
|
||
|
||
|
||
def test_connect_flow_shows_code(api):
|
||
at = AppTest.from_file(CONNECT_PAGE, default_timeout=20).run()
|
||
assert not at.exception
|
||
at.button[0].click().run() # «Заполнить демо-данными»
|
||
at.checkbox(key="cp_consent").check().run()
|
||
assert at.text_input(key="cp_name").value == "Андрей С."
|
||
submit = next(b for b in at.button if b.label == "Подключить и получить код")
|
||
submit.click().run()
|
||
assert not at.exception
|
||
assert ("POST", "/patients/connect") in api.calls
|
||
assert "DEMO-0001" in _html(at)
|
||
assert at.get("image") # QR image
|
||
|
||
|
||
def test_connect_requires_consent(api):
|
||
at = AppTest.from_file(CONNECT_PAGE, default_timeout=20).run()
|
||
at.text_input(key="cp_name").set_value("Пётр К.").run()
|
||
next(b for b in at.button if b.label == "Подключить и получить код").click().run()
|
||
assert at.error and "согласие" in at.error[0].value
|
||
assert ("POST", "/patients/connect") not in api.calls
|
||
|
||
|
||
def test_app_entry_navigation_and_acute_banner(api):
|
||
at = AppTest.from_file(str(PAGES.parent / "app.py"), default_timeout=20).run()
|
||
assert not at.exception
|
||
html = _html(at)
|
||
assert "Кризисный сигнал · Андрей С." in html # open acute alert pinned above every page
|
||
assert "Кабинет д-ра Смирновой Е. А." in html # default page = patient list
|
||
assert "Прототип на синтетических данных" in html
|
||
assert ("GET", "/doctors/doc-smirnova/alerts") in api.calls
|
||
|
||
|
||
def test_new_alert_toasts_once(api):
|
||
at = AppTest.from_file(str(PAGES.parent / "app.py"), default_timeout=20).run()
|
||
assert not at.toast # alerts open at session start are not "new"
|
||
new = fixtures.all_alerts()[0].model_copy(
|
||
update={"id": 10, "token": "tok-sergey", "patient_name": "Сергей П.", "reason": "C-SSRS: намерение"}
|
||
)
|
||
api.extra_alerts.append(new)
|
||
at.run()
|
||
assert not at.exception
|
||
assert any("Сергей П." in t.value for t in at.toast)
|
||
assert "Кризисный сигнал · Сергей П." in _html(at)
|
||
at.run()
|
||
assert not any("Сергей П." in t.value for t in at.toast)
|