test_intake_api_e2e.py

#!/usr/bin/env python3
"""API-based intake end-to-end regression (no external network dependency)."""

from __future__ import annotations

import json
import os
import time
from typing import Any

# Keep regression deterministic and offline.
os.environ["LLM_ENABLED"] = "false"
os.environ["RISK_LLM_ENABLED"] = "false"

from fastapi.testclient import TestClient

from api.server import app


def _pick_option_ids(question: dict[str, Any]) -> list[str]:
    options = list(question.get("options") or [])
    if not options:
        return []
    if question.get("allow_multiple"):
        return [str(opt.get("option_id")) for opt in options[:2] if opt.get("option_id")]
    first = options[0].get("option_id")
    return [str(first)] if first else []


def _default_value_by_type(field_type: str) -> Any:
    t = field_type.strip().lower()
    if t == "number":
        return 1
    if t == "date":
        return "2026-01-01"
    if t in {"single_select", "multi_select"}:
        return None
    return "API回归自动填充"


def _build_structured_answer(question: dict[str, Any]) -> dict[str, Any]:
    input_type = str(question.get("input_type") or "single_select").strip().lower()
    payload: dict[str, Any] = {}

    if input_type in {"text", "date"}:
        payload["value"] = "API回归输入" if input_type == "text" else "2026-01-01"
        return payload

    if input_type == "composite":
        for idx, field in enumerate(question.get("sub_fields") or [], start=1):
            key = str(field.get("key") or f"field_{idx}")
            f_type = str(field.get("type") or "text")
            options = list(field.get("options") or [])
            if f_type == "single_select" and options:
                payload[key] = str(options[0].get("option_id"))
            elif f_type == "multi_select" and options:
                payload[key] = [str(options[0].get("option_id"))]
            else:
                payload[key] = _default_value_by_type(f_type)
        return payload

    if input_type == "matrix":
        matrix = question.get("matrix") or {}
        rows = list(matrix.get("rows") or [])
        cols = list(matrix.get("columns") or [])
        row_payload: dict[str, Any] = {}
        for r_idx, row in enumerate(rows, start=1):
            row_key = str(row.get("row_id") or f"row_{r_idx}")
            col_payload: dict[str, Any] = {}
            for c_idx, col in enumerate(cols, start=1):
                col_key = str(col.get("column_key") or f"col_{c_idx}")
                c_type = str(col.get("type") or "text")
                options = list(col.get("options") or [])
                if c_type == "single_select" and options:
                    col_payload[col_key] = str(options[0].get("option_id"))
                elif c_type == "multi_select" and options:
                    col_payload[col_key] = [str(options[0].get("option_id"))]
                else:
                    col_payload[col_key] = _default_value_by_type(c_type)
            if col_payload:
                row_payload[row_key] = col_payload
        if row_payload:
            payload["matrix"] = row_payload
        return payload

    return payload


def _build_answer_payload(question: dict[str, Any]) -> dict[str, Any]:
    selected_option_ids = _pick_option_ids(question)
    structured_answer = _build_structured_answer(question)
    return {
        "selected_option_ids": selected_option_ids,
        "free_text": "API端到端回归补充说明",
        "structured_answer": structured_answer or None,
    }


def main() -> int:
    intake_mode = (os.environ.get("INTAKE_MODE") or "quick").strip().lower()
    if intake_mode not in {"quick", "full"}:
        intake_mode = "quick"
    summary: dict[str, Any] = {
        "ok": False,
        "scenario": "intake_e2e_api_local",
        "intake_mode": intake_mode,
        "steps": [],
    }
    started_at = time.time()

    with TestClient(app) as client:
        start_resp = client.post(
            "/api/v1/intake/start",
            json={
                "name": "首检API回归测试",
                "age": 45,
                "gender": "2",
                "chief_concern": "API链路回归",
                "intake_mode": intake_mode,
            },
        )
        if start_resp.status_code != 200:
            summary["error"] = f"start failed: {start_resp.status_code}"
            summary["body"] = start_resp.text
            print(json.dumps(summary, ensure_ascii=False, indent=2))
            return 1

        start_data = start_resp.json()
        session_id = start_data["session_id"]
        max_rounds = int(start_data.get("max_rounds") or 0)
        question = start_data["question"]
        asked_question_ids = [question["question_id"]]
        summary["first_question_id"] = question["question_id"]
        summary["max_rounds"] = max_rounds
        expected_quick_rounds = 9  # female age 45 triggers S8 conditional insertion in quick mode
        if intake_mode == "quick" and max_rounds != expected_quick_rounds:
            summary["error"] = f"quick mode max_rounds expected {expected_quick_rounds}, got {max_rounds}"
            print(json.dumps(summary, ensure_ascii=False, indent=2))
            return 1
        if intake_mode == "full" and max_rounds <= 8:
            summary["error"] = f"full mode max_rounds expected >8, got {max_rounds}"
            print(json.dumps(summary, ensure_ascii=False, indent=2))
            return 1

        last_answer_count = 0
        is_processing = False
        is_finished = False
        final_data: dict[str, Any] | None = None

        max_steps = max(16, max_rounds + 2)
        for step in range(1, max_steps + 1):
            payload = _build_answer_payload(question)
            answer_resp = client.post(f"/api/v1/intake/answer/{session_id}", json=payload)
            if answer_resp.status_code != 200:
                summary["error"] = f"answer failed at step {step}: {answer_resp.status_code}"
                summary["body"] = answer_resp.text
                print(json.dumps(summary, ensure_ascii=False, indent=2))
                return 1

            data = answer_resp.json()
            history = list(data.get("answer_history") or [])
            latest = data.get("latest_answer")
            current_count = len(history)

            summary["steps"].append(
                {
                    "step": step,
                    "question_id": question.get("question_id"),
                    "input_type": question.get("input_type"),
                    "selected_count": len(payload.get("selected_option_ids") or []),
                    "has_structured": bool(payload.get("structured_answer")),
                    "echo_has_latest": bool(latest is not None),
                    "echo_history_count": current_count,
                    "is_processing": bool(data.get("is_processing")),
                    "is_finished": bool(data.get("is_finished")),
                }
            )

            if current_count <= last_answer_count:
                summary["error"] = "answer_history did not grow monotonically"
                print(json.dumps(summary, ensure_ascii=False, indent=2))
                return 1
            last_answer_count = current_count

            asked_question_ids = list(data.get("asked_question_ids") or asked_question_ids)
            is_processing = bool(data.get("is_processing"))
            is_finished = bool(data.get("is_finished"))

            if is_finished:
                final_data = data
                break
            if is_processing:
                break

            next_question = data.get("next_question")
            if not next_question:
                summary["error"] = "next_question missing before processing/finish"
                print(json.dumps(summary, ensure_ascii=False, indent=2))
                return 1
            question = next_question

        if is_processing and not is_finished:
            for _ in range(120):
                poll_resp = client.get(f"/api/v1/intake/result/{session_id}")
                if poll_resp.status_code != 200:
                    summary["error"] = f"poll failed: {poll_resp.status_code}"
                    summary["body"] = poll_resp.text
                    print(json.dumps(summary, ensure_ascii=False, indent=2))
                    return 1
                data = poll_resp.json()
                history = list(data.get("answer_history") or [])
                if len(history) < last_answer_count:
                    summary["error"] = "poll answer_history unexpectedly shrank"
                    print(json.dumps(summary, ensure_ascii=False, indent=2))
                    return 1
                last_answer_count = len(history)
                if data.get("is_finished"):
                    final_data = data
                    is_finished = True
                    break
                if data.get("processing_error"):
                    summary["error"] = f"processing_error: {data.get('processing_error')}"
                    print(json.dumps(summary, ensure_ascii=False, indent=2))
                    return 1
                time.sleep(0.2)

        if not is_finished or not final_data:
            summary["error"] = "did not reach finished state"
            print(json.dumps(summary, ensure_ascii=False, indent=2))
            return 1

        final_recommendation = final_data.get("final_recommendation") or {}
        recommendations = list(final_recommendation.get("recommendations") or [])
        recommendation_groups = final_recommendation.get("recommendation_groups") or {}
        if not isinstance(recommendation_groups, dict):
            summary["error"] = "recommendation_groups should be an object"
            print(json.dumps(summary, ensure_ascii=False, indent=2))
            return 1
        for field in ("must_do", "recommended", "optional", "alternatives"):
            if field not in recommendation_groups or not isinstance(recommendation_groups.get(field), list):
                summary["error"] = f"recommendation_groups missing list field: {field}"
                summary["recommendation_groups"] = recommendation_groups
                print(json.dumps(summary, ensure_ascii=False, indent=2))
                return 1
        trace_cards = final_recommendation.get("evidence_trace_cards")
        if not isinstance(trace_cards, list):
            summary["error"] = "evidence_trace_cards should be list"
            print(json.dumps(summary, ensure_ascii=False, indent=2))
            return 1
        if recommendations and len(trace_cards) != len(recommendations):
            summary["error"] = "evidence_trace_cards count should equal recommendations count"
            summary["trace_count"] = len(trace_cards)
            summary["recommendation_count"] = len(recommendations)
            print(json.dumps(summary, ensure_ascii=False, indent=2))
            return 1
        recommended_names = [str(item.get("recommended_check_item")) for item in recommendations if item.get("recommended_check_item")]
        if "健康体检自测问卷" not in recommended_names:
            summary["error"] = "mandatory table1 item missing: 健康体检自测问卷"
            summary["first_10_recommendations"] = recommended_names[:10]
            print(json.dumps(summary, ensure_ascii=False, indent=2))
            return 1

        knowledge_manifest = list(final_recommendation.get("knowledge_manifest") or [])
        if not knowledge_manifest:
            summary["error"] = "knowledge_manifest missing in final_recommendation"
            print(json.dumps(summary, ensure_ascii=False, indent=2))
            return 1
        if not all(str(item.get("version") or "").strip() and str(item.get("effective_date") or "").strip() for item in knowledge_manifest):
            summary["error"] = "knowledge_manifest has empty version/effective_date"
            summary["knowledge_manifest"] = knowledge_manifest
            print(json.dumps(summary, ensure_ascii=False, indent=2))
            return 1

        kb_resp = client.get("/api/v1/knowledge/manifest")
        if kb_resp.status_code != 200:
            summary["error"] = f"knowledge manifest api failed: {kb_resp.status_code}"
            summary["body"] = kb_resp.text
            print(json.dumps(summary, ensure_ascii=False, indent=2))
            return 1

        metrics_resp = client.get("/api/v1/intake/metrics")
        if metrics_resp.status_code != 200:
            summary["error"] = f"intake metrics api failed: {metrics_resp.status_code}"
            summary["body"] = metrics_resp.text
            print(json.dumps(summary, ensure_ascii=False, indent=2))
            return 1
        metrics_json = metrics_resp.json()
        if int(metrics_json.get("started_sessions") or 0) < 1:
            summary["error"] = "intake metrics started_sessions should be >=1"
            summary["metrics"] = metrics_json
            print(json.dumps(summary, ensure_ascii=False, indent=2))
            return 1

        public_config_resp = client.get("/api/v1/public/config")
        if public_config_resp.status_code != 200:
            summary["error"] = f"public config api failed: {public_config_resp.status_code}"
            summary["body"] = public_config_resp.text
            print(json.dumps(summary, ensure_ascii=False, indent=2))
            return 1
        public_config = public_config_resp.json()
        if "urgent_on_any_high_risk" not in public_config:
            summary["error"] = "public config missing urgent_on_any_high_risk"
            summary["public_config"] = public_config
            print(json.dumps(summary, ensure_ascii=False, indent=2))
            return 1

        summary.update(
            {
                "ok": True,
                "elapsed_sec": round(time.time() - started_at, 2),
                "session_id": session_id,
                "asked_question_ids": asked_question_ids,
                "answer_history_count": last_answer_count,
                "risk_count": len(final_recommendation.get("risks") or []),
                "recommendation_count": len(recommendations),
                "first_10_recommendations": recommended_names[:10],
                "extraction_mode": ((final_recommendation.get("narrative_insight") or {}).get("extraction_mode") or "rule"),
                "knowledge_manifest_count": len(knowledge_manifest),
                "metrics_completion_rate": metrics_json.get("completion_rate"),
                "urgent_on_any_high_risk": bool(public_config.get("urgent_on_any_high_risk")),
            }
        )

    print(json.dumps(summary, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())