life-echo/api/tests/test_json_and_memory_utils.py

"""JSON 载荷解析、证据格式化、Story 批量规划校验（纯函数）。"""

import pytest

from app.agents.chat.reply_limits import truncate_chat_segments

from app.agents.memoir.classification_agent import _normalize_llm_category
from app.agents.memoir.prompts import format_evidence_chunks_for_prompt
from app.features.memory.evidence_format import (
    format_evidence_chunks_for_prompt as format_evidence_from_memory,
)
from app.agents.memoir.story_route_agent import (
    StoryBatchPlan,
    StoryBatchPlanUnit,
    validate_story_batch_plan,
)
from app.core.json_utils import extract_json_payload


def test_extract_json_payload_strips_markdown_fence() -> None:
    raw = """```json
{"a": 1}
```"""
    assert '"a"' in extract_json_payload(raw)


def test_extract_json_payload_balanced_nested_braces() -> None:
    raw = 'noise {"outer": {"inner": 1}} trailing'
    assert extract_json_payload(raw) == '{"outer": {"inner": 1}}'


def test_normalize_llm_category_strips_quotes() -> None:
    assert _normalize_llm_category('"childhood"') == "childhood"
    assert _normalize_llm_category("`beliefs`") == "beliefs"


def test_format_evidence_chunks_includes_timeline() -> None:
    ev = {
        "relevant_chunks": [{"content": "chunk1"}],
        "relevant_facts": [
            {"subject": "我", "predicate": "生于", "object_json": "1950"}
        ],
        "timeline_hints": [
            {
                "id": "1",
                "event_year": 1977,
                "event_date": None,
                "title": "恢复高考",
                "description": "参加了考试",
            }
        ],
        "relevant_summaries": [],
        "relevant_stories": [],
    }
    out = format_evidence_chunks_for_prompt(ev)
    assert "chunk1" in out
    assert "1950" in out or "生于" in out
    assert "1977" in out or "恢复高考" in out
    assert format_evidence_from_memory(ev) == out


def test_validate_story_batch_plan_ok() -> None:
    ordered = ["s1", "s2"]
    plan = StoryBatchPlan(
        units=[
            StoryBatchPlanUnit(
                segment_ids=["s1", "s2"],
                decision="new_story",
                target_story_id=None,
                new_story_title="标题",
                reason=None,
            )
        ]
    )
    ok, err = validate_story_batch_plan(ordered, plan, valid_story_ids=set())
    assert ok is True
    assert err is None


def test_truncate_chat_segments() -> None:
    out = truncate_chat_segments(
        ["a" * 300, "b"],
        max_segments=2,
        max_chars_per_segment=220,
    )
    assert out[0] == "a" * 219 + "…"
    assert len(out[0]) == 220
    assert out[1] == "b"


def test_validate_story_batch_plan_duplicate_segment() -> None:
    plan = StoryBatchPlan(
        units=[
            StoryBatchPlanUnit(
                segment_ids=["s1"],
                decision="new_story",
                target_story_id=None,
                new_story_title="A",
                reason=None,
            ),
            StoryBatchPlanUnit(
                segment_ids=["s1"],
                decision="new_story",
                target_story_id=None,
                new_story_title="B",
                reason=None,
            ),
        ]
    )
    ok, err = validate_story_batch_plan(["s1", "s1"], plan, valid_story_ids=set())
    assert ok is False
    assert err == "duplicate_segment"