261 lines
9.6 KiB
Python
261 lines
9.6 KiB
Python
"""Story 路由：batch_transcript 仅含本批口述，不含 evidence（与 story_pipeline_sync 行为一致）。"""
|
||
|
||
# 与 alembic/env.py 一致：注册全部 ORM，避免 relationship 解析 KeyError
|
||
from types import SimpleNamespace
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
from app.agents.memoir.story_route_agent import StoryRouteDecision
|
||
from app.agents.state_schema import MemoirStateSchema
|
||
from app.features.asset import models as _asset_models # noqa: F401
|
||
from app.features.auth import models as _auth_models # noqa: F401
|
||
from app.features.conversation import models as _conv_models # noqa: F401
|
||
from app.features.memoir import models as _memoir_models # noqa: F401
|
||
from app.features.memoir.story_pipeline_sync import (
|
||
run_story_pipeline_for_category_batch,
|
||
)
|
||
from app.features.memory import models as _memory_models # noqa: F401
|
||
from app.features.memory.evidence_format import format_evidence_chunks_for_prompt
|
||
from app.features.payment import models as _payment_models # noqa: F401
|
||
from app.features.story import models as _story_models # noqa: F401
|
||
from app.features.user import models as _user_models # noqa: F401
|
||
|
||
|
||
def test_single_segment_decide_receives_only_combined_text_not_evidence() -> None:
|
||
"""路由输入不变量：decide 的 batch_transcript 等于 oral combined_text，且不含证据标记。"""
|
||
oral = "这是一条用于测试路由输入的用户口述内容足够长以避免叙事回退误判"
|
||
captured: dict[str, str] = {}
|
||
|
||
def decide_capture(
|
||
*,
|
||
batch_transcript: str,
|
||
**kwargs: object,
|
||
) -> StoryRouteDecision:
|
||
captured["batch_transcript"] = batch_transcript
|
||
return StoryRouteDecision(
|
||
decision="new_story",
|
||
new_story_title="测试新故事标题六个字以上",
|
||
reason="测试",
|
||
)
|
||
|
||
seg = SimpleNamespace(id="seg-route-test-1", user_input_text=oral)
|
||
evidence_payload = {
|
||
"relevant_chunks": [],
|
||
"relevant_summaries": [
|
||
{
|
||
"content": "滚动摘要里的旧内容不应出现在路由输入",
|
||
"summary_type": "rolling",
|
||
}
|
||
],
|
||
"relevant_facts": [{"subject": "X", "predicate": "y", "object_json": {}}],
|
||
"relevant_stories": [],
|
||
}
|
||
evidence_formatted = format_evidence_chunks_for_prompt(evidence_payload)
|
||
assert "[摘要:rolling]" in evidence_formatted
|
||
|
||
route_agent_mock = MagicMock()
|
||
|
||
with (
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.list_active_stories_for_user_sync",
|
||
return_value=[],
|
||
),
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.StoryRouteAgent",
|
||
return_value=route_agent_mock,
|
||
),
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.NarrativeAgent",
|
||
) as nac,
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.create_story_with_version_sync",
|
||
) as csw,
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.ensure_chapter_story_link_sync",
|
||
),
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.reorder_chapter_story_links_by_life_order_sync",
|
||
),
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.mark_chapter_dirty_sync",
|
||
),
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.chapter_needs_cover_enqueue",
|
||
return_value=False,
|
||
),
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.MemoirImageSettings",
|
||
) as mis,
|
||
patch(
|
||
"app.tasks.story_title_tasks.generate_story_title_after_create.delay",
|
||
),
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.refresh_chapter_evidence_snapshot_with_retry_sync",
|
||
),
|
||
):
|
||
route_agent_mock.plan_batch.return_value = None
|
||
route_agent_mock.decide.side_effect = decide_capture
|
||
|
||
na = MagicMock()
|
||
nac.return_value = na
|
||
na.generate_narrative.return_value = '{"paragraphs": [{"content": "叙事正文段落足够长用于测试合并逻辑避免触发过短回退"}]}'
|
||
|
||
mock_story = MagicMock()
|
||
mock_story.id = "11111111-1111-1111-1111-111111111111"
|
||
csw.return_value = mock_story
|
||
|
||
mis.from_env.return_value = MagicMock(enabled=False)
|
||
|
||
session = MagicMock()
|
||
exec_result = MagicMock()
|
||
exec_result.unique.return_value.scalar_one_or_none.return_value = None
|
||
session.execute.return_value = exec_result
|
||
|
||
state = MemoirStateSchema(
|
||
stage_order=["childhood"],
|
||
current_stage="childhood",
|
||
covered_stages=[],
|
||
slots={},
|
||
)
|
||
run_story_pipeline_for_category_batch(
|
||
session,
|
||
user_id="user-1",
|
||
chapter_category="summary",
|
||
category_segments=[seg],
|
||
state=state,
|
||
user_profile="",
|
||
user_birth_year=None,
|
||
llm=object(),
|
||
memory_evidence=evidence_payload,
|
||
)
|
||
|
||
assert captured["batch_transcript"] == oral
|
||
assert "[摘要:rolling]" not in captured["batch_transcript"]
|
||
assert "滚动摘要" not in captured["batch_transcript"]
|
||
route_agent_mock.decide.assert_called_once()
|
||
assert route_agent_mock.decide.call_args.kwargs["batch_transcript"] == oral
|
||
|
||
|
||
def test_decide_receives_only_same_stage_story_candidates() -> None:
|
||
"""路由候选仅含 story.stage == chapter_category，禁止跨章节 append 导致多章内容相同。"""
|
||
oral = "这是一条用于测试路由候选过滤的用户口述内容足够长以避免叙事回退误判"
|
||
captured: dict[str, list] = {}
|
||
|
||
def decide_capture(
|
||
*, candidate_stories: object, **kwargs: object
|
||
) -> StoryRouteDecision:
|
||
captured["candidates"] = list(candidate_stories)
|
||
return StoryRouteDecision(
|
||
decision="new_story",
|
||
new_story_title="测试新故事标题六个字以上",
|
||
reason="测试",
|
||
)
|
||
|
||
seg = SimpleNamespace(id="seg-route-stage-filter-1", user_input_text=oral)
|
||
childhood_story = SimpleNamespace(
|
||
id="s-child", stage="childhood", canonical_markdown=""
|
||
)
|
||
education_story = SimpleNamespace(
|
||
id="s-edu", stage="education", canonical_markdown=""
|
||
)
|
||
|
||
route_agent_mock = MagicMock()
|
||
|
||
with (
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.list_active_stories_for_user_sync",
|
||
return_value=[childhood_story, education_story],
|
||
),
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.StoryRouteAgent",
|
||
return_value=route_agent_mock,
|
||
),
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.NarrativeAgent",
|
||
) as nac,
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.create_story_with_version_sync",
|
||
) as csw,
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.ensure_chapter_story_link_sync",
|
||
),
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.reorder_chapter_story_links_by_life_order_sync",
|
||
),
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.mark_chapter_dirty_sync",
|
||
),
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.chapter_needs_cover_enqueue",
|
||
return_value=False,
|
||
),
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.MemoirImageSettings",
|
||
) as mis,
|
||
patch(
|
||
"app.tasks.story_title_tasks.generate_story_title_after_create.delay",
|
||
),
|
||
patch(
|
||
"app.features.memoir.story_pipeline_sync.refresh_chapter_evidence_snapshot_with_retry_sync",
|
||
),
|
||
):
|
||
route_agent_mock.plan_batch.return_value = None
|
||
route_agent_mock.decide.side_effect = decide_capture
|
||
|
||
na = MagicMock()
|
||
nac.return_value = na
|
||
na.generate_narrative.return_value = '{"paragraphs": [{"content": "叙事正文段落足够长用于测试合并逻辑避免触发过短回退"}]}'
|
||
|
||
mock_story = MagicMock()
|
||
mock_story.id = "22222222-2222-2222-2222-222222222222"
|
||
csw.return_value = mock_story
|
||
|
||
mis.from_env.return_value = MagicMock(enabled=False)
|
||
|
||
session = MagicMock()
|
||
exec_result = MagicMock()
|
||
exec_result.unique.return_value.scalar_one_or_none.return_value = None
|
||
session.execute.return_value = exec_result
|
||
|
||
state = MemoirStateSchema(
|
||
stage_order=["education"],
|
||
current_stage="education",
|
||
covered_stages=[],
|
||
slots={},
|
||
)
|
||
run_story_pipeline_for_category_batch(
|
||
session,
|
||
user_id="user-1",
|
||
chapter_category="education",
|
||
category_segments=[seg],
|
||
state=state,
|
||
user_profile="",
|
||
user_birth_year=None,
|
||
llm=object(),
|
||
memory_evidence={
|
||
"relevant_chunks": [],
|
||
"relevant_summaries": [],
|
||
"relevant_facts": [],
|
||
"relevant_stories": [],
|
||
},
|
||
)
|
||
|
||
assert len(captured["candidates"]) == 1
|
||
assert captured["candidates"][0].id == "s-edu"
|
||
assert captured["candidates"][0].stage == "education"
|
||
route_agent_mock.decide.assert_called_once()
|
||
va = route_agent_mock.decide.call_args.kwargs["valid_story_ids"]
|
||
assert va == {"s-edu"}
|
||
|
||
|
||
def test_get_story_route_prompt_includes_route_boundary_rule() -> None:
|
||
from app.agents.memoir.prompts import get_story_route_prompt
|
||
|
||
out = get_story_route_prompt(
|
||
chapter_category="summary",
|
||
chapter_title="标题",
|
||
batch_transcript="仅口述",
|
||
candidate_stories_json="[]",
|
||
)
|
||
assert "路由边界" in out
|
||
assert "仅根据" in out or "本批口述" in out
|