import os import sys import tempfile from pathlib import Path TEST_DB_PATH = Path(tempfile.gettempdir()) / "medical_agent_test_api_contract.db" TEST_DB_PATH.unlink(missing_ok=True) os.environ["DATABASE_URL"] = f"sqlite:///{TEST_DB_PATH.as_posix()}" os.environ["REPORT_STORAGE_DIR"] = str(Path(tempfile.gettempdir()) / "medical_agent_test_api_reports") os.environ.setdefault("RUNTIME_MEMORY_BACKEND", "memory") os.environ.setdefault("LLM_MOCK_ENABLED", "true") os.environ.setdefault("AUTH_USER_ME_URL", "http://django-user-center.test/api/user/users/me/") sys.path.insert(0, str(Path(__file__).resolve().parents[1])) try: from fastapi.testclient import TestClient except ImportError: TestClient = None def run_api_contract_tests() -> None: """API 合约：验证统一响应、user_id 校验、核心接口和跨用户隔离。""" if TestClient is None: print("api contract tests skipped: fastapi is not installed") return from app.main import app from app.services.external_auth_service import AuthenticatedUser, ExternalAuthService from scripts.init_demo_db import init_database async def fake_authenticate(self, request): # noqa: ARG001 """测试认证：模拟 Django `/me` 返回 200 后的标准用户解析结果。""" authorization = request.headers.get("Authorization") if not authorization: from app.core.exceptions import AppError raise AppError("AUTH_CREDENTIAL_REQUIRED", "Authorization header is required", 401) user_id = "api_user_002" if "api_user_002_token" in authorization else "api_user_001" return AuthenticatedUser( user_id=user_id, username=f"{user_id}_name", display_name="Swagger测试", role="student", tenant_id="1", status=1, profile={ "id": user_id, "username": f"{user_id}_name", "real_name": "Swagger测试", "role_type": "student", "institution": 1, "institution_name": "测试机构", "status": 1, }, ) ExternalAuthService.authenticate = fake_authenticate init_database() client = TestClient(app) headers = {"Authorization": "Bearer api_user_001_token", "X-Entry-Scene": "api_test"} live = client.get("/health/live") assert live.status_code == 200 assert live.json()["data"]["status"] == "live" missing_user = client.get("/api/v1/agent/hello") assert missing_user.status_code == 401 assert missing_user.json()["code"] == "AUTH_CREDENTIAL_REQUIRED" hello = client.get("/api/v1/agent/hello", headers=headers) assert hello.status_code == 200 assert hello.json()["code"] == "OK" auth_me = client.get("/api/v1/auth/me", headers=headers) assert auth_me.status_code == 200 assert auth_me.json()["data"]["user_id"] == "api_user_001" assert auth_me.json()["data"]["source"] == "django_user_center" assert auth_me.json()["data"]["display_name"] == "Swagger测试" assert "department_id" in auth_me.json()["data"] openapi = client.get("/openapi.json") assert openapi.status_code == 200 openapi_payload = openapi.json() auth_me_operation = openapi_payload["paths"]["/api/v1/auth/me"]["get"] assert any("HTTPBearer" in item for item in auth_me_operation.get("security", [])) assert "HTTPBearer" in openapi_payload["components"]["securitySchemes"] assert "/api/v1/imports/case-sql/preview" not in openapi_payload["paths"] assert "/api/v1/imports/case-sql/apply" not in openapi_payload["paths"] assert "/api/v1/cases/{case_id}/delete-preview" not in openapi_payload["paths"] assert "delete" not in openapi_payload["paths"]["/api/v1/cases/{case_id}"] assert "/api/v1/training-config/recommended" in openapi_payload["paths"] assert "/api/v1/training-config/options" in openapi_payload["paths"] assert "/api/v1/sessions/{session_id}/hints/stream" in openapi_payload["paths"] assert "/api/v1/sessions/{session_id}/physical-exams" in openapi_payload["paths"] assert "/api/v1/sessions/{session_id}/auxiliary-exams" in openapi_payload["paths"] assert "/api/v1/sessions/{session_id}/physical-exams/{item_code}" in openapi_payload["paths"] assert "/api/v1/sessions/{session_id}/auxiliary-exams/{item_code}" in openapi_payload["paths"] assert "/api/v1/evaluations/{evaluation_id}/download-pdf" in openapi_payload["paths"] cases = client.get("/api/v1/cases", headers=headers) assert cases.status_code == 200 case_id = cases.json()["data"]["items"][0]["id"] recommended_config = client.get(f"/api/v1/training-config/recommended?case_id={case_id}", headers=headers) assert recommended_config.status_code == 200 assert recommended_config.json()["data"]["recommended"]["visit_environment"] == "outpatient" assert recommended_config.json()["data"]["recommended"]["age_group"] == "child" assert recommended_config.json()["data"]["recommended_labels"]["age_group"] == "儿童" config_options = client.get(f"/api/v1/training-config/options?case_id={case_id}", headers=headers) assert config_options.status_code == 200 assert config_options.json()["data"]["options"]["personality"] recommended_session = client.post( "/api/v1/sessions", headers=headers, json={"case_id": case_id, "training_type": "diagnosis_treatment", "mode": "practice", "score_type": "percentage"}, ) assert recommended_session.status_code == 200 assert recommended_session.json()["data"]["patient_config"]["values"]["age_group"] == "child" created = client.post( "/api/v1/sessions", headers=headers, json={ "case_id": case_id, "training_type": "diagnosis_treatment", "mode": "practice", "score_type": "percentage", "patient_config": { "visit_environment": "outpatient", "age_group": "youth", "education_level": "higher", "personality": "calm", }, }, ) assert created.status_code == 200 session_id = created.json()["data"]["session_id"] assert created.json()["data"]["patient_config"]["labels"]["personality"] == "平和" with client.stream( "POST", f"/api/v1/sessions/{session_id}/chat/stream", headers=headers, json={"message": "孩子发热几天了？最高体温多少？"}, ) as chat_stream: assert chat_stream.status_code == 200 chat_stream_text = "".join(chat_stream.iter_text()) assert "event: message_delta" in chat_stream_text assert "event: message_done" in chat_stream_text invalid_config = client.post( "/api/v1/sessions", headers=headers, json={ "case_id": case_id, "training_type": "diagnosis_treatment", "mode": "practice", "score_type": "percentage", "patient_config": { "visit_environment": "invalid_scene", "age_group": "youth", "education_level": "higher", "personality": "calm", }, }, ) assert invalid_config.status_code == 422 cross_user = client.get( f"/api/v1/sessions/{session_id}/order-items", headers={"Authorization": "Bearer api_user_002_token", "X-Entry-Scene": "api_test"}, ) assert cross_user.status_code == 404 assert cross_user.json()["code"] == "SESSION_NOT_FOUND" invalid_order = client.post(f"/api/v1/sessions/{session_id}/orders", headers=headers, json={"item_code": "bad_item"}) assert invalid_order.status_code == 404 assert invalid_order.json()["code"] == "ORDER_ITEM_NOT_FOUND" order_one = client.post(f"/api/v1/sessions/{session_id}/orders", headers=headers, json={"item_code": "blood_routine"}) assert order_one.status_code == 200 assert order_one.json()["data"]["already_ordered"] is False order_two = client.post(f"/api/v1/sessions/{session_id}/orders", headers=headers, json={"item_code": "blood_routine"}) assert order_two.status_code == 200 assert order_two.json()["data"]["already_ordered"] is True physical_list = client.get(f"/api/v1/sessions/{session_id}/physical-exams", headers=headers) assert physical_list.status_code == 200 assert "items" in physical_list.json()["data"] assert any(item["item_code"] == "lung_auscultation" for item in physical_list.json()["data"]["items"]) physical_result = client.post(f"/api/v1/sessions/{session_id}/physical-exams/lung_auscultation", headers=headers) assert physical_result.status_code == 200 assert physical_result.json()["data"]["item_type"] == "physical_exam" physical_mismatch = client.post(f"/api/v1/sessions/{session_id}/physical-exams/blood_routine", headers=headers) assert physical_mismatch.status_code == 400 assert physical_mismatch.json()["code"] == "ORDER_ITEM_TYPE_MISMATCH" auxiliary_list = client.get(f"/api/v1/sessions/{session_id}/auxiliary-exams", headers=headers) assert auxiliary_list.status_code == 200 assert any(item["item_code"] == "blood_routine" for item in auxiliary_list.json()["data"]["items"]) auxiliary_result = client.post(f"/api/v1/sessions/{session_id}/auxiliary-exams/blood_routine", headers=headers) assert auxiliary_result.status_code == 200 assert auxiliary_result.json()["data"]["already_ordered"] is True completed = client.post(f"/api/v1/sessions/{session_id}/complete-inquiry", headers=headers) assert completed.status_code == 200 assert completed.json()["data"]["status"] == "diagnosis" submitted_diagnosis = client.post( f"/api/v1/sessions/{session_id}/diagnosis", headers=headers, json={ "primary_diagnosis": "支气管肺炎", "differential_diagnoses": ["毛细支气管炎", "支气管哮喘急性发作"], "diagnosis_basis": "发热、咳嗽、喘息，肺部湿啰音，胸片和炎症指标支持肺部感染。", }, ) assert submitted_diagnosis.status_code == 200 assert submitted_diagnosis.json()["data"]["status"] == "treatment" submitted_treatment = client.post( f"/api/v1/sessions/{session_id}/treatment", headers=headers, json={ "treatment_principle": "抗感染、止咳平喘、改善氧合并观察病情变化。", "treatment_measures": "根据病情选择抗感染治疗，必要时雾化吸入，监测体温、呼吸和血氧。", "risk_plan": "关注低氧、呼吸困难加重、持续高热和精神反应差。", "communication": "向家属说明病情、用药注意事项和复诊指征。", "follow_up": "治疗后复查体温、呼吸、血氧和必要炎症指标。", }, ) assert submitted_treatment.status_code == 200 assert submitted_treatment.json()["data"]["status"] == "evaluating" evaluation = client.post(f"/api/v1/sessions/{session_id}/evaluation", headers=headers, json={"score_type": "percentage"}) assert evaluation.status_code == 200 evaluation_id = evaluation.json()["data"]["evaluation_id"] assert evaluation.json()["data"]["score_details"] detail = client.get(f"/api/v1/evaluations/{evaluation_id}", headers=headers) assert detail.status_code == 200 assert detail.json()["data"]["evaluation_id"] == evaluation_id cross_user_detail = client.get( f"/api/v1/evaluations/{evaluation_id}", headers={"Authorization": "Bearer api_user_002_token", "X-Entry-Scene": "api_test"}, ) assert cross_user_detail.status_code == 404 assert cross_user_detail.json()["code"] == "EVALUATION_NOT_FOUND" pdf = client.post(f"/api/v1/evaluations/{evaluation_id}/export-pdf", headers=headers) assert pdf.status_code == 200 assert pdf.json()["data"]["file_path"] pdf_download = client.get(f"/api/v1/evaluations/{evaluation_id}/download-pdf", headers=headers) assert pdf_download.status_code == 200 assert pdf_download.headers["content-type"].startswith("application/pdf") assert "attachment" in pdf_download.headers.get("content-disposition", "") assert pdf_download.content.startswith(b"%PDF") cross_user_pdf = client.post( f"/api/v1/evaluations/{evaluation_id}/export-pdf", headers={"Authorization": "Bearer api_user_002_token", "X-Entry-Scene": "api_test"}, ) assert cross_user_pdf.status_code == 404 assert cross_user_pdf.json()["code"] == "EVALUATION_NOT_FOUND" cross_user_pdf_download = client.get( f"/api/v1/evaluations/{evaluation_id}/download-pdf", headers={"Authorization": "Bearer api_user_002_token", "X-Entry-Scene": "api_test"}, ) assert cross_user_pdf_download.status_code == 404 assert cross_user_pdf_download.json()["code"] == "EVALUATION_NOT_FOUND" practice_hint_session = client.post( "/api/v1/sessions", headers=headers, json={"case_id": case_id, "training_type": "diagnosis_treatment", "mode": "practice", "score_type": "percentage"}, ) assert practice_hint_session.status_code == 200 practice_hint_session_id = practice_hint_session.json()["data"]["session_id"] hint = client.post( f"/api/v1/sessions/{practice_hint_session_id}/hints", headers=headers, json={"scope": "current_conversation"}, ) assert hint.status_code == 200 assert hint.json()["data"]["hints"] assert "recommended_orders" in hint.json()["data"] with client.stream( "POST", f"/api/v1/sessions/{practice_hint_session_id}/hints/stream", headers=headers, json={"scope": "current_conversation"}, ) as hint_stream: assert hint_stream.status_code == 200 hint_stream_text = "".join(hint_stream.iter_text()) assert "event: hint_delta" in hint_stream_text assert "event: hint_done" in hint_stream_text teaching = client.post( "/api/v1/sessions", headers=headers, json={"case_id": case_id, "training_type": "diagnosis_treatment", "mode": "teaching", "score_type": "percentage"}, ) assert teaching.status_code == 200 teaching_hint = client.post( f"/api/v1/sessions/{teaching.json()['data']['session_id']}/hints", headers=headers, json={"scope": "current_conversation"}, ) assert teaching_hint.status_code == 400 assert teaching_hint.json()["code"] == "SESSION_STATUS_INVALID" llm_fast = client.post("/api/v1/llm/test/deepseek-fast", headers=headers, json={"message": "hello"}) assert llm_fast.status_code == 200 assert llm_fast.json()["code"] == "OK" assert llm_fast.json()["data"]["stream"] is False llm_reason = client.post("/api/v1/llm/test/deepseek-reason", headers=headers, json={"message": "hello"}) assert llm_reason.status_code == 200 assert llm_reason.json()["code"] == "OK" assert "total_latency_ms" in llm_reason.json()["data"] if __name__ == "__main__": run_api_contract_tests() print("api contract tests passed")