475 lines
17 KiB
Python
475 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
from PySide6.QtCore import QSettings
|
|
from PySide6.QtWidgets import QApplication, QDialogButtonBox, QMessageBox
|
|
|
|
from doctor_workstation import app as app_module
|
|
from doctor_workstation.app import ApplicationController
|
|
from doctor_workstation.config import AppConfig
|
|
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 (
|
|
friendly_error,
|
|
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",),
|
|
"appointments": ("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()
|
|
|
|
|
|
def test_remembered_password_survives_a_new_login_window(tmp_path: Any) -> None:
|
|
application = QApplication.instance() or QApplication([])
|
|
settings_path = tmp_path / "remember-password.ini"
|
|
secrets: dict[tuple[str, str], str] = {}
|
|
|
|
def load_password(*, account: str, scope: str) -> str | None:
|
|
return secrets.get((account, scope.rstrip("/")))
|
|
|
|
def save_password(password: str, *, account: str, scope: str) -> bool:
|
|
secrets[(account, scope.rstrip("/"))] = password
|
|
return True
|
|
|
|
def clear_password(*, account: str, scope: str) -> None:
|
|
secrets.pop((account, scope.rstrip("/")), None)
|
|
|
|
credentials = SimpleNamespace(
|
|
load_password=load_password,
|
|
save_password=save_password,
|
|
clear_password=clear_password,
|
|
)
|
|
config = SimpleNamespace(
|
|
api_base_url="https://example.test/adminapi",
|
|
request_timeout=30,
|
|
verify_ssl=True,
|
|
demo_mode=False,
|
|
remembered_account="",
|
|
)
|
|
first = LoginWindow(
|
|
object(),
|
|
config=config,
|
|
settings=QSettings(str(settings_path), QSettings.Format.IniFormat),
|
|
credential_store=credentials,
|
|
)
|
|
first._on_login_success({}, "admin", True, "secret-value")
|
|
first.close()
|
|
application.processEvents()
|
|
|
|
restored = LoginWindow(
|
|
object(),
|
|
config=config,
|
|
settings=QSettings(str(settings_path), QSettings.Format.IniFormat),
|
|
credential_store=credentials,
|
|
)
|
|
assert restored.account_edit.text() == "admin"
|
|
assert restored.password_edit.text() == "secret-value"
|
|
assert restored.remember_check.text() == "记住密码"
|
|
assert restored.remember_check.isChecked()
|
|
|
|
restored._on_login_success({}, "admin", False, "secret-value")
|
|
restored.close()
|
|
application.processEvents()
|
|
|
|
forgotten = LoginWindow(
|
|
object(),
|
|
config=config,
|
|
settings=QSettings(str(settings_path), QSettings.Format.IniFormat),
|
|
credential_store=credentials,
|
|
)
|
|
assert forgotten.password_edit.text() == ""
|
|
assert not forgotten.remember_check.isChecked()
|
|
forgotten.close()
|
|
application.processEvents()
|
|
|
|
|
|
def test_server_settings_panel_keeps_controls_separated_at_minimum_window(
|
|
tmp_path: Any,
|
|
) -> None:
|
|
application = QApplication.instance() or QApplication([])
|
|
settings = QSettings(str(tmp_path / "server-panel.ini"), QSettings.Format.IniFormat)
|
|
config = SimpleNamespace(
|
|
api_base_url="",
|
|
request_timeout=30,
|
|
demo_mode=False,
|
|
remembered_account="",
|
|
)
|
|
window = LoginWindow(object(), config=config, settings=settings)
|
|
window.resize(860, 590)
|
|
window.show()
|
|
window.server_toggle.setChecked(True)
|
|
window._toggle_server_panel(True)
|
|
application.processEvents()
|
|
|
|
assert window.server_panel.isVisible()
|
|
assert window.server_panel.height() >= window.server_panel.minimumSizeHint().height()
|
|
assert window.server_url_label.geometry().bottom() < window.server_url_edit.geometry().top()
|
|
assert window.server_url_edit.geometry().bottom() < window.timeout_label.geometry().top()
|
|
assert window.timeout_spin.geometry().right() < window.save_server_button.geometry().left()
|
|
assert window.timeout_label.geometry().bottom() < window.allow_self_signed_check.geometry().top()
|
|
assert window.allow_self_signed_check.geometry().bottom() < window.server_hint.geometry().top()
|
|
|
|
window.allow_self_signed_check.setChecked(True)
|
|
application.processEvents()
|
|
assert window.ssl_warning.isVisible()
|
|
assert window.allow_self_signed_check.geometry().bottom() < window.ssl_warning.geometry().top()
|
|
assert window.ssl_warning.geometry().bottom() < window.server_hint.geometry().top()
|
|
|
|
window.close()
|
|
application.processEvents()
|
|
|
|
|
|
def test_server_settings_can_persist_self_signed_debug_mode(tmp_path: Any) -> None:
|
|
application = QApplication.instance() or QApplication([])
|
|
settings = QSettings(str(tmp_path / "self-signed.ini"), QSettings.Format.IniFormat)
|
|
config = SimpleNamespace(
|
|
api_base_url="https://internal.example.test",
|
|
request_timeout=30,
|
|
verify_ssl=True,
|
|
demo_mode=False,
|
|
remembered_account="",
|
|
)
|
|
window = LoginWindow(object(), config=config, settings=settings)
|
|
emitted: list[dict[str, Any]] = []
|
|
window.server_settings_changed.connect(emitted.append)
|
|
window.allow_self_signed_check.setChecked(True)
|
|
|
|
window._save_server_settings()
|
|
|
|
assert emitted[-1]["verify_ssl"] is False
|
|
assert settings.value("server/verify_ssl", type=bool) is False
|
|
assert "证书校验已关闭" in window.error_banner.label.text()
|
|
window.close()
|
|
application.processEvents()
|
|
|
|
|
|
def test_login_applies_self_signed_setting_before_authentication(
|
|
monkeypatch: Any,
|
|
tmp_path: Any,
|
|
) -> None:
|
|
application = QApplication.instance() or QApplication([])
|
|
settings = QSettings(str(tmp_path / "self-signed-login.ini"), QSettings.Format.IniFormat)
|
|
config = AppConfig(
|
|
api_base_url="https://internal.example.test/adminapi",
|
|
demo_mode=False,
|
|
verify_ssl=True,
|
|
)
|
|
calls: list[str] = []
|
|
|
|
class Repository:
|
|
def __init__(self, name: str) -> None:
|
|
self.name = name
|
|
|
|
def login(self, **_payload: Any) -> object:
|
|
calls.append(self.name)
|
|
return object()
|
|
|
|
def get_current_user(self) -> object:
|
|
return object()
|
|
|
|
old_repository = Repository("old")
|
|
rebuilt_repository = Repository("rebuilt-without-verification")
|
|
window = LoginWindow(old_repository, config=config, settings=settings)
|
|
|
|
def rebuild_on_config_change(updated: AppConfig) -> None:
|
|
assert updated.verify_ssl is False
|
|
window.repository = rebuilt_repository
|
|
window.active_repository = rebuilt_repository
|
|
|
|
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.config_changed.connect(rebuild_on_config_change)
|
|
window.account_edit.setText("admin")
|
|
window.password_edit.setText("secret")
|
|
window.allow_self_signed_check.setChecked(True)
|
|
|
|
window.submit()
|
|
|
|
assert calls == ["rebuilt-without-verification"]
|
|
assert settings.value("server/verify_ssl", type=bool) is False
|
|
window.close()
|
|
application.processEvents()
|
|
|
|
|
|
def test_certificate_error_explains_self_signed_server_setting() -> None:
|
|
message = friendly_error(
|
|
RuntimeError("[SSL: CERTIFICATE_VERIFY_FAILED] self-signed certificate")
|
|
)
|
|
assert "信任自签名证书" in message
|
|
assert "服务器设置" in message
|
|
|
|
|
|
def test_qt_standard_dialog_buttons_are_localized_to_chinese() -> None:
|
|
application = QApplication.instance() or QApplication([])
|
|
app_module._install_chinese_translations(application)
|
|
|
|
question = QMessageBox()
|
|
question.setStandardButtons(
|
|
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.Cancel
|
|
)
|
|
assert question.button(QMessageBox.StandardButton.Yes).text() == "是"
|
|
assert question.button(QMessageBox.StandardButton.Cancel).text() == "取消"
|
|
|
|
buttons = QDialogButtonBox(
|
|
QDialogButtonBox.StandardButton.Ok
|
|
| QDialogButtonBox.StandardButton.Save
|
|
| QDialogButtonBox.StandardButton.Close
|
|
)
|
|
assert buttons.button(QDialogButtonBox.StandardButton.Ok).text() == "确定"
|
|
assert buttons.button(QDialogButtonBox.StandardButton.Save).text() == "保存"
|
|
assert buttons.button(QDialogButtonBox.StandardButton.Close).text() == "关闭"
|
|
|
|
|
|
def test_friendly_error_hides_english_technical_messages() -> None:
|
|
message = friendly_error(
|
|
TypeError("invoke() takes 2 positional arguments but 3 were given")
|
|
)
|
|
assert message == "程序执行失败，请重试；若问题持续出现，请联系管理员。"
|
|
assert friendly_error(RuntimeError("API response envelope must be an object")) == (
|
|
"服务器返回的数据格式不正确，请联系管理员检查接口。"
|
|
)
|
|
|
|
|
|
def test_certificate_error_opens_server_settings(tmp_path: Any) -> None:
|
|
application = QApplication.instance() or QApplication([])
|
|
settings = QSettings(str(tmp_path / "certificate-error.ini"), QSettings.Format.IniFormat)
|
|
config = SimpleNamespace(
|
|
api_base_url="https://internal.example.test",
|
|
request_timeout=30,
|
|
verify_ssl=True,
|
|
demo_mode=False,
|
|
remembered_account="",
|
|
)
|
|
window = LoginWindow(object(), config=config, settings=settings)
|
|
|
|
window._on_login_error(
|
|
RuntimeError("[SSL: CERTIFICATE_VERIFY_FAILED] self-signed certificate")
|
|
)
|
|
|
|
assert window.server_toggle.isChecked()
|
|
assert not window.server_panel.isHidden()
|
|
assert "信任自签名证书" in window.error_banner.label.text()
|
|
window.close()
|
|
application.processEvents()
|