cfc/cfc-langgraph/tests/test_health_plan.py

from app.models.health_plan import PlanTask, HealthPlanSection, HealthPlanResponse, MultiMemberPlanResponse, PlanPart
from app.api.adapter import _parse_plan_response, _fallback_extract_tasks, _synthesize_multi_member_plan


def test_plan_task_valid():
    t = PlanTask(action_type="buy", title="购买维生素D3", dimension="wealth", frequency="once")
    assert t.frequency == "once"
    assert t.notes is None


def test_plan_task_default_frequency_daily():
    t = PlanTask(action_type="exercise", title="每周跑步3次", dimension="body")
    assert t.frequency == "daily"


def test_plan_task_invalid_action_type():
    import pytest
    from pydantic import ValidationError
    with pytest.raises(ValidationError):
        PlanTask.model_validate({"action_type": "cook", "title": "做饭", "dimension": "body"})


def test_plan_task_invalid_dimension():
    import pytest
    from pydantic import ValidationError
    with pytest.raises(ValidationError):
        PlanTask.model_validate({"action_type": "diet", "title": "少油少盐", "dimension": "earth"})


def test_plan_response_with_tasks():
    resp = HealthPlanResponse(
        overview="总览",
        sections=[HealthPlanSection(
            key="nutrition", title="营养补充", content="## 建议",
            tasks=[PlanTask(action_type="buy", title="购买鱼油", dimension="wealth", frequency="once")]
        )]
    )
    assert resp.sections[0].tasks[0].action_type == "buy"
    # 序列化后 tasks 字段存在
    data = resp.model_dump()
    assert "tasks" in data["sections"][0]


def test_parse_plan_response_valid():
    answer = '{"overview":"o","sections":[{"key":"diet","title":"t","content":"c","items":[],"tasks":[{"action_type":"diet","title":"少油少盐","dimension":"body","frequency":"daily"}]}],"abnormal_indicators":[]}'
    res = _parse_plan_response(answer)
    assert res["success"] is True
    assert res["data"]["sections"][0]["tasks"][0]["action_type"] == "diet"


def test_parse_plan_response_invalid_action_falls_back():
    # LLM 输出含非法 action_type（"cook"），校验失败但可被 fallback 处理
    answer = '{"overview":"o","sections":[{"key":"diet","title":"t","content":"1. 少油少盐\\n2. 多吃蔬菜\\n3. 纯原理描述无动作","items":[],"tasks":[{"action_type":"cook","title":"做饭","dimension":"body"}]}],"abnormal_indicators":[]}'
    res = _parse_plan_response(answer)
    # fallback 从 content 提取动作行
    assert res["success"] is True
    tasks = res["data"]["sections"][0].get("tasks", [])
    assert len(tasks) >= 2


def test_parse_plan_response_no_json():
    res = _parse_plan_response("这是纯文本没有 JSON")
    assert res["success"] is False


def test_fallback_skips_when_tasks_exist():
    parsed = {"sections": [{"key": "diet", "title": "t", "content": "1. 少油少盐", "tasks": [{"action_type": "diet", "title": "已有", "dimension": "body"}]}]}
    assert _fallback_extract_tasks(parsed) is None


def test_regenerate_tasks_validation_filter():
    raw_tasks = [
        {"action_type": "diet", "title": "少油少盐", "dimension": "body", "frequency": "daily"},
        {"action_type": "cook", "title": "非法", "dimension": "body"},  # 非法 action，应被丢弃
    ]
    valid = []
    for t in raw_tasks:
        try:
            valid.append(PlanTask.model_validate(t).model_dump())
        except Exception:
            continue
    assert len(valid) == 1
    assert valid[0]["action_type"] == "diet"


def test_multi_member_plan_response_structure():
    """多人方案结构校验：common + individual 按 memberId 分组"""
    resp = MultiMemberPlanResponse()
    assert resp.common.overview == ""
    assert resp.individual == {}


def test_multi_member_plan_response_with_data():
    resp = MultiMemberPlanResponse(
        common=PlanPart(overview="全家共性方案", sections=[
            HealthPlanSection(
                key="diet", title="饮食建议", content="## 共同饮食",
                tasks=[PlanTask(action_type="diet", title="每日三餐均衡", dimension="body", frequency="daily")]
            )
        ]),
        individual={
            "123": PlanPart(overview="成员A个性方案", sections=[
                HealthPlanSection(
                    key="nutrition", title="营养补充", content="## 维生素D3",
                    tasks=[PlanTask(action_type="buy", title="购买维生素D3", dimension="wealth", frequency="once")]
                )
            ])
        }
    )
    data = resp.model_dump()
    assert data["common"]["sections"][0]["tasks"][0]["action_type"] == "diet"
    assert data["individual"]["123"]["sections"][0]["tasks"][0]["action_type"] == "buy"
    assert "123" in data["individual"]


def test_multi_member_plan_response_fallback_structure():
    """LLM 合成失败时程序化兜底：common 置空，individual 透传各成员方案"""
    resp = MultiMemberPlanResponse(common=PlanPart(overview=""), individual={
        "456": PlanPart(overview="成员B个性", sections=[
            HealthPlanSection(key="exercise", title="运动计划", content="## 跑步",
                             tasks=[PlanTask(action_type="exercise", title="每周跑步3次", dimension="body", frequency="daily")])
        ])
    })
    data = resp.model_dump()
    assert data["common"]["overview"] == ""
    assert data["individual"]["456"]["sections"][0]["tasks"][0]["title"] == "每周跑步3次"