216 lines
8.2 KiB
Python
216 lines
8.2 KiB
Python
"""Permission-scoped department loading and diagnosis HTTP filter contracts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
import pytest
|
|
from PySide6.QtCore import Signal
|
|
from PySide6.QtWidgets import QApplication, QWidget
|
|
|
|
from doctor_workstation.core.errors import ApiBusinessError, ApiProtocolError
|
|
from doctor_workstation.services.repository import RemoteDoctorRepository
|
|
from doctor_workstation.ui.pages import consultations as module
|
|
|
|
TREE = [
|
|
{"id": 10, "name": "医助部", "children": [
|
|
{"id": "11", "name": "一组", "children": [
|
|
{"id": 12, "name": "专病组", "children": []},
|
|
]},
|
|
]},
|
|
{"id": 20, "name": "另一部门", "children": []},
|
|
]
|
|
|
|
|
|
class Client:
|
|
def __init__(self, departments: Any = TREE) -> None:
|
|
self.departments = departments
|
|
self.calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any:
|
|
self.calls.append((endpoint, dict(params or {})))
|
|
if endpoint == "dept.dept/all":
|
|
if isinstance(self.departments, Exception):
|
|
raise self.departments
|
|
return self.departments
|
|
return {"lists": [], "count": 0}
|
|
|
|
|
|
class DetailStub(QWidget):
|
|
saved = Signal()
|
|
|
|
def __init__(self, _repository: Any, parent: QWidget | None = None) -> None:
|
|
super().__init__(parent)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def application() -> QApplication:
|
|
return QApplication.instance() or QApplication([])
|
|
|
|
|
|
@pytest.fixture
|
|
def page(application: QApplication, monkeypatch: pytest.MonkeyPatch):
|
|
monkeypatch.setattr(module, "DiagnosisDialog", DetailStub)
|
|
# Test only this page's controls; loading a global stylesheet is unrelated.
|
|
client = Client()
|
|
result = module.ConsultationsPage(
|
|
RemoteDoctorRepository(client),
|
|
current_user={"department_id": 99, "department_name": "个人所属部门"},
|
|
)
|
|
yield result, client
|
|
result.poll_timer.stop()
|
|
result.close()
|
|
result.deleteLater()
|
|
application.processEvents()
|
|
|
|
|
|
def immediate(function: Any, *, on_success: Any = None, on_error: Any = None,
|
|
on_finished: Any = None, **_kwargs: Any) -> None:
|
|
try:
|
|
value = function()
|
|
except Exception as error:
|
|
if on_error:
|
|
on_error(error)
|
|
else:
|
|
if on_success:
|
|
on_success(value)
|
|
finally:
|
|
if on_finished:
|
|
on_finished()
|
|
|
|
|
|
@pytest.mark.parametrize("payload", [TREE, {"lists": TREE}, {"tree": TREE}, {"data": {"lists": TREE}}])
|
|
def test_repository_scoped_tree_shapes(payload: Any) -> None:
|
|
client = Client(payload)
|
|
assert RemoteDoctorRepository(client).list_departments(apply_data_scope=True) == TREE
|
|
assert client.calls == [("dept.dept/all", {"apply_data_scope": 1})]
|
|
|
|
|
|
def test_repository_malformed_is_not_a_permission_empty_result() -> None:
|
|
with pytest.raises(ApiProtocolError):
|
|
RemoteDoctorRepository(Client({"unexpected": 1})).list_departments(apply_data_scope=True)
|
|
|
|
|
|
def test_repository_maps_legacy_department_key_and_preserves_canonical() -> None:
|
|
client = Client()
|
|
repository = RemoteDoctorRepository(client)
|
|
repository.list_consultations(department_id=12)
|
|
assert client.calls[-1][1]["assistant_dept_id"] == 12
|
|
assert "department_id" not in client.calls[-1][1]
|
|
repository.list_consultations(department_id=12, assistant_dept_id=20)
|
|
assert client.calls[-1][1]["assistant_dept_id"] == 20
|
|
|
|
|
|
def test_nested_options_search_and_selected_id_reach_list_and_counts(page: Any, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
widget, client = page
|
|
monkeypatch.setattr(module, "run_async", immediate)
|
|
widget._load_departments()
|
|
combo = widget.department_combo
|
|
assert combo.count() == 5
|
|
assert combo.itemText(3) == "医助部 / 一组 / 专病组"
|
|
assert combo.findData(99) == -1
|
|
assert combo.isEditable()
|
|
combo.completer().setCompletionPrefix("专病")
|
|
assert combo.completer().completionCount() == 1
|
|
combo.setCurrentIndex(combo.findData(12))
|
|
widget._search()
|
|
calls = [params for endpoint, params in client.calls if endpoint == "tcm.diagnosis/lists"]
|
|
assert calls
|
|
assert all(params["assistant_dept_id"] == 12 for params in calls)
|
|
assert all("department_id" not in params for params in calls)
|
|
assert calls[-1]["page_size"] == 15
|
|
|
|
|
|
def test_unselected_search_text_never_becomes_an_id_or_silently_searches_all(page: Any, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
widget, client = page
|
|
monkeypatch.setattr(module, "run_async", immediate)
|
|
widget._load_departments()
|
|
widget.department_combo.setEditText("不存在的部门")
|
|
widget._search()
|
|
assert not any(endpoint == "tcm.diagnosis/lists" for endpoint, _ in client.calls)
|
|
widget.department_combo.setEditText("")
|
|
widget._search()
|
|
assert widget._shared_filters()["assistant_dept_id"] == ""
|
|
|
|
|
|
def test_permission_empty_has_no_profile_fallback_and_can_retry(page: Any, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
widget, client = page
|
|
monkeypatch.setattr(module, "run_async", immediate)
|
|
client.departments = []
|
|
widget._load_departments()
|
|
assert widget.department_combo.count() == 1
|
|
assert widget.department_combo.currentData() == ""
|
|
assert widget.department_combo.currentText() == "暂无可选部门"
|
|
assert not widget.department_combo.isEnabled()
|
|
assert not widget.department_retry_button.isHidden()
|
|
client.departments = TREE
|
|
widget.department_retry_button.click()
|
|
assert widget.department_combo.isEnabled()
|
|
assert widget.department_combo.findData(12) > 0
|
|
assert all(params == {"apply_data_scope": 1} for _, params in client.calls)
|
|
|
|
|
|
def test_permission_error_stays_scoped_and_retry_recovers(page: Any, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
widget, client = page
|
|
monkeypatch.setattr(module, "run_async", immediate)
|
|
client.departments = ApiBusinessError("无权限访问部门")
|
|
widget._load_departments()
|
|
assert not widget._departments_loaded
|
|
assert not widget._departments_loading
|
|
assert widget.department_combo.currentText() == "部门加载失败"
|
|
assert "无权限" in widget.department_combo.toolTip()
|
|
assert not widget.department_retry_button.isHidden()
|
|
client.departments = TREE
|
|
widget.department_retry_button.click()
|
|
assert widget._departments_loaded
|
|
assert widget.department_retry_button.isHidden()
|
|
assert all(params == {"apply_data_scope": 1} for _, params in client.calls)
|
|
|
|
|
|
def test_inflight_dedup_and_stale_success_or_failure_cannot_override_selection(page: Any, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
widget, _client = page
|
|
jobs: list[dict[str, Any]] = []
|
|
monkeypatch.setattr(module, "run_async", lambda function, **callbacks: jobs.append(callbacks))
|
|
widget._load_departments()
|
|
widget._load_departments()
|
|
assert len(jobs) == 1
|
|
widget._load_departments(force=True)
|
|
assert len(jobs) == 2
|
|
jobs[1]["on_success"]([(12, "当前部门")])
|
|
widget.department_combo.setCurrentIndex(1)
|
|
jobs[0]["on_success"]([(99, "旧部门")])
|
|
jobs[0]["on_error"](RuntimeError("旧请求失败"))
|
|
assert widget.department_combo.currentData() == 12
|
|
assert widget.department_combo.findData(99) == -1
|
|
assert widget._departments_loaded
|
|
assert widget.department_combo.isEnabled()
|
|
|
|
|
|
def test_refresh_of_options_preserves_selected_id(page: Any, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
widget, _client = page
|
|
monkeypatch.setattr(module, "run_async", immediate)
|
|
widget._load_departments()
|
|
widget.department_combo.setCurrentIndex(widget.department_combo.findData(12))
|
|
widget._load_departments(force=True)
|
|
assert widget.department_combo.currentData() == 12
|
|
|
|
|
|
def test_legacy_repository_is_not_called_without_scope(page: Any, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
widget, _client = page
|
|
monkeypatch.setattr(module, "run_async", immediate)
|
|
calls: list[bool] = []
|
|
|
|
class OldRepository:
|
|
def list_departments(self):
|
|
calls.append(True)
|
|
return TREE
|
|
|
|
widget.repository = OldRepository()
|
|
widget._load_departments()
|
|
assert calls == []
|
|
assert not widget._departments_loaded
|
|
assert not widget.department_retry_button.isHidden()
|