241 lines
8.9 KiB
Python
241 lines
8.9 KiB
Python
"""Critical-path request counts and late metadata safety, without global Qt styling."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from typing import Any
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
import pytest
|
|
from PySide6.QtWidgets import QApplication
|
|
|
|
from doctor_workstation.core import PermissionSet
|
|
from doctor_workstation.services.repository import RemoteDoctorRepository
|
|
from doctor_workstation.ui.dialogs import diagnosis as module
|
|
|
|
|
|
class Client:
|
|
def __init__(self, delay: float = 0) -> None:
|
|
self.calls: list[tuple[str, dict[str, Any]]] = []
|
|
self.delay = delay
|
|
|
|
def get(self, endpoint: str, params: dict[str, Any]) -> Any:
|
|
self.calls.append((endpoint, params))
|
|
time.sleep(self.delay)
|
|
if endpoint == "config/dict":
|
|
return {
|
|
key: [{"name": "口干", "value": "dry"}, {"name": "口苦", "value": "bitter"}]
|
|
for key in params["type"].split(",")
|
|
}
|
|
if endpoint.endswith("Detail") or endpoint.endswith("/detail"):
|
|
return {
|
|
"id": params["id"], "patient_id": params["id"] + 1000,
|
|
"patient_name": f"Patient {params['id']}", "appetite": ["dry"],
|
|
"diagnosis_type": "first_visit", "chief_complaint": "Original",
|
|
}
|
|
return {"lists": [], "count": 0}
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def application() -> QApplication:
|
|
return QApplication.instance() or QApplication([])
|
|
|
|
|
|
@pytest.fixture
|
|
def queued(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]:
|
|
jobs: list[dict[str, Any]] = []
|
|
|
|
def enqueue(function: Any, **callbacks: Any) -> object:
|
|
jobs.append({"function": function, **callbacks})
|
|
return object()
|
|
|
|
monkeypatch.setattr(module, "run_async", enqueue)
|
|
return jobs
|
|
|
|
|
|
def complete(job: dict[str, Any]) -> None:
|
|
job["on_success"](job["function"]())
|
|
if job.get("on_finished"):
|
|
job["on_finished"]()
|
|
|
|
|
|
def dialog_for(client: Client) -> module.DiagnosisDialog:
|
|
return module.DiagnosisDialog(
|
|
RemoteDoctorRepository(client), permissions=PermissionSet(["*"])
|
|
)
|
|
|
|
|
|
def test_detail_unlocks_before_single_batch_and_preserves_draft(
|
|
application: QApplication, queued: list[dict[str, Any]]
|
|
) -> None:
|
|
client = Client()
|
|
dialog = dialog_for(client)
|
|
try:
|
|
dialog.open_for(501, editable=True, seed={"patient_name": "Unverified"})
|
|
assert dialog.isVisible()
|
|
assert dialog.drawer_loading.isVisibleTo(dialog)
|
|
assert dialog.edit_fields["patient_name"].isReadOnly()
|
|
assert not dialog.save_button.isEnabled()
|
|
assert len(queued) == 1
|
|
complete(queued[0])
|
|
assert client.calls == [("tcm.diagnosis/detail", {"id": 501})]
|
|
assert dialog._authoritative_detail_loaded
|
|
assert dialog.save_button.isEnabled()
|
|
assert not dialog.drawer_loading.isVisibleTo(dialog)
|
|
assert len(queued) == 2
|
|
dialog.edit_fields["chief_complaint"].setPlainText("Unsaved draft")
|
|
dialog.edit_fields["appetite"].setPlainText("bitter")
|
|
complete(queued[1])
|
|
assert len(client.calls) == 2
|
|
assert client.calls[1][0] == "config/dict"
|
|
assert len(client.calls[1][1]["type"].split(",")) == 16
|
|
assert dialog.edit_fields["appetite"].toPlainText() == "bitter"
|
|
assert dialog.edit_fields["chief_complaint"].toPlainText() == "Unsaved draft"
|
|
assert dialog._loaded_tabs == set()
|
|
assert not any("getCallRecords" in endpoint for endpoint, _params in client.calls)
|
|
dialog._ensure_tab_loaded("basic")
|
|
dialog.refresh_permissions()
|
|
assert len(queued) == 2
|
|
assert len(client.calls) == 2
|
|
finally:
|
|
dialog.close()
|
|
|
|
|
|
@pytest.mark.parametrize("close", [False, True])
|
|
def test_late_detail_and_dictionary_callbacks_cannot_touch_another_opening(
|
|
application: QApplication, queued: list[dict[str, Any]], close: bool
|
|
) -> None:
|
|
dialog = dialog_for(Client())
|
|
try:
|
|
dialog.open_for(501, editable=True)
|
|
detail_job = queued[0]
|
|
complete(detail_job)
|
|
metadata_job = queued[1]
|
|
if close:
|
|
dialog.close()
|
|
else:
|
|
dialog.open_for(502, editable=True)
|
|
complete(queued[2])
|
|
dialog.edit_fields["chief_complaint"].setPlainText("Keep current draft")
|
|
before = dialog.edit_fields["patient_name"].toPlainText()
|
|
complete(detail_job)
|
|
complete(metadata_job)
|
|
assert dialog.edit_fields["patient_name"].toPlainText() == before
|
|
assert dialog.edit_fields["chief_complaint"].toPlainText() == "Keep current draft"
|
|
if not close:
|
|
assert before == "Patient 502"
|
|
assert dialog._patient_id == 1502
|
|
finally:
|
|
dialog.close()
|
|
|
|
|
|
def test_dictionary_failure_does_not_relock_valid_detail_or_erase_choices(
|
|
application: QApplication, queued: list[dict[str, Any]]
|
|
) -> None:
|
|
dialog = dialog_for(Client())
|
|
try:
|
|
dialog.open_for(501, editable=True)
|
|
complete(queued[0])
|
|
queued[1]["on_error"](RuntimeError("Delayed metadata unavailable"))
|
|
assert dialog._authoritative_detail_loaded
|
|
assert dialog.save_button.isEnabled()
|
|
assert dialog.edit_fields["appetite"].toPlainText() == "dry"
|
|
assert not dialog.drawer_loading.isVisibleTo(dialog)
|
|
dialog._retry_current()
|
|
assert len(queued) == 3
|
|
complete(queued[2])
|
|
finally:
|
|
dialog.close()
|
|
|
|
|
|
def test_inactive_basic_tab_defers_dictionary_request_until_selected(
|
|
application: QApplication, queued: list[dict[str, Any]]
|
|
) -> None:
|
|
client = Client()
|
|
dialog = dialog_for(client)
|
|
try:
|
|
notes_index = next(
|
|
i for i in range(dialog.tabs.count()) if dialog.tabs.tabBar().tabData(i) == "notes"
|
|
)
|
|
dialog.tabs.setCurrentIndex(notes_index)
|
|
dialog.open_for(501, editable=True)
|
|
complete(queued[0])
|
|
assert dialog._current_tab_key() == "notes"
|
|
assert dialog._dictionary_requested_generation != dialog._generation
|
|
assert len(client.calls) == 1
|
|
# The only secondary request is for the active notes tab, not all tabs.
|
|
assert len(queued) == 2
|
|
dialog.tabs.setCurrentIndex(0)
|
|
assert len(queued) == 3
|
|
complete(queued[2])
|
|
assert client.calls[-1][0] == "config/dict"
|
|
dialog.tabs.setCurrentIndex(notes_index)
|
|
dialog.tabs.setCurrentIndex(0)
|
|
assert len(queued) == 3
|
|
finally:
|
|
dialog.close()
|
|
|
|
|
|
@pytest.mark.parametrize("retry_from_banner", [False, True])
|
|
def test_dictionary_error_survives_other_tab_success_and_retries_correct_request(
|
|
application: QApplication, queued: list[dict[str, Any]], retry_from_banner: bool,
|
|
) -> None:
|
|
client = Client()
|
|
dialog = dialog_for(client)
|
|
try:
|
|
dialog.open_for(501, editable=True)
|
|
complete(queued[0])
|
|
metadata_job = queued[1]
|
|
notes_index = next(i for i in range(dialog.tabs.count())
|
|
if dialog.tabs.tabBar().tabData(i) == "notes")
|
|
dialog.tabs.setCurrentIndex(notes_index)
|
|
notes_job = queued[2]
|
|
metadata_job["on_error"](RuntimeError("Metadata unavailable"))
|
|
complete(notes_job)
|
|
assert dialog.drawer_banner.label.text() == dialog._dictionary_error_message
|
|
assert dialog.drawer_banner.label.text().startswith("病历选项加载失败")
|
|
dialog.edit_fields["chief_complaint"].setPlainText("Keep draft across retry")
|
|
if retry_from_banner:
|
|
dialog.drawer_banner.action_requested.emit()
|
|
else:
|
|
dialog.tabs.setCurrentIndex(0)
|
|
assert len(queued) == 4
|
|
complete(queued[3])
|
|
assert client.calls[-1][0] == "config/dict"
|
|
assert not dialog._dictionary_error_message
|
|
assert dialog.edit_fields["chief_complaint"].toPlainText() == "Keep draft across retry"
|
|
assert dialog._authoritative_detail_loaded
|
|
finally:
|
|
dialog.close()
|
|
|
|
|
|
def test_controlled_latency_removes_sixteen_round_trips_from_detail_gate(
|
|
application: QApplication,
|
|
) -> None:
|
|
client = Client(delay=0.02)
|
|
dialog = dialog_for(client)
|
|
try:
|
|
types = sorted({"diagnosis_type", *(row[0] for row in module._CHOICE_DICTIONARIES.values())})
|
|
started = time.perf_counter()
|
|
dialog.repository.get_diagnosis_detail(501)
|
|
for key in types:
|
|
dialog.repository.get_dictionary(key)
|
|
old_time = time.perf_counter() - started
|
|
old_calls = len(client.calls)
|
|
client.calls.clear()
|
|
started = time.perf_counter()
|
|
detail = dialog._load_bundle(501, "edit")
|
|
new_time = time.perf_counter() - started
|
|
assert detail["detail"]["id"] == 501
|
|
assert len(client.calls) == 1
|
|
dialog._load_dictionary_choices()
|
|
assert len(client.calls) == 2
|
|
assert old_calls == 17
|
|
assert new_time < old_time / 4
|
|
print(f"controlled 20 ms RTT: old gate={old_time:.3f}s/17 calls; "
|
|
f"new gate={new_time:.3f}s/1 call; total new=2 calls")
|
|
finally:
|
|
dialog.close()
|