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