230 lines
7.7 KiB
Python
230 lines
7.7 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
from PySide6.QtCore import QSettings
|
|
from PySide6.QtWidgets import QApplication
|
|
|
|
from doctor_workstation import app as app_module
|
|
from doctor_workstation.app import ApplicationController
|
|
from doctor_workstation.core.errors import AuthenticationExpiredError
|
|
from doctor_workstation.services import DemoDoctorRepository
|
|
from doctor_workstation.ui import login as login_module
|
|
from doctor_workstation.ui import widgets as widget_module
|
|
from doctor_workstation.ui.login import LoginWindow
|
|
from doctor_workstation.ui.pages.consultations import _video_payload
|
|
from doctor_workstation.ui.shell import NAVIGATION
|
|
from doctor_workstation.ui.widgets import (
|
|
gender_text,
|
|
invoke,
|
|
set_authentication_expired_handler,
|
|
)
|
|
|
|
|
|
def test_consultation_video_payload_keeps_appointment_and_diagnosis_ids_distinct() -> None:
|
|
record = SimpleNamespace(
|
|
id=501,
|
|
appointment_id=101,
|
|
patient_id=301,
|
|
patient_name="林晓岚",
|
|
)
|
|
|
|
payload = _video_payload(record)
|
|
|
|
assert payload["appointment_id"] == 101
|
|
assert payload["diagnosis_id"] == 501
|
|
assert payload["patient_id"] == 301
|
|
|
|
|
|
def test_gender_text_maps_legacy_codes_and_preserves_labels() -> None:
|
|
assert gender_text(1) == "男"
|
|
assert gender_text("2") == "女"
|
|
assert gender_text(0) == "未知"
|
|
assert gender_text("未知标签") == "未知标签"
|
|
|
|
|
|
def test_navigation_requires_each_pages_actual_list_capability() -> None:
|
|
assert {item.key: item.permissions for item in NAVIGATION} == {
|
|
"reception": ("doctor.appointment/lists",),
|
|
"prescription_library": ("tcm.prescriptionLibrary/lists",),
|
|
"prescriptions": ("tcm.prescription/lists",),
|
|
"patients": ("firstvisit.myPatient/lists",),
|
|
"consultations": ("tcm.diagnosis/lists",),
|
|
}
|
|
|
|
|
|
def test_video_release_does_not_remove_a_newer_call() -> None:
|
|
older = object()
|
|
newer = object()
|
|
controller = SimpleNamespace(video_calls={"501": newer})
|
|
|
|
ApplicationController._release_video_call(controller, "501", older)
|
|
assert controller.video_calls == {"501": newer}
|
|
|
|
ApplicationController._release_video_call(controller, "501", newer)
|
|
assert controller.video_calls == {}
|
|
|
|
|
|
def test_invoke_leaves_prescription_filters_for_repository_mapping() -> None:
|
|
"""The UI adapter only normalises pagination, not API-specific DTO fields."""
|
|
|
|
class Repository:
|
|
def list_prescriptions(self, **filters: Any) -> dict[str, Any]:
|
|
return filters
|
|
|
|
assert invoke(
|
|
Repository(),
|
|
"prescriptions",
|
|
keyword="CF-2026-8",
|
|
status=2,
|
|
page=3,
|
|
page_size=15,
|
|
) == {
|
|
"keyword": "CF-2026-8",
|
|
"status": 2,
|
|
"page_no": 3,
|
|
"page_size": 15,
|
|
}
|
|
|
|
|
|
def test_async_error_dispatch_consumes_active_session_expiry_globally() -> None:
|
|
"""A consumed authentication expiry does not also reach a stale page."""
|
|
|
|
global_errors: list[Exception] = []
|
|
local_errors: list[Exception] = []
|
|
error = AuthenticationExpiredError("expired", code=-1)
|
|
try:
|
|
set_authentication_expired_handler(lambda caught: global_errors.append(caught) is None)
|
|
widget_module._dispatch_async_error(error, local_errors.append)
|
|
finally:
|
|
set_authentication_expired_handler(None)
|
|
|
|
assert global_errors == [error]
|
|
assert local_errors == []
|
|
|
|
|
|
def test_controller_session_expiry_returns_to_login_once() -> None:
|
|
"""The composition root owns the single transition out of an active shell."""
|
|
|
|
messages: list[str] = []
|
|
controller = SimpleNamespace(
|
|
shell_window=object(),
|
|
current_repository=object(),
|
|
_authentication_expiry_in_progress=False,
|
|
_logout=lambda *, message="": messages.append(message),
|
|
)
|
|
error = AuthenticationExpiredError("expired", code=-1)
|
|
|
|
assert ApplicationController._on_authentication_expired(controller, error)
|
|
assert ApplicationController._on_authentication_expired(controller, error)
|
|
assert messages == ["登录状态已失效，请重新登录。"]
|
|
|
|
|
|
def test_persisted_session_restore_blocks_manual_submit_before_worker_start(
|
|
monkeypatch: Any,
|
|
) -> None:
|
|
"""Login controls are locked before the asynchronous restore is dispatched."""
|
|
|
|
pending_states: list[bool] = []
|
|
worker_calls: list[tuple[Any, dict[str, Any]]] = []
|
|
worker = object()
|
|
repository = SimpleNamespace(restore_session=lambda: None)
|
|
login_window = SimpleNamespace(set_session_restore_pending=pending_states.append)
|
|
controller = SimpleNamespace(
|
|
remote_repository=repository,
|
|
_shutting_down=False,
|
|
config=SimpleNamespace(demo_mode=False),
|
|
current_repository=None,
|
|
_restore_generation=0,
|
|
_restore_in_progress=False,
|
|
_restore_worker=None,
|
|
login_window=login_window,
|
|
_on_restore_success=lambda *args: None,
|
|
_on_restore_error=lambda *args: None,
|
|
_on_restore_finished=lambda *args: None,
|
|
)
|
|
|
|
def fake_run_async(function: Any, **callbacks: Any) -> object:
|
|
assert pending_states == [True]
|
|
worker_calls.append((function, callbacks))
|
|
return worker
|
|
|
|
monkeypatch.setattr(app_module, "run_async", fake_run_async)
|
|
ApplicationController._begin_session_restore(controller)
|
|
|
|
assert controller._restore_in_progress
|
|
assert controller._restore_worker is worker
|
|
assert worker_calls[0][0] == repository.restore_session
|
|
|
|
|
|
def test_login_restore_pending_uses_existing_submit_guard() -> None:
|
|
"""A manual submit is a no-op for the whole persisted-token validation window."""
|
|
|
|
class Banner:
|
|
def clear(self) -> None:
|
|
pass
|
|
|
|
class LoginDouble:
|
|
def __init__(self) -> None:
|
|
self._loading = False
|
|
self.error_banner = Banner()
|
|
self.loading_options: dict[str, Any] = {}
|
|
|
|
def _set_loading(self, loading: bool, **options: Any) -> None:
|
|
self._loading = loading
|
|
self.loading_options = options
|
|
|
|
login = LoginDouble()
|
|
LoginWindow.set_session_restore_pending(login, True) # type: ignore[arg-type]
|
|
LoginWindow.submit(login) # type: ignore[arg-type]
|
|
|
|
assert login._loading
|
|
assert login.loading_options["button_text"] == "正在恢复登录…"
|
|
|
|
|
|
def test_real_demo_login_reaches_success_without_widget_adapter(
|
|
monkeypatch: Any,
|
|
tmp_path: Any,
|
|
) -> None:
|
|
"""Exercise the actual login widgets and demo repository as one contract."""
|
|
|
|
application = QApplication.instance() or QApplication([])
|
|
repository = DemoDoctorRepository()
|
|
settings = QSettings(str(tmp_path / "login.ini"), QSettings.Format.IniFormat)
|
|
config = SimpleNamespace(
|
|
api_base_url="https://127.0.0.1:9",
|
|
request_timeout=30,
|
|
demo_mode=True,
|
|
remembered_account="",
|
|
)
|
|
payloads: list[dict[str, Any]] = []
|
|
|
|
def run_immediately(function: Any, **callbacks: Any) -> object:
|
|
try:
|
|
callbacks["on_success"](function())
|
|
except Exception as error: # pragma: no cover - assertion output is more useful
|
|
callbacks["on_error"](error)
|
|
finally:
|
|
callbacks["on_finished"]()
|
|
return object()
|
|
|
|
monkeypatch.setattr(login_module, "run_async", run_immediately)
|
|
window = LoginWindow(
|
|
object(),
|
|
config=config,
|
|
demo_repository=repository,
|
|
settings=settings,
|
|
)
|
|
window.login_succeeded.connect(payloads.append)
|
|
|
|
window.submit()
|
|
|
|
assert len(payloads) == 1
|
|
assert payloads[0]["repository"] is repository
|
|
assert payloads[0]["demo_mode"] is True
|
|
assert window.busy_overlay.label.text() == "正在验证账号…"
|
|
assert not window._loading
|
|
window.close()
|
|
application.processEvents()
|