fastapi/tests/test_api_contract.py

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.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"]

    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_labels"]["visit_environment"] == "门诊"

    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"]

    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"] == "平和"

    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"]

    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

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