"""Six task actions reach the real repository and preserve mocked model answers. Finite offscreen Qt tests only. HTTP terminates in MockTransport; all clinical text and transfer metadata below are synthetic, including incomplete sources. """ from __future__ import annotations import json import os from time import monotonic os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") import httpx import pytest from PySide6.QtCore import QEvent, Qt, QThreadPool, QTimer from PySide6.QtTest import QTest from PySide6.QtWidgets import QApplication, QLabel, QTextBrowser from doctor_workstation.core import PermissionSet from doctor_workstation.services import ApiClient, RemoteDoctorRepository from doctor_workstation.services.repository import _diagnosis_ai_request from doctor_workstation.ui.dialogs import ai_consult as module from doctor_workstation.ui.dialogs.prescription_ai import diagnosis_ai_task # Explicit acceptance values; neither tasks nor answer content are derived from # the implementation's classifier or tool-task mapping. TOOLS = ( ("分析病历", "case_analysis", "病历分析", "已有口渴记录，结合血糖异常可分析津液受损的可能；不能据此直接确立证型。"), ("血糖趋势", "glucose_trend", "血糖趋势", "已有一条空腹血糖 8.2 mmol/L；单点不能确定升降趋势，应结合日期与连续监测解释。"), ("用药评估", "medication_review", "用药评估", "已记录二甲双胍，但剂量与肾功能缺失；可提示胃肠不适监测，不能据此增加剂量。"), ("并发症风险", "complication_risk", "并发症风险", "糖尿病病史提示需关注眼、肾和足部风险；未记录相应筛查不能推定已患或排除并发症。"), ("生成问诊小结", "summary", "问诊小结", "本次主诉口渴，既往有糖尿病；已记录二甲双胍及一条空腹血糖，病程与检查日期待明确。"), ("健康建议", "lifestyle_advice", "健康建议", "可先规律分配主食并记录饮食与血糖；运动宜从适量步行开始，出现低血糖症状时应停止活动。"), ) SOURCE_KEYS = ( "basic_info", "vital_signs", "chief_complaint", "present_illness", "past_history", "other_history", "diagnosis_info", "doctor_image_notes", "doctor_file_notes", "daily_tracking", "prescriptions", "video_consultation_text", "chat", ) def source_receipt() -> dict: return { "transfer_manifest_final": True, "attachment_count": 0, "submitted_attachment_count": 0, "items": [], "required_sections": [ {"key": key, "availability": "present" if key in { "chief_complaint", "past_history", "prescriptions", "daily_tracking", } else "absent", "status": "submitted"} for key in SOURCE_KEYS ], } def answer_for(tool) -> str: _label, task, title, analysis = tool return ( f"当前{title}依据已记录的部分资料完成。\n\n" f"## {title}判断\n- {analysis}\n- 专属结果标记：{task}_ACTUAL_MODEL_ANSWER。\n\n" f"## {title}后续建议\n- 根据以上已有信息执行相应随访，不将缺项改写为正常。" "\n- 缺失资料仅影响相应结论，不取代上述分析。" ) def sse(kind: str, data: dict) -> str: return f"event: {kind}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" def wait_until(predicate, timeout: float = 4.0) -> None: deadline = monotonic() + timeout while not predicate() and monotonic() < deadline: QTest.qWait(5) assert predicate(), "finite mocked request did not finish" @pytest.fixture(scope="module") def application(): return QApplication.instance() or QApplication([]) @pytest.fixture def dialog_factory(application, monkeypatch): resources = [] def create(handler): client = ApiClient("https://example.test", transport=httpx.MockTransport(handler)) dialog = module.AiConsultDialog( RemoteDoctorRepository(client), PermissionSet(["tcm.diagnosis/aiAssistant"]), ) monkeypatch.setattr(dialog, "_load_workspace", lambda _: None) dialog.open_for(diagnosis_id=501, patient_id=1501) dialog._detail = {"diagnosis": {"id": 501, "patient_id": 1501}} # The intentionally absent case and errored local previews cannot block # inference, synthesize a result, or become a stale request context. dialog._workspace_sections = {"diagnosis": {"value": None, "error": "合成本地资料未加载"}} dialog._patient_ai_context = "不能发往服务的本地旧预览" dialog.show() application.processEvents() resources.append((dialog, client)) return dialog yield create for dialog, client in resources: assert QThreadPool.globalInstance().waitForDone(4000) dialog.close() for timer in dialog.findChildren(QTimer): timer.stop() client.close() dialog.deleteLater() application.sendPostedEvents(None, QEvent.Type.DeferredDelete) application.processEvents() def click_tool(dialog, label: str) -> None: cards = dialog.findChildren(module._ClickCard, "AiConsultToolCard") target = next(child for card in cards for child in card.findChildren(QLabel) if child.text() == label) QTest.mouseClick(target, Qt.MouseButton.LeftButton) def readable_answer(bubble) -> str: if bubble._clinical_panel is None: return bubble.body.toPlainText() return "\n".join(view.toPlainText() for view in bubble._clinical_panel.findChildren(QTextBrowser)) @pytest.mark.parametrize("tool", TOOLS, ids=[tool[1] for tool in TOOLS]) @pytest.mark.parametrize("route", ["stream", "legacy404"]) def test_incomplete_case_click_dispatches_distinct_task_and_shows_actual_task_answer( dialog_factory, tool, route, ): requests = [] answer = answer_for(tool) label, task, title, analysis = tool def handler(request): assert request.url.host == "example.test" requests.append(request) if request.url.path.endswith("/aiAssistantStream"): if route == "legacy404": return httpx.Response(404, json={"msg": "合成旧部署无流式路由"}) events = sse("start", {"model_key": "qwen", "model_label": "原配置千问"}) events += sse("progress", {"phase": "attachments", "status": "submitted", **source_receipt()}) events += sse("delta", {"text": answer[:35]}) events += sse("delta", {"text": answer[35:]}) events += sse("done", {"model_key": "qwen", "model_label": "原配置千问", **source_receipt()}) return httpx.Response(200, headers={"content-type": "text/event-stream"}, text=events) assert route == "legacy404" and request.url.path.endswith("/aiAssistant") return httpx.Response(200, json={"code": 1, "data": { "answer": answer, "model_key": "qwen", "model_label": "原配置千问", **source_receipt(), }}) dialog = dialog_factory(handler) click_tool(dialog, label) bubble = dialog._stream_bubble assert bubble is not None wait_until(lambda: dialog._stream_worker is None and not dialog._asking) assert len(requests) == (1 if route == "stream" else 2) expected_prompt = next(prompt for _, name, prompt in module.TOOL_ITEMS if name == label) for request in requests: assert request.method == "POST" body = json.loads(request.content) assert set(body) == ({"id", "prompt", "task", "request_id"} if request.url.path.endswith("/aiAssistantStream") else {"id", "prompt", "task"}) assert body["task"] == task and body["id"] == 501 assert body["prompt"] == expected_prompt assert "本地旧预览" not in request.content.decode() assert "model" not in body and "model_key" not in body # Server model selection remains authoritative. assert bubble._raw_payload == answer assert isinstance(bubble._clinical_panel, module._StructuredReportPanel) assert bubble._clinical_panel.findChild(QLabel, "AiConsultClinicalTitle").text() == f"AI · {title}" visible = readable_answer(bubble) assert analysis in visible and f"{task}_ACTUAL_MODEL_ANSWER" in visible assert [section.title for section in bubble._clinical_panel.sections] == [ f"{title}判断", f"{title}后续建议", ] assert len(dialog._stream_meta["required_sections"]) == 13 assert [item["key"] for item in dialog._stream_meta["required_sections"]] == list(SOURCE_KEYS) assert dialog._stream_meta["model_key"] == "qwen" and "原配置千问" in bubble.stamp.text() assert dialog._stream_completed and dialog.send_button.isEnabled() assert not any(timer.isActive() for timer in ( dialog._delivery_timer, dialog._stream_timeout_timer, dialog._stream_elapsed_timer, )) @pytest.mark.parametrize("tool", TOOLS, ids=[tool[1] for tool in TOOLS]) def test_each_tool_failed_model_request_is_honest_and_does_not_show_fake_success(dialog_factory, tool): requests = [] label, task, title, _analysis = tool def handler(request): assert request.url.host == "example.test" requests.append(request) events = sse("start", {"model_key": "qwen", "model_label": "原配置千问"}) events += sse("error", {"code": "SYNTHETIC_MODEL_FAILURE", "message": f"{title}合成模型失败"}) events += sse("done", {"answer": f"{task}_FAKE_LATE_SUCCESS"}) return httpx.Response(200, headers={"content-type": "text/event-stream"}, text=events) dialog = dialog_factory(handler) click_tool(dialog, label) bubble = dialog._stream_bubble wait_until(lambda: dialog._stream_worker is None and not dialog._asking) assert len(requests) == 1 and json.loads(requests[0].content)["task"] == task assert f"{title}合成模型失败" in bubble._raw_payload assert "FAKE_LATE_SUCCESS" not in bubble._raw_payload assert "ACTUAL_MODEL_ANSWER" not in bubble._raw_payload assert bubble._role_name.text() == f"AI · {title}" assert bubble._clinical_panel is None and not dialog._stream_completed assert dialog._stream_meta["code"] == "SYNTHETIC_MODEL_FAILURE" assert dialog.send_button.isEnabled() @pytest.mark.parametrize("tool", TOOLS, ids=[tool[1] for tool in TOOLS]) def test_task_classifier_and_real_repository_allowlist_keep_six_explicit_tasks(tool): label, expected_task, _title, _analysis = tool prompt = next(prompt for _, name, prompt in module.TOOL_ITEMS if name == label) assert diagnosis_ai_task(prompt) == expected_task assert _diagnosis_ai_request(501, prompt, expected_task) == (prompt, expected_task) assert len(prompt) <= 500 assert "舌脉" not in prompt and "脉象" not in prompt def test_unsupported_task_still_rejected(): with pytest.raises(ValueError, match="task is not supported"): _diagnosis_ai_request(501, "合成提问", "not_a_supported_task") @pytest.mark.parametrize("tool", TOOLS, ids=[tool[1] for tool in TOOLS]) def test_tool_answers_keep_all_points_and_extra_sections_in_old_clinical_heading_shape(application, tool): _label, task, title, _analysis = tool headings = ("证候分析", "信息缺口", "建议下一步", "风险提示", f"{title}专属补充") points = [f"{heading}要点{index}：" + "这段完整推理与建议需要保留原意和全部内容。" * 6 + f"{task}_{ordinal}_{index}_TAIL" for ordinal, heading in enumerate(headings) for index in range(5)] reply = "已按当前任务完成分析。\n\n" + "\n\n".join( f"## {heading}\n" + "\n".join(f"- {point}" for point in points[index * 5:(index + 1) * 5]) for index, heading in enumerate(headings) ) bubble = module._ChatBubble(role="ai", text=reply) try: bubble.set_analysis_task(task) assert bubble.finalize_clinical_analysis() assert isinstance(bubble._clinical_panel, module._StructuredReportPanel) assert bubble._raw_payload == reply assert [section.title for section in bubble._clinical_panel.sections] == list(headings) assert sum(len(section.items) for section in bubble._clinical_panel.sections) == 25 visible = readable_answer(bubble) for point in points: assert any(point in section.items for section in bubble._clinical_panel.sections) lead, body = point.split("：", 1) assert lead in visible and body in visible assert "当前更倾向" not in visible finally: bubble.close() bubble.deleteLater() application.sendPostedEvents(None, QEvent.Type.DeferredDelete) application.processEvents()