cfc/cfc-langgraph/tests/test_health_plan.py

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


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(action_type="cook", title="做饭", dimension="body")


def test_plan_task_invalid_dimension():
    import pytest
    from pydantic import ValidationError
    with pytest.raises(ValidationError):
        PlanTask(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"