"""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()