1038 lines
56 KiB
Python
1038 lines
56 KiB
Python
"""Dual-model report and API contracts, exercised offscreen without network access."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import socket
|
||
from copy import deepcopy
|
||
from typing import Any
|
||
from uuid import UUID
|
||
|
||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||
|
||
import pytest
|
||
from PySide6.QtCore import Qt
|
||
from PySide6.QtWidgets import QApplication, QPushButton
|
||
|
||
from doctor_workstation.services import DemoDoctorRepository
|
||
from doctor_workstation.services.repository import RemoteDoctorRepository
|
||
from doctor_workstation.ui.dialogs import issued_prescription_ai as ai
|
||
from doctor_workstation.ui.dialogs import prescription as editor_module
|
||
from doctor_workstation.ui.pages import prescriptions as page_module
|
||
|
||
|
||
def batch(batch_id: int = 40, prescription_id: int = 801, status: str = "running") -> dict[str, Any]:
|
||
return {
|
||
"id": batch_id, "prescription_id": prescription_id, "prescription_revision": 2,
|
||
"patient_id": 301, "diagnosis_id": 501, "status": status,
|
||
"validity": "current", "comparison_type": "independent_baseline",
|
||
"source_summary": {"diagnosis_count": 2, "attachment_count": 4},
|
||
"missing": [{"source": "tongue", "reason": "第4张图片不可读"}],
|
||
"cutoff_at": "2026-09-09 10:00:00", "created_at": "2026-09-09 10:01:00",
|
||
"models": {
|
||
"qwen": {
|
||
"status": "succeeded", "report": {"summary": "已保存千问分析 <script>bad()</script>"},
|
||
"candidate": {"status": "available_for_review", "herbs": [{"name": "黄芪", "dosage": 12, "unit": "g", "dose_basis": "每剂", "formula_type": "主方"}]},
|
||
"comparison": {"status": "comparable", "score": 0, "herb_score": 0, "algorithm_version": "v1", "rows": [{"name": "黄芪", "doctor_dose": None, "candidate_dose": 12, "contribution": 0}], "usage_differences": ["疗程需复核"]},
|
||
"coverage": {"status": "incomplete", "manifest": [{"name": "舌象4", "status": "unreadable"}]},
|
||
"review": {"status": "needs_information", "comment": "需补充清晰舌象"},
|
||
},
|
||
"openai": {"status": "running", "report": None, "candidate": None, "comparison": None, "coverage": {"status": "pending"}},
|
||
},
|
||
}
|
||
|
||
|
||
class Repository(DemoDoctorRepository):
|
||
def __init__(self) -> None:
|
||
super().__init__()
|
||
self.calls: list[tuple[str, Any]] = []
|
||
self.batches = [batch()]
|
||
self.enabled = True
|
||
|
||
def list_prescription_ai_reports(self, **params: Any) -> dict[str, Any]:
|
||
self.calls.append(("reports", params))
|
||
return {"enabled": self.enabled, "lists": deepcopy(self.batches), "count": len(self.batches), "page_no": 1, "page_size": 20}
|
||
|
||
def get_prescription_ai_report(self, batch_id: int) -> dict[str, Any]:
|
||
self.calls.append(("detail", batch_id))
|
||
return deepcopy(next(row for row in self.batches if row["id"] == batch_id))
|
||
|
||
def retry_prescription_ai(self, batch_id: int, model_key: str) -> dict[str, Any]:
|
||
self.calls.append(("retry", (batch_id, model_key)))
|
||
return {"batch_id": batch_id, "status": "queued"}
|
||
|
||
def regenerate_prescription_ai(self, prescription_id: int, reason: str) -> dict[str, Any]:
|
||
self.calls.append(("regenerate", (prescription_id, reason)))
|
||
return {"batch_id": 41, "status": "queued"}
|
||
|
||
def review_prescription_ai(self, batch_id: int, model_key: str, status: str, comment: str) -> dict[str, Any]:
|
||
self.calls.append(("review", (batch_id, model_key, status, comment)))
|
||
return {"saved": True}
|
||
|
||
def list_prescription_ai_statuses(self, ids: list[int]) -> dict[str, Any]:
|
||
self.calls.append(("statuses", list(ids)))
|
||
return {"enabled": self.enabled, "items": [{"prescription_id": value, "batch_id": 40, "status": "running", "validity": "current", "comparison_type": "independent_baseline", "models": {"qwen": {"status": "succeeded", "score": 0, "comparison_status": "comparable"}, "openai": {"status": "running", "score": None}}} for value in ids] if self.enabled else []}
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def application() -> QApplication:
|
||
return QApplication.instance() or QApplication([])
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def no_network(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
def denied(*_args: Any, **_kwargs: Any) -> None:
|
||
pytest.fail("These tests must never contact external systems")
|
||
monkeypatch.setattr(socket.socket, "connect", denied)
|
||
monkeypatch.setattr(socket.socket, "connect_ex", denied)
|
||
monkeypatch.setattr(socket, "create_connection", denied)
|
||
|
||
|
||
@pytest.fixture
|
||
def immediate(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
def run(function: Any, *, on_success: Any = None, on_error: Any = None, on_finished: Any = None) -> object:
|
||
try:
|
||
result = function()
|
||
except Exception as error:
|
||
if on_error:
|
||
on_error(error)
|
||
else:
|
||
raise
|
||
else:
|
||
if on_success:
|
||
on_success(result)
|
||
finally:
|
||
if on_finished:
|
||
on_finished()
|
||
return object()
|
||
for module in (ai, page_module, editor_module):
|
||
monkeypatch.setattr(module, "run_async", run)
|
||
|
||
|
||
@pytest.mark.parametrize(("value", "expected"), [(None, "—"), (0, "0%"), (100, "100%"), (42.3, "42%"), (float("nan"), "—"), (float("inf"), "—"), (False, "—"), (-1, "—")])
|
||
def test_percentage_never_turns_missing_into_zero(value: Any, expected: str) -> None:
|
||
assert ai.percentage(value) == expected
|
||
|
||
|
||
def test_shared_report_reads_only_and_renders_each_model_independently(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
application.processEvents()
|
||
assert [call[0] for call in repository.calls] == ["reports", "detail"]
|
||
assert "千问分析" in dialog.model_views["qwen"]["report"].toPlainText()
|
||
assert dialog.model_views["qwen"]["score"].text() == "0.0%"
|
||
assert dialog.model_views["openai"]["score"].text() == "—"
|
||
assert "资料不全" in dialog.model_views["qwen"]["title"].text()
|
||
assert "黄芪" in dialog.model_views["qwen"]["candidate"].toPlainText()
|
||
assert "医生剂量" in dialog.model_views["qwen"]["comparison"].toPlainText()
|
||
assert "第4张图片不可读" in dialog.model_views["qwen"]["sources"].toPlainText()
|
||
assert "<script>" in dialog.model_views["qwen"]["report"].toHtml()
|
||
assert dialog._timer.isActive()
|
||
assert not dialog.model_views["qwen"]["retry"].isEnabled()
|
||
assert not dialog.model_views["openai"]["retry"].isEnabled()
|
||
dialog.hide()
|
||
assert not dialog._timer.isActive()
|
||
calls = list(repository.calls)
|
||
dialog._poll()
|
||
assert repository.calls == calls
|
||
|
||
|
||
def test_retry_and_review_target_only_selected_model(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
repository.batches[0]["status"] = "partial"
|
||
repository.batches[0]["models"]["openai"].update(status="failed", error_code="timeout", error_message="模型超时")
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
assert dialog.model_views["openai"]["retry"].isEnabled()
|
||
assert not dialog._timer.isActive()
|
||
dialog._retry("openai")
|
||
assert ("retry", (40, "openai")) in repository.calls
|
||
dialog.model_views["qwen"]["review_state"].setCurrentIndex(3)
|
||
dialog.model_views["qwen"]["comment"].setPlainText("已核对逐味差异")
|
||
dialog._review("qwen")
|
||
assert ("review", (40, "qwen", "reviewed", "已核对逐味差异")) in repository.calls
|
||
assert not any(call[0] == "regenerate" for call in repository.calls)
|
||
dialog.close()
|
||
|
||
|
||
def test_history_and_hidden_dialog_ignore_late_detail(application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
repository = Repository()
|
||
repository.batches.append(batch(41))
|
||
repository.batches[1]["models"]["qwen"]["report"] = {"summary": "第二批报告"}
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
pending = []
|
||
monkeypatch.setattr(ai, "run_async", lambda function, **callbacks: pending.append((function, callbacks)))
|
||
dialog._poll()
|
||
assert len(pending) == 1
|
||
old_generation = dialog._generation
|
||
dialog.history.setCurrentIndex(1)
|
||
assert dialog._generation > old_generation
|
||
old_function, old_callbacks = pending[0]
|
||
old_callbacks["on_success"](old_function())
|
||
assert dialog._batch == {}
|
||
new_function, new_callbacks = pending[1]
|
||
new_callbacks["on_success"](new_function())
|
||
assert dialog._batch["id"] == 41
|
||
assert "第二批报告" in dialog.model_views["qwen"]["report"].toPlainText()
|
||
dialog.hide()
|
||
new_callbacks["on_success"](batch(41))
|
||
assert "第二批报告" in dialog.model_views["qwen"]["report"].toPlainText()
|
||
|
||
|
||
def test_wrong_patient_batch_is_not_displayed(application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
repository = Repository()
|
||
wrong_patient = batch()
|
||
wrong_patient["patient_id"] = 999
|
||
monkeypatch.setattr(repository, "get_prescription_ai_report", lambda _batch_id: wrong_patient)
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], diagnosis_id=501)
|
||
dialog.show()
|
||
assert repository.calls[0][1]["diagnosis_id"] == 501
|
||
assert dialog._batch == {}
|
||
assert "归属" in dialog.message.text()
|
||
assert not dialog.model_views["qwen"]["report"].toPlainText()
|
||
dialog.close()
|
||
|
||
|
||
def test_patient_history_accepts_other_authorized_visits(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
repository.batches[0]["diagnosis_id"] = 499
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], diagnosis_id=501)
|
||
dialog.show()
|
||
assert dialog._batch["diagnosis_id"] == 499
|
||
assert "千问分析" in dialog.model_views["qwen"]["report"].toPlainText()
|
||
dialog.close()
|
||
|
||
|
||
def test_disabled_generation_keeps_history_readable(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
repository.enabled = False
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
assert "千问分析" in dialog.model_views["qwen"]["report"].toPlainText()
|
||
assert not dialog.regenerate_button.isEnabled()
|
||
assert "未启用" in dialog.message.text()
|
||
dialog.close()
|
||
|
||
|
||
def test_view_permission_cannot_mutate(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
permissions = ["tcm.prescriptionAi/reports", "tcm.prescriptionAi/detail"]
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, permissions, prescription_id=801)
|
||
dialog.show()
|
||
assert dialog.regenerate_button.isHidden()
|
||
dialog._review("qwen")
|
||
dialog._retry("openai")
|
||
assert not any(call[0] in {"review", "retry", "regenerate"} for call in repository.calls)
|
||
dialog.close()
|
||
|
||
|
||
def test_list_batches_visible_ids_and_stops_on_hide(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
page = page_module.PrescriptionsPage(repository, ["*"])
|
||
page.resize(1366, 800)
|
||
page.show()
|
||
application.processEvents()
|
||
statuses = [call for call in repository.calls if call[0] == "statuses"]
|
||
assert statuses and set(statuses[-1][1]) <= {801, 802}
|
||
assert len(statuses[-1][1]) <= 100
|
||
assert not page.table.isColumnHidden(11)
|
||
assert not page.table.isColumnHidden(12)
|
||
assert "千问 0%" in page.table.item(0, 12).text()
|
||
assert "OpenAI —" in page.table.item(0, 12).text()
|
||
assert any(button.accessibleName() == "AI 报告" for button in page.table.findChildren(QPushButton))
|
||
assert page._ai_timer.isActive()
|
||
page.hide()
|
||
assert not page._ai_timer.isActive()
|
||
calls = list(repository.calls)
|
||
page._load_ai_statuses()
|
||
assert repository.calls == calls
|
||
|
||
|
||
def test_list_disabled_explains_availability_and_keeps_history(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
repository.enabled = False
|
||
page = page_module.PrescriptionsPage(repository, ["*"])
|
||
page.resize(1366, 800)
|
||
page.show()
|
||
application.processEvents()
|
||
assert page.table.isColumnHidden(11) and page.table.isColumnHidden(12)
|
||
assert not page._ai_timer.isActive()
|
||
assert page.ai_report_button.isVisible()
|
||
assert page.ai_status_notice.isVisible()
|
||
assert "未启用" in page.ai_status_notice.text()
|
||
page.close()
|
||
|
||
|
||
def test_list_service_error_is_visible_and_clears_after_recovery(application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
repository = Repository()
|
||
|
||
def unavailable(_ids: list[int]) -> dict[str, Any]:
|
||
raise RuntimeError("AI service unavailable")
|
||
|
||
monkeypatch.setattr(repository, "list_prescription_ai_statuses", unavailable)
|
||
page = page_module.PrescriptionsPage(repository, ["*"])
|
||
page.resize(1366, 800)
|
||
page.show()
|
||
application.processEvents()
|
||
assert page.ai_status_notice.isVisible()
|
||
assert "暂不可用" in page.ai_status_notice.text()
|
||
assert page.table.isColumnHidden(12)
|
||
assert not page._ai_timer.isActive()
|
||
monkeypatch.setattr(repository, "list_prescription_ai_statuses", Repository.list_prescription_ai_statuses.__get__(repository))
|
||
page.refresh()
|
||
application.processEvents()
|
||
assert "已启用" in page.ai_status_notice.text()
|
||
assert not page.ai_status_notice.toolTip()
|
||
assert not page.table.isColumnHidden(12)
|
||
page.close()
|
||
|
||
|
||
def test_list_missing_permission_explains_without_loading_ai(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
page = page_module.PrescriptionsPage(repository, ["cf.prescription/read"])
|
||
page.resize(1366, 800)
|
||
page.show()
|
||
application.processEvents()
|
||
assert page.ai_status_notice.isVisible()
|
||
assert "权限" in page.ai_status_notice.text()
|
||
assert page.table.isColumnHidden(11) and page.table.isColumnHidden(12)
|
||
assert not page.ai_report_button.isVisible()
|
||
assert not any(call[0] in {"statuses", "reports", "detail"} for call in repository.calls)
|
||
page.close()
|
||
|
||
|
||
def test_list_ignores_response_after_filter_generation_changes(application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
repository = Repository()
|
||
page = page_module.PrescriptionsPage(repository, ["*"])
|
||
page.resize(1366, 800)
|
||
page.show()
|
||
application.processEvents()
|
||
callbacks = []
|
||
monkeypatch.setattr(page_module, "run_async", lambda function, **kwargs: callbacks.append((function, kwargs)))
|
||
page._load_ai_statuses()
|
||
assert callbacks
|
||
page._generation += 1
|
||
page._ai_statuses = {}
|
||
function, call = callbacks[0]
|
||
call["on_success"](function())
|
||
assert page._ai_statuses == {}
|
||
page.close()
|
||
|
||
|
||
def test_statistics_nulls_and_absent_review_samples(application: QApplication, immediate: None) -> None:
|
||
class StatisticsRepository:
|
||
def prescription_ai_statistics(self, *_args: Any) -> dict[str, Any]:
|
||
return {"total_count": 3, "patient_count": 2, "doctors": [{"doctor_name": "张医生", "total_count": 3, "patient_count": 2, "models": {"qwen": {"eligible_count": 0, "coverage_rate": 0, "mean": None, "median": None, "excluded_reasons": {"non_independent": 3}}, "openai": {"eligible_count": 1, "coverage_rate": 33.3, "mean": 0, "median": 0}}, "review": {"evaluated_count": 0, "qualified_count": 0, "qualified_rate": None}}]}
|
||
dialog = ai.PrescriptionAiStatisticsDialog(StatisticsRepository(), ["*"])
|
||
dialog.show()
|
||
text = dialog.report.toPlainText()
|
||
assert "均值：—" in text and "均值：0.0%" in text
|
||
assert "未建立复核样本" in text
|
||
assert "开方事件：3" in dialog.summary.text()
|
||
dialog.close()
|
||
|
||
|
||
def test_editor_keeps_request_key_for_identical_content_and_preserves_ai_assistance(application: QApplication, immediate: None) -> None:
|
||
editor = editor_module.PrescriptionEditorDialog(DemoDoctorRepository(), {"ai_assisted": True})
|
||
first = editor.payload()
|
||
assert first["ai_assisted"] is True
|
||
assert str(UUID(first["request_key"])) == first["request_key"]
|
||
assert first["request_key"] == editor.payload()["request_key"]
|
||
editor.patient_name.setText("新患者")
|
||
assert first["request_key"] != editor.payload()["request_key"]
|
||
editor.close()
|
||
unknown = editor_module.PrescriptionEditorDialog(DemoDoctorRepository())
|
||
assert "ai_assisted" not in unknown.payload()
|
||
unknown.close()
|
||
unconfirmed = editor_module.PrescriptionEditorDialog(DemoDoctorRepository(), {"ai_assisted": False})
|
||
assert "ai_assisted" not in unconfirmed.payload()
|
||
unconfirmed.close()
|
||
|
||
|
||
def test_repository_exact_read_and_mutation_contracts() -> None:
|
||
class Client:
|
||
def __init__(self) -> None:
|
||
self.calls = []
|
||
def get(self, endpoint: str, params: Any) -> dict[str, Any]:
|
||
self.calls.append(("GET", endpoint, params))
|
||
return {"enabled": False, "items": []}
|
||
def post(self, endpoint: str, params: Any) -> dict[str, Any]:
|
||
self.calls.append(("POST", endpoint, params))
|
||
return {"saved": True}
|
||
client = Client()
|
||
repository = RemoteDoctorRepository(client)
|
||
assert repository.list_prescription_ai_statuses([801, 802, 801]) == {"enabled": False, "items": []}
|
||
repository.list_prescription_ai_reports(diagnosis_id=501, page_no=2, page_size=20)
|
||
repository.get_prescription_ai_report(40)
|
||
assert all(call[0] == "GET" for call in client.calls)
|
||
repository.regenerate_prescription_ai(801, "补充病历")
|
||
repository.retry_prescription_ai(40, "openai")
|
||
repository.review_prescription_ai(40, "qwen", "reviewed", "核对完成")
|
||
repository.prescription_ai_statistics("2026-09-01", "2026-09-09", 7)
|
||
assert client.calls == [
|
||
("GET", "tcm.prescriptionAi/statuses", {"ids": "801,802"}),
|
||
("GET", "tcm.prescriptionAi/reports", {"diagnosis_id": 501, "page_no": 2, "page_size": 20}),
|
||
("GET", "tcm.prescriptionAi/detail", {"batch_id": 40}),
|
||
("POST", "tcm.prescriptionAi/regenerate", {"prescription_id": 801, "reason": "补充病历"}),
|
||
("POST", "tcm.prescriptionAi/retry", {"batch_id": 40, "model_key": "openai"}),
|
||
("POST", "tcm.prescriptionAi/review", {"batch_id": 40, "model_key": "qwen", "status": "reviewed", "comment": "核对完成"}),
|
||
("GET", "tcm.prescriptionAi/statistics", {"date_from": "2026-09-01", "date_to": "2026-09-09", "doctor_id": 7}),
|
||
]
|
||
with pytest.raises(ValueError):
|
||
repository.list_prescription_ai_statuses(list(range(1, 102)))
|
||
with pytest.raises(ValueError):
|
||
repository.list_prescription_ai_reports(prescription_id=801, diagnosis_id=501)
|
||
with pytest.raises(ValueError):
|
||
repository.retry_prescription_ai(40, "other")
|
||
|
||
|
||
def test_stale_list_score_is_not_attached_to_current_prescription() -> None:
|
||
value = {"validity": "superseded", "models": {"qwen": {"score": 100}, "openai": {"score": 90}}}
|
||
assert ai.agreement_text(value) == "千问 —\nOpenAI —"
|
||
assert ai.state_text(value) == "处方已变更"
|
||
assert not ai.batch_running({**value, "status": "running"})
|
||
|
||
|
||
@pytest.mark.parametrize("status", ["preparing", "waiting_sources", "retry_wait", "queued", "running"])
|
||
def test_server_pending_states_continue_polling(status: str) -> None:
|
||
assert ai.batch_running({"status": status, "validity": "current"})
|
||
assert not ai.batch_running({"status": status, "validity": "prescription_changed"})
|
||
|
||
|
||
def test_regeneration_selects_new_batch_and_preserves_old_history(application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
repository = Repository()
|
||
repository.batches[0]["status"] = "success"
|
||
repository.batches[0]["models"]["openai"]["status"] = "success"
|
||
def regenerate(prescription_id: int, reason: str) -> dict[str, Any]:
|
||
repository.calls.append(("regenerate", (prescription_id, reason)))
|
||
repository.batches.insert(0, batch(41))
|
||
return {"batch_id": 41, "status": "queued"}
|
||
monkeypatch.setattr(repository, "regenerate_prescription_ai", regenerate)
|
||
monkeypatch.setattr(ai.QInputDialog, "getMultiLineText", lambda *_args: ("已补充资料", True))
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
dialog._regenerate()
|
||
assert dialog.history.count() == 2
|
||
assert dialog.history.currentData() == 41
|
||
assert dialog._batch["id"] == 41
|
||
assert dialog._timer.isActive()
|
||
assert ("regenerate", (801, "已补充资料")) in repository.calls
|
||
dialog.close()
|
||
|
||
|
||
def test_list_append_retains_finished_cached_scores(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
page = page_module.PrescriptionsPage(repository, ["*"])
|
||
page.resize(1366, 800)
|
||
page.show()
|
||
application.processEvents()
|
||
for cached in page._ai_statuses.values():
|
||
cached["status"] = "success"
|
||
cached["models"]["openai"].update(status="success", score=62)
|
||
calls = len([call for call in repository.calls if call[0] == "statuses"])
|
||
page._apply_result(repository.list_prescriptions(), page._generation, 1)
|
||
assert "OpenAI 62%" in page.table.item(0, 12).text()
|
||
assert len([call for call in repository.calls if call[0] == "statuses"]) == calls
|
||
assert not page._ai_timer.isActive()
|
||
page.close()
|
||
|
||
|
||
def test_context_menu_targets_clicked_prescription(application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
repository = Repository()
|
||
opened = []
|
||
monkeypatch.setattr(page_module, "present_issued_prescription_ai", lambda *args, **kwargs: opened.append(kwargs))
|
||
page = page_module.PrescriptionsPage(repository, ["*"])
|
||
page.resize(1366, 800)
|
||
page.show()
|
||
application.processEvents()
|
||
item = page.table.item(1, 1)
|
||
expected_id = page.table.item(1, 0).data(Qt.ItemDataRole.UserRole).id
|
||
page._open_ai_context_menu(page.table.visualItemRect(item).center())
|
||
page._ai_context_menu.actions()[0].trigger()
|
||
assert opened == [{"prescription_id": expected_id}]
|
||
page._ai_context_menu.close()
|
||
page.close()
|
||
|
||
|
||
def test_real_server_comparison_rows_keep_both_doses_and_basis() -> None:
|
||
rows = ai._comparison_rows([{
|
||
"key": "1|raw|main", "medicine_id": 1, "name": "黄芪", "processing": "生品", "formula_type": "主方",
|
||
"administration_route": "口服", "group": "", "doctor": {"dosage": 30, "unit": "g", "dose_basis": "per_dose"},
|
||
"candidate": {"dosage": 15, "unit": "g", "dose_basis": "per_dose"},
|
||
"doctor_dosage": 30, "candidate_dosage": 15, "unit": "g", "dose_basis": "per_dose",
|
||
"match_type": "matched", "contribution": 0.5,
|
||
}])
|
||
assert rows[0]["doctor_dosage"] == "30 克"
|
||
assert rows[0]["candidate_dosage"] == "15 克"
|
||
assert rows[0]["dose_basis"] == "per_dose"
|
||
assert rows[0]["contribution"] == 0.5
|
||
assert "生品 / 主方 / 口服" in rows[0]["name"]
|
||
|
||
|
||
def test_unavailable_candidate_has_reason_and_pending_review_is_disabled(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
repository.batches[0]["models"]["qwen"]["candidate"] = {"status": "withheld_for_risk", "reason": "过敏用药信息待核实", "herbs": []}
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
text = dialog.model_views["qwen"]["candidate"].toPlainText()
|
||
assert "因风险暂缓候选用药" in text and "过敏用药信息待核实" in text
|
||
assert not dialog.model_views["openai"]["save"].isEnabled()
|
||
assert "分析尚未完成" in dialog.model_views["openai"]["candidate"].toPlainText()
|
||
dialog.close()
|
||
|
||
|
||
def test_preparation_failure_reason_is_visible_without_models(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
repository.batches[0].update(status="blocked", models={}, error_message="患者关联冲突，需要核对诊单", coverage_status="partial")
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
assert "患者关联冲突" in dialog.batch_summary.text()
|
||
assert "资料不全" in dialog.batch_summary.text()
|
||
assert "患者关联冲突" in dialog.model_views["qwen"]["candidate"].toPlainText()
|
||
dialog.close()
|
||
|
||
|
||
def test_actual_history_pagination_contract(application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
repository = Repository()
|
||
repository.batches = [batch(value) for value in range(61, 40, -1)]
|
||
def reports(**params: Any) -> dict[str, Any]:
|
||
repository.calls.append(("reports", params))
|
||
page = params["page_no"]
|
||
return {"lists": repository.batches[(page - 1) * 20:page * 20], "count": 21, "page_no": page, "page_size": 20}
|
||
monkeypatch.setattr(repository, "list_prescription_ai_reports", reports)
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
assert dialog.history.count() == 20 and dialog.next.isEnabled()
|
||
dialog._history_page(1)
|
||
assert dialog.history.count() == 1
|
||
assert dialog._batch["id"] == 41
|
||
assert dialog.previous.isEnabled() and not dialog.next.isEnabled()
|
||
assert repository.calls[-2] == ("reports", {"prescription_id": 801, "diagnosis_id": 0, "page_no": 2, "page_size": 20})
|
||
dialog.close()
|
||
|
||
|
||
@pytest.mark.parametrize(("status", "expected"), [("blank", "尚未开方"), ("not_generated", "尚无分析记录")])
|
||
def test_server_empty_batch_statuses_are_not_fake_scores(status: str, expected: str) -> None:
|
||
data = {"status": status, "models": {}, "coverage_status": "pending"}
|
||
assert ai.state_text(data) == expected
|
||
assert ai.agreement_text(data) == "千问 —\nOpenAI —"
|
||
|
||
|
||
def test_statistics_displays_actual_version_strata_without_filling_null_mean(application: QApplication, immediate: None) -> None:
|
||
class StatisticsRepository:
|
||
def prescription_ai_statistics(self, *_args: Any) -> dict[str, Any]:
|
||
strata = [{"versions": {"model_version": version, "algorithm_version": "v1", "prompt_version": "p1", "dictionary_version": "d1"}, "count": 1, "mean": mean, "median": mean, "distribution": {"[0,20)": 1 if mean == 0 else 0}, "sample_status": "insufficient_sample"} for version, mean in (("model-1", 0), ("model-2", 80))]
|
||
return {"total_count": 2, "patient_count": 1, "doctors": [{"doctor_id": 7, "doctor_name": "张医生", "total_count": 2, "patient_count": 1, "paired_count": 0, "models": {"qwen": {"eligible_count": 2, "coverage_rate": 100, "mean": None, "median": None, "excluded_reasons": {}, "strata": strata}}, "review": {"evaluated_count": 0, "qualified_count": 0, "qualified_rate": None}}]}
|
||
dialog = ai.PrescriptionAiStatisticsDialog(StatisticsRepository(), ["*"])
|
||
dialog.show()
|
||
text = dialog.report.toPlainText()
|
||
assert "双模型共同有效样本：0" in text
|
||
assert "整体均值与中位数不合并" in text
|
||
assert "model-1" in text and "model-2" in text
|
||
assert "均值：—" in text and "均值：0.0%" in text and "均值：80.0%" in text
|
||
assert "样本不足" in text and "未建立复核样本" in text
|
||
dialog.close()
|
||
|
||
|
||
@pytest.mark.parametrize(("code", "expected"), [
|
||
("SOURCE_HISTORY_VERSIONS_UNAVAILABLE", "来源历史版本无法核验"),
|
||
("ARCHIVE_SYNC_WATERMARK_UNAVAILABLE: chat_records", "归档同步完整性尚未核验：聊天记录"),
|
||
("TRANSCRIPT_NOT_VERIFIED_COMPLETE：call_records:1820", "问诊转写完整性尚未核验：问诊通话（编号：1820）"),
|
||
("TRANSCRIPT_PARTIAL: video_calls:1821", "问诊转写仅部分完成：视频问诊（编号：1821）"),
|
||
("CRITICAL_CLINICAL_FACT_MISSING：clinical.allergy_history", "关键临床资料缺失：临床资料 · 过敏史"),
|
||
("MODEL_FILE_OUTPUT_INVALID", "模型未能正确解析这组附件"),
|
||
("missing_or_unknown_dose_basis", "每剂或每日剂量基准不明确"),
|
||
("review_not_independent", "复核不具独立性"),
|
||
])
|
||
def test_metadata_and_historical_composite_reasons_are_chinese(code: str, expected: str) -> None:
|
||
assert ai.label(code) == expected
|
||
assert expected in ai._html({"missing_information": [code]})
|
||
|
||
|
||
def test_source_labels_keep_exact_ids_and_distinguish_source_kinds() -> None:
|
||
fingerprint = "0123456789abcdef" * 4
|
||
payload = {"evidence_references": ["diagnoses:501", "video_calls:1820", "call_records:1820", "file:" + fingerprint],
|
||
"source_id": "clinical.pregnancy_history", "redaction_manifest": ["diagnoses:501:doctor_advice"]}
|
||
original = deepcopy(payload)
|
||
rendered = ai._html(payload)
|
||
assert "诊单（编号：501）" in rendered
|
||
assert "视频问诊（编号：1820）" in rendered and "问诊通话（编号：1820）" in rendered
|
||
assert f"附件（编号：{fingerprint}）" in rendered
|
||
assert "临床资料 · 妊娠与哺乳情况" in rendered
|
||
assert "诊单（编号：501） · 医嘱" in rendered
|
||
assert all(raw not in rendered for raw in ("diagnoses:", "video_calls:", "file:", "clinical.", "doctor_advice"))
|
||
assert payload == original
|
||
|
||
|
||
def test_unknown_metadata_has_chinese_fallback_without_translating_clinical_prose() -> None:
|
||
clinical_text = "CT、MRI、HbA1c 6.2%，HbA1c_result 待复查，Follow-up in 2 weeks；unknown"
|
||
payload = {"summary": clinical_text, "treatment_advice": "ALT_AST 需结合 COPD 病史评估",
|
||
"status": "newphase", "error_code": "FUTURE_FAILURE_CODE", "future_field": 3,
|
||
"future_status": "newstate", "future_enum": "new_enum_value",
|
||
"missing_information": ["FUTURE_GAP_CODE：future_records:1820"], "source_id": "future_records:1821"}
|
||
original = deepcopy(payload)
|
||
rendered = ai._html(payload)
|
||
assert clinical_text in rendered and "ALT_AST 需结合 COPD 病史评估" in rendered
|
||
assert "未识别的系统状态" in rendered and "其他来源（编号：1820）" in rendered
|
||
assert "其他来源（编号：1821）" in rendered and "3" in rendered
|
||
for raw in ("newphase", "newstate", "new_enum_value", "FUTURE_FAILURE_CODE", "FUTURE_GAP_CODE", "future_field", "future_records"):
|
||
assert raw not in rendered
|
||
assert payload == original
|
||
|
||
|
||
def test_nested_metadata_translates_fields_values_and_retains_all_statistics() -> None:
|
||
payload = {
|
||
"coverage": {"status": "partial", "source_complete": True, "files": {"file:42": {"status": "unreadable", "version_verified": False}}},
|
||
"source_summary": {"diagnoses_count": 2, "video_calls_count": 3, "source_record_count": 5, "history_versioning": "unavailable"},
|
||
"normalization": {"doctor": {"defaults": [{"field": "unit", "value": "g"}, {"field": "dose_basis", "value": "per_dose"}]},
|
||
"issues": [{"side": "both", "code": "unit_mismatch", "row": 0}], "unit_policy": "spelling_aliases_only_no_quantity_conversion"},
|
||
"exclusion_reasons": {"SOURCE_HISTORY_VERSIONS_UNAVAILABLE": 12, "future_reason_a": 7, "future_reason_b": 8},
|
||
"review": {"sampling_groups": [{"sampling_method": "risk_directed", "outcomes": {"qualified": 2, "needs_revision": 1}}]},
|
||
}
|
||
original = deepcopy(payload)
|
||
rendered = ai._html(payload)
|
||
for expected in ("资料不全", "历史版本无法核验", "诊单数", "视频问诊数", "克", "每剂", "剂量单位不同", "双方", "按风险抽样", "合格", "需修订"):
|
||
assert expected in rendered
|
||
for count in (12, 7, 8, 0):
|
||
assert f"<p>{count}</p>" in rendered
|
||
assert rendered.count("未识别的系统标识") == 2
|
||
for raw in ("partial", "version_verified", "diagnoses_count", "per_dose", "SOURCE_HISTORY", "future_reason", "risk_directed", "qualified"):
|
||
assert raw not in rendered
|
||
assert payload == original
|
||
|
||
|
||
def test_metadata_and_prose_remain_html_escaped(application: QApplication) -> None:
|
||
attack = '<img src="https://example.invalid/tracker" onerror="alert(1)"><script>bad()</script>'
|
||
html = ai._html({"summary": attack, "missing_information": ["TRANSCRIPT_FAILED：call_records:1820 " + attack],
|
||
"source_id": "call_records:1820 " + attack, "status": attack})
|
||
assert "<img" not in html and "<script" not in html
|
||
assert "<img" in html and "<script>" in html
|
||
assert "TRANSCRIPT_FAILED" not in html and "call_records:" not in html
|
||
browser = ai._browser()
|
||
browser.setHtml(html)
|
||
assert not browser.openLinks() and not browser.openExternalLinks()
|
||
assert attack in browser.toPlainText()
|
||
browser.close()
|
||
|
||
|
||
def chinese_history_batch() -> dict[str, Any]:
|
||
"""Synthetic fixture; no patient or network data is used in tests or visual QA."""
|
||
value = batch(39, status="success")
|
||
value.update(
|
||
error_message="SOURCE_HISTORY_VERSIONS_UNAVAILABLE", coverage_status="partial", baseline_eligible=False,
|
||
baseline_exclusion_reasons=["SOURCE_HISTORY_VERSIONS_UNAVAILABLE", "ATTACHMENT_TARGET_PLAN_LEAKAGE_UNVERIFIED"],
|
||
source_summary={"diagnoses_count": 2, "video_calls_count": 1, "source_record_count": 3, "attachment_count": 1,
|
||
"missing_count": 2, "snapshot_complete": False, "history_versioning": "unavailable", "archive_sync_verified": False},
|
||
missing=[{"source_id": "chat_records", "code": "ARCHIVE_SYNC_WATERMARK_UNAVAILABLE", "critical": False},
|
||
{"source_id": "call_records:1820", "code": "TRANSCRIPT_NOT_VERIFIED_COMPLETE", "critical": False}],
|
||
)
|
||
for model in value["models"].values():
|
||
model.update(status="success", error_message="SOURCE_HISTORY_VERSIONS_UNAVAILABLE",
|
||
report={"summary": "测试病例：依据已读取病历形成辅助分析，CT 与 HbA1c 结果需结合后续复查。",
|
||
"missing_information": ["ARCHIVE_SYNC_WATERMARK_UNAVAILABLE: chat_records", "TRANSCRIPT_NOT_VERIFIED_COMPLETE：call_records:1820"],
|
||
"evidence_references": ["diagnoses:501", "video_calls:1820"]},
|
||
coverage={"status": "partial", "complete": False, "source_ids": ["diagnoses:501", "video_calls:1820"]},
|
||
candidate={"status": "available_for_review", "reason": "基于已读资料生成，资料尚不完整，须由医生核对后决定是否采用。",
|
||
"herbs": [{"name": "测试药项", "dosage": 12, "unit": "g", "dose_basis": "per_dose", "formula_type": "main"}]},
|
||
comparison={"status": "not_comparable", "reason": "SOURCE_HISTORY_VERSIONS_UNAVAILABLE", "score": None,
|
||
"prompt_version": "manual-prescription-independent-v1"})
|
||
return value
|
||
|
||
|
||
def test_historical_reports_localize_every_panel_without_rewriting_saved_data(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
repository.batches.append(chinese_history_batch())
|
||
original = deepcopy(repository.batches)
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
dialog.history.setCurrentIndex(1)
|
||
views = dialog.model_views["qwen"]
|
||
content = "\n".join([dialog.batch_summary.text(), views["title"].text()] + [views[field].toPlainText() for field in ("report", "candidate", "comparison", "sources")])
|
||
for raw in ("SOURCE_HISTORY_VERSIONS_UNAVAILABLE", "ARCHIVE_SYNC_WATERMARK_UNAVAILABLE", "TRANSCRIPT_NOT_VERIFIED_COMPLETE", "call_records:", "diagnoses:", "main", "per_dose", "manual-prescription-independent-v1"):
|
||
assert raw not in content
|
||
for translated in ("来源历史版本无法核验", "归档同步完整性尚未核验", "问诊通话（编号：1820）", "CT 与 HbA1c", "资料不全", "手动处方独立分析 · 第 1 版"):
|
||
assert translated in content
|
||
assert dialog._batch == original[1] and repository.batches == original
|
||
assert dialog.batch_summary.textFormat() == Qt.TextFormat.PlainText
|
||
assert views["title"].textFormat() == Qt.TextFormat.PlainText
|
||
assert all(call[0] in {"reports", "detail"} for call in repository.calls)
|
||
assert not dialog._timer.isActive()
|
||
dialog.close()
|
||
|
||
|
||
@pytest.mark.parametrize(("version", "expected"), [
|
||
("manual-prescription-independent-v1", "手动处方独立分析 · 第 1 版"),
|
||
("manual-prescription-available-evidence-v2", "手动处方已读资料分析 · 第 2 版"),
|
||
("prescription-soft-dice-v1.0.1", "处方药味剂量一致度算法 · 第 1.0.1 版"),
|
||
])
|
||
def test_internal_version_labels_keep_revision_numbers(version: str, expected: str) -> None:
|
||
html = ai._html({"prompt_version": version})
|
||
assert expected in html and version not in html
|
||
|
||
|
||
class ChineseStatisticsRepository:
|
||
def prescription_ai_statistics(self, *_args: Any) -> dict[str, Any]:
|
||
exclusions = {"SOURCE_HISTORY_VERSIONS_UNAVAILABLE": 3, "transcript_not_final": 1,
|
||
"ATTACHMENT_TARGET_PLAN_LEAKAGE_UNVERIFIED": 2, "future_reason_a": 4, "future_reason_b": 5}
|
||
return {"total_count": 15, "patient_count": 12, "doctors": [{"doctor_name": "测试医生", "total_count": 15, "patient_count": 12,
|
||
"paired_count": 0, "models": {key: {"eligible_count": 0, "coverage_rate": 0, "mean": None, "median": None,
|
||
"excluded_reasons": exclusions} for key in ("qwen", "openai")},
|
||
"review": {"evaluated_count": 0, "qualified_count": 0, "qualified_rate": None}}]}
|
||
|
||
|
||
def test_statistics_localizes_exclusion_keys_without_combining_unknown_reasons(application: QApplication, immediate: None) -> None:
|
||
dialog = ai.PrescriptionAiStatisticsDialog(ChineseStatisticsRepository(), ["*"])
|
||
dialog.show()
|
||
content = dialog.report.toPlainText()
|
||
assert "来源历史版本无法核验" in content and "附件可能包含本次处方，独立性未核验" in content
|
||
assert "问诊转写尚未完整归档" in content
|
||
assert content.count("未识别的系统标识") == 4
|
||
for raw in ("SOURCE_HISTORY", "transcript_not_final", "ATTACHMENT_TARGET", "future_reason"):
|
||
assert raw not in content
|
||
assert "4" in content and "5" in content
|
||
assert "均值：—" in content and "覆盖率：0.0%" in content and "未建立复核样本" in content
|
||
assert "开方事件：15" in dialog.summary.text()
|
||
dialog.close()
|
||
|
||
|
||
def progress_batch() -> dict[str, Any]:
|
||
"""Synthetic progress fixture shared by tests and offscreen visual review."""
|
||
value = batch()
|
||
value["progress"] = {"stage": "unknown", "phase": "running", "elapsed_seconds": 190,
|
||
"stage_elapsed_seconds": None, "updated_at": 1000, "server_time": 1008,
|
||
"notice": "模型正在分别处理，具体进度见各模型。"}
|
||
value["models"]["qwen"].update(status="running", report=None, candidate=None, comparison=None)
|
||
value["models"]["qwen"]["progress"] = {
|
||
"stage": "files", "stage_label": "处理附件", "phase": "waiting", "completed_units": 2, "total_units": 5,
|
||
"elapsed_seconds": 176, "stage_elapsed_seconds": 46, "updated_at": 1000, "server_time": 1008,
|
||
"notice": "等待模型返回。组数包含已处理及已明确无法读取的附件组；不代表附件全部读懂。",
|
||
}
|
||
value["models"]["openai"]["progress"] = {
|
||
"stage": "final", "stage_label": "生成分析报告", "phase": "waiting", "completed_units": None, "total_units": None,
|
||
"elapsed_seconds": 174, "stage_elapsed_seconds": 70, "updated_at": 970, "server_time": 1008,
|
||
"notice": "等待模型返回。完成报告后还需校验和用药对照。",
|
||
}
|
||
return value
|
||
|
||
|
||
def test_progress_two_models_show_distinct_real_stages_and_counts(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
repository.batches = [progress_batch()]
|
||
original = deepcopy(repository.batches)
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
qwen, openai = dialog.model_views["qwen"], dialog.model_views["openai"]
|
||
assert qwen["progress_title"].text() == "分析附件 · 本阶段 2/5 组"
|
||
assert qwen["progress_bar"].maximum() == 5 and qwen["progress_bar"].value() == 2
|
||
assert openai["progress_title"].text() == "生成完整报告"
|
||
assert openai["progress_bar"].maximum() == 0
|
||
assert "校验和用药对照" in openai["progress_detail"].text()
|
||
assert "%" not in qwen["progress_title"].text() + openai["progress_title"].text()
|
||
assert "用药对照：待处理" in dialog.progress_flow.text()
|
||
assert dialog._timer.isActive() and dialog._progress_timer.isActive()
|
||
assert "千问 附件 2/5\nOpenAI 生成报告" in ai.state_text(original[0])
|
||
assert "本阶段 2/5 组" in ai.status_tooltip(original[0])
|
||
assert repository.batches == original
|
||
assert all(call[0] in {"reports", "detail"} for call in repository.calls)
|
||
dialog.close()
|
||
assert not dialog._timer.isActive() and not dialog._progress_timer.isActive()
|
||
|
||
|
||
def test_progress_local_clock_uses_server_durations_without_additional_requests(application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
ticks = [100.0]
|
||
monkeypatch.setattr(ai, "monotonic", lambda: ticks[0])
|
||
repository = Repository()
|
||
repository.batches = [progress_batch()]
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
calls = list(repository.calls)
|
||
ticks[0] = 112.0
|
||
dialog._render_progress()
|
||
detail = dialog.model_views["qwen"]["progress_detail"].text()
|
||
assert "已用时 3 分 08 秒" in detail and "本阶段 58 秒" in detail
|
||
assert "阶段更新于 20 秒前" in detail
|
||
assert repository.calls == calls
|
||
dialog.hide()
|
||
frozen = dialog.model_views["qwen"]["progress_detail"].text()
|
||
ticks[0] = 200.0
|
||
dialog._render_progress()
|
||
assert dialog.model_views["qwen"]["progress_detail"].text() == frozen
|
||
assert not dialog._timer.isActive() and not dialog._progress_timer.isActive()
|
||
assert dialog.model_views["openai"]["progress_bar"].maximum() != 0
|
||
|
||
|
||
def test_preparing_countdown_reaches_deadline_without_implying_report_completion(application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
ticks = [100.0]
|
||
monkeypatch.setattr(ai, "monotonic", lambda: ticks[0])
|
||
repository = Repository()
|
||
value = batch(status="waiting_sources")
|
||
value["models"] = {}
|
||
value["progress"] = {"stage": "waiting_sources", "phase": "waiting", "elapsed_seconds": 12,
|
||
"wait_remaining_seconds": 10, "updated_at": 1000, "server_time": 1002}
|
||
repository.batches = [value]
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
assert "资料等待窗口剩余 10 秒" in dialog.batch_progress.text()
|
||
assert "双模型分析：待开始" in dialog.progress_flow.text()
|
||
ticks[0] = 115.0
|
||
dialog._render_progress()
|
||
assert "等待窗口已到，等待服务端确认" in dialog.batch_progress.text()
|
||
assert "已完成" not in dialog.batch_progress.text()
|
||
assert dialog._progress_timer.isActive()
|
||
assert all(view["progress_bar"].maximum() == 0 for view in dialog.model_views.values())
|
||
dialog.close()
|
||
|
||
|
||
@pytest.mark.parametrize(("stage", "expected"), [("text", "分析文字资料"), ("reduce", "汇总资料要点"), ("validating", "校验报告"), ("comparing", "计算用药对照"), ("future_stage", "等待阶段详情")])
|
||
def test_all_progress_stages_have_safe_chinese_fallback(stage: str, expected: str) -> None:
|
||
view = ai.progress_view({"status": "running", "progress": {"stage": stage, "stage_label": "<img src=x>", "elapsed_seconds": None}})
|
||
assert expected in view.headline
|
||
assert "future_stage" not in view.headline and "<img" not in view.headline
|
||
assert "0%" not in view.headline and view.total is None
|
||
|
||
|
||
@pytest.mark.parametrize("counts", [(None, None), (-1, 5), (3, 0), (8, 5), (True, 5), (1, 2**40), ("2", "5")])
|
||
def test_invalid_progress_counts_do_not_become_zero_or_overflow(counts: tuple[Any, Any]) -> None:
|
||
view = ai.progress_view({"status": "running", "progress": {"stage": "files", "completed_units": counts[0], "total_units": counts[1]}})
|
||
assert view.completed is None and view.total is None and view.busy
|
||
|
||
|
||
def test_old_api_and_unknown_progress_fields_remain_readable(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
repository.batches[0]["models"]["openai"]["progress"] = {"stage": "future_stage", "future_field": "untrusted", "server_time": 2000, "updated_at": 0}
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
assert dialog.model_views["qwen"]["progress_title"].text() == "已完成"
|
||
assert "等待阶段详情" in dialog.model_views["openai"]["progress_title"].text()
|
||
assert "阶段更新" not in dialog.model_views["openai"]["progress_detail"].text()
|
||
repository.batches[0]["models"]["openai"].pop("progress")
|
||
dialog._poll()
|
||
assert "服务端暂未提供分阶段进度" in dialog.model_views["openai"]["progress_detail"].text()
|
||
dialog.close()
|
||
|
||
|
||
def test_partial_completion_stops_both_timers_and_keeps_retry_target(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
value = progress_batch()
|
||
value["status"] = "partial"
|
||
value["models"]["qwen"].update(status="success", report={"summary": "已保存结果"})
|
||
value["models"]["openai"].update(status="failed", error_code="UPSTREAM_TIMEOUT")
|
||
repository.batches = [value]
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
assert dialog.model_views["qwen"]["progress_title"].text() == "已完成"
|
||
assert dialog.model_views["openai"]["progress_title"].text() == "处理失败"
|
||
assert "1/2 完成，1 个失败" in dialog.progress_flow.text()
|
||
assert "部分完成" in dialog.progress_flow.text()
|
||
assert not dialog._timer.isActive() and not dialog._progress_timer.isActive()
|
||
assert all(view["progress_bar"].maximum() != 0 for view in dialog.model_views.values())
|
||
assert dialog.model_views["openai"]["retry"].isEnabled()
|
||
assert not dialog.model_views["qwen"]["retry"].isEnabled()
|
||
dialog.close()
|
||
|
||
|
||
@pytest.mark.parametrize("validity", ["stale", "revoked", "superseded"])
|
||
def test_invalid_batch_progress_is_only_a_saved_record(application: QApplication, immediate: None, validity: str) -> None:
|
||
repository = Repository()
|
||
value = progress_batch()
|
||
value["validity"] = validity
|
||
repository.batches = [value]
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
assert "上次记录" in dialog.model_views["openai"]["progress_title"].text()
|
||
assert "已失效" in dialog.progress_flow.text()
|
||
assert not dialog._timer.isActive() and not dialog._progress_timer.isActive()
|
||
assert dialog.model_views["openai"]["progress_bar"].maximum() != 0
|
||
dialog.close()
|
||
|
||
|
||
def test_poll_failure_marks_cached_progress_and_refresh_recovers_without_losing_draft(application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
ticks = [100.0]
|
||
monkeypatch.setattr(ai, "monotonic", lambda: ticks[0])
|
||
repository = Repository()
|
||
value = progress_batch()
|
||
value["models"]["qwen"].update(status="success", report={"summary": "可供复核的已完成报告"})
|
||
repository.batches = [value]
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
dialog.model_views["qwen"]["comment"].setPlainText("尚未保存的复核意见")
|
||
ticks[0] = 120.0
|
||
def offline(_batch_id: int) -> None:
|
||
raise RuntimeError("connection refused")
|
||
monkeypatch.setattr(repository, "get_prescription_ai_report", offline)
|
||
dialog._poll()
|
||
assert "上次读取的数据" in dialog.message.text()
|
||
assert "上次记录" in dialog.model_views["openai"]["progress_title"].text()
|
||
assert "进度同步已暂停" in dialog.model_views["openai"]["progress_detail"].text()
|
||
assert not dialog._timer.isActive() and not dialog._progress_timer.isActive()
|
||
frozen = dialog.model_views["openai"]["progress_detail"].text()
|
||
ticks[0] = 200.0
|
||
dialog._render_progress()
|
||
assert dialog.model_views["openai"]["progress_detail"].text() == frozen
|
||
assert "可供复核" in dialog.model_views["qwen"]["report"].toPlainText()
|
||
assert not dialog.model_views["qwen"]["save"].isEnabled()
|
||
monkeypatch.setattr(repository, "get_prescription_ai_report", Repository.get_prescription_ai_report.__get__(repository))
|
||
dialog.refresh()
|
||
assert dialog._timer.isActive() and dialog._progress_timer.isActive()
|
||
assert "上次记录" not in dialog.model_views["openai"]["progress_title"].text()
|
||
assert "失败" not in dialog.message.text()
|
||
assert dialog.model_views["qwen"]["comment"].toPlainText() == "尚未保存的复核意见"
|
||
assert dialog.model_views["qwen"]["save"].isEnabled()
|
||
dialog.close()
|
||
|
||
|
||
def test_access_revocation_clears_reports_and_all_progress(application: QApplication, immediate: None) -> None:
|
||
from doctor_workstation.core.errors import AuthenticationExpiredError
|
||
repository = Repository()
|
||
repository.batches = [progress_batch()]
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
dialog._detail_error(AuthenticationExpiredError("Expired"), dialog._generation)
|
||
assert dialog._batch == {}
|
||
assert not dialog._timer.isActive() and not dialog._progress_timer.isActive()
|
||
assert all(view["progress_bar"].maximum() != 0 for view in dialog.model_views.values())
|
||
assert not dialog.model_views["qwen"]["report"].toPlainText()
|
||
assert not dialog.model_views["qwen"]["comment"].toPlainText()
|
||
assert "登录" in dialog.message.text()
|
||
assert not dialog.regenerate_button.isEnabled()
|
||
dialog.close()
|
||
|
||
|
||
def test_progress_updates_preserve_report_document_scroll_and_selection(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
value = progress_batch()
|
||
value["models"]["qwen"].update(status="success", report={"summary": "\n".join(f"报告第 {index} 行" for index in range(180))})
|
||
repository.batches = [value]
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
application.processEvents()
|
||
report = dialog.model_views["qwen"]["report"]
|
||
report.verticalScrollBar().setValue(300)
|
||
scroll = report.verticalScrollBar().value()
|
||
revision = report.document().revision()
|
||
changes = []
|
||
report.textChanged.connect(lambda: changes.append(True))
|
||
repository.batches[0]["models"]["openai"]["progress"]["stage_elapsed_seconds"] += 5
|
||
dialog._poll()
|
||
assert report.document().revision() == revision
|
||
assert report.verticalScrollBar().value() == scroll
|
||
assert changes == []
|
||
dialog.close()
|
||
|
||
|
||
def test_progress_notice_and_source_references_remain_plain_and_escaped(application: QApplication, immediate: None) -> None:
|
||
repository = Repository()
|
||
value = progress_batch()
|
||
attack = '<img src="https://example.invalid/tracker" onerror="alert(1)"><script>bad()</script>'
|
||
value["models"]["qwen"]["progress"]["notice"] = attack
|
||
value["models"]["qwen"]["coverage"] = {"source_ids": ["call_records:1820 " + attack]}
|
||
repository.batches = [value]
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
views = dialog.model_views["qwen"]
|
||
assert views["progress_detail"].textFormat() == Qt.TextFormat.PlainText
|
||
assert attack in views["progress_detail"].text()
|
||
assert "<script>" in views["sources"].toHtml()
|
||
assert "<img" not in ai.status_tooltip(value) and "<img" in ai.status_tooltip(value)
|
||
assert not views["sources"].openLinks() and not views["sources"].openExternalLinks()
|
||
dialog.close()
|
||
|
||
|
||
def test_list_shows_compact_two_model_progress_with_full_detail_in_tooltip(application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
repository = Repository()
|
||
monkeypatch.setattr(repository, "list_prescription_ai_statuses", lambda ids: {"enabled": True, "items": [dict(progress_batch(), prescription_id=value) for value in ids]})
|
||
page = page_module.PrescriptionsPage(repository, ["*"])
|
||
page.resize(1366, 800)
|
||
page.show()
|
||
application.processEvents()
|
||
item = page.table.item(0, 11)
|
||
assert item.text() == "千问 附件 2/5\nOpenAI 生成报告"
|
||
assert "本阶段 2/5 组" in item.toolTip()
|
||
assert "不代表附件全部读懂" in item.toolTip()
|
||
assert page.table.rowHeight(0) >= 2 * page.table.fontMetrics().height()
|
||
assert page._ai_timer.isActive()
|
||
page.close()
|
||
|
||
|
||
@pytest.mark.parametrize("status", ["success", "failed", "cancelled", "partial"])
|
||
def test_terminal_batch_overrides_obsolete_running_model_checkpoint(application: QApplication, immediate: None, status: str) -> None:
|
||
repository = Repository()
|
||
value = progress_batch()
|
||
value["status"] = status
|
||
repository.batches = [value]
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
assert not ai.batch_running(value)
|
||
assert not dialog._timer.isActive() and not dialog._progress_timer.isActive()
|
||
assert "最后记录" in dialog.model_views["openai"]["progress_title"].text()
|
||
assert dialog.model_views["openai"]["progress_bar"].maximum() != 0
|
||
dialog.close()
|
||
|
||
|
||
@pytest.mark.parametrize(("status", "notice"), [
|
||
("failed", "处理未完成，请查看失败原因。 上次进度：校验报告。 耗时按本次尝试计算。"),
|
||
("cancelled", "任务已取消。 耗时按本次尝试计算。"),
|
||
])
|
||
def test_terminal_progress_preserves_matching_server_failure_notice(status: str, notice: str) -> None:
|
||
value = {"status": status, "progress": {"stage": status, "phase": "failed", "notice": notice, "elapsed_seconds": 180, "attempt": 2}}
|
||
view = ai.progress_view(value, seconds=45)
|
||
assert notice in view.detail and "第 2 次尝试" in view.detail
|
||
assert "已用时 3 分 00 秒" in view.detail
|
||
assert not view.busy and view.total is None
|
||
assert notice in ai.status_tooltip({"models": {"qwen": value}})
|
||
|
||
|
||
@pytest.mark.parametrize("status", ["failed", "cancelled", "success"])
|
||
def test_terminal_progress_discards_obsolete_waiting_notice(status: str) -> None:
|
||
view = ai.progress_view({"status": status, "progress": {"stage": "final", "phase": "waiting", "notice": "等待模型返回。正在生成报告。", "elapsed_seconds": 90}}, seconds=45)
|
||
assert "等待模型返回" not in view.detail and "正在生成报告" not in view.detail
|
||
assert "已用时 1 分 30 秒" in view.detail
|
||
assert not view.busy
|
||
|
||
|
||
def test_retry_wait_freezes_attempt_duration_while_countdown_advances_and_poll_remains_consistent(application: QApplication, immediate: None, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
ticks = [100.0]
|
||
monkeypatch.setattr(ai, "monotonic", lambda: ticks[0])
|
||
repository = Repository()
|
||
value = batch(status="running")
|
||
value["models"]["openai"].update(status="retry_wait", progress={
|
||
"stage": "retry_wait", "phase": "waiting", "elapsed_seconds": 180, "stage_elapsed_seconds": 0,
|
||
"wait_remaining_seconds": 30, "updated_at": 1000, "server_time": 1010, "attempt": 2,
|
||
"notice": "本次未完成，已安排自动重试。 上次进度：校验报告。 耗时按本次尝试计算。",
|
||
})
|
||
repository.batches = [value]
|
||
dialog = ai.IssuedPrescriptionAiDialog(repository, ["*"], prescription_id=801)
|
||
dialog.show()
|
||
detail = dialog.model_views["openai"]["progress_detail"]
|
||
assert "第 2 次尝试" in detail.text() and "距下次重试 30 秒" in detail.text()
|
||
ticks[0] = 104.0
|
||
dialog._render_progress()
|
||
assert "已用时 3 分 00 秒" in detail.text() and "距下次重试 26 秒" in detail.text()
|
||
assert "阶段更新于 14 秒前" in detail.text()
|
||
assert "本阶段" not in detail.text()
|
||
progress = repository.batches[0]["models"]["openai"]["progress"]
|
||
progress.update(server_time=1015, wait_remaining_seconds=25)
|
||
ticks[0] = 105.0
|
||
dialog._poll()
|
||
assert "已用时 3 分 00 秒" in detail.text() and "距下次重试 25 秒" in detail.text()
|
||
assert "上次进度：校验报告。" in detail.text()
|
||
repository.batches[0]["models"]["openai"].update(status="running", progress={
|
||
"stage": "final", "phase": "waiting", "elapsed_seconds": 0, "stage_elapsed_seconds": 0,
|
||
"attempt": 3, "updated_at": 1040, "server_time": 1040,
|
||
})
|
||
ticks[0] = 130.0
|
||
dialog._poll()
|
||
ticks[0] = 133.0
|
||
dialog._render_progress()
|
||
assert "第 3 次尝试" in detail.text() and "已用时 3 秒" in detail.text()
|
||
assert "距下次重试" not in detail.text()
|
||
dialog.close()
|
||
|
||
|
||
@pytest.mark.parametrize("attempt", [None, 0, -1, True, "2", 1_000_001])
|
||
def test_unknown_or_invalid_attempt_count_is_not_invented(attempt: Any) -> None:
|
||
view = ai.progress_view({"status": "retry_wait", "progress": {"stage": "retry_wait", "attempt": attempt}})
|
||
assert "次尝试" not in view.detail
|