185 lines
6.3 KiB
Python
185 lines
6.3 KiB
Python
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
|
||
from app.services.analysis_project_snapshot_loader import load_analysis_project_snapshot
|
||
|
||
|
||
class _FakeScalarResult:
|
||
def __init__(self, row):
|
||
self._row = row
|
||
|
||
def scalar_one_or_none(self):
|
||
return self._row
|
||
|
||
|
||
class _FakeScalarsResult:
|
||
def __init__(self, rows):
|
||
self._rows = list(rows)
|
||
|
||
def scalars(self):
|
||
return self
|
||
|
||
def all(self):
|
||
return list(self._rows)
|
||
|
||
|
||
class _FakeDB:
|
||
def __init__(self, project_row, task_rows, signal_rows):
|
||
self._responses = [
|
||
_FakeScalarResult(project_row),
|
||
_FakeScalarsResult(task_rows),
|
||
_FakeScalarsResult(signal_rows),
|
||
]
|
||
|
||
async def execute(self, _statement):
|
||
return self._responses.pop(0)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_load_analysis_project_snapshot_returns_richer_tasks_and_signals() -> None:
|
||
db = _FakeDB(
|
||
SimpleNamespace(
|
||
id=123,
|
||
tenant_id=9,
|
||
title="采购经理每日实时库存上报服务",
|
||
name="inventory-briefing",
|
||
description="首期聚焦总水位与TOP10缺货两个高价值信号。",
|
||
),
|
||
[
|
||
SimpleNamespace(
|
||
id=201,
|
||
title="每日固定时段生成并推送结构化摘要卡片",
|
||
description="每日凌晨生成结构化文本卡片，并在上午10点前推送。",
|
||
status="todo",
|
||
),
|
||
SimpleNamespace(
|
||
id=202,
|
||
title="WMS库存快照自动拉取与清洗",
|
||
description="每小时拉取库存快照并清洗异常值。",
|
||
status="todo",
|
||
),
|
||
],
|
||
[
|
||
SimpleNamespace(
|
||
id=301,
|
||
title="外部信号源 - WMS实时库存接口",
|
||
summary="提供各仓库每小时更新的库存快照，作为水位与缺货计算的原始数据源。",
|
||
reasoning="由 signal_source_registry 直接物化为正式 Signal。",
|
||
status="待处理",
|
||
source="custom:wms",
|
||
),
|
||
],
|
||
)
|
||
|
||
snapshot = await load_analysis_project_snapshot(
|
||
db=db,
|
||
tenant_id=9,
|
||
project_id=123,
|
||
)
|
||
|
||
assert snapshot["project_name"] == "采购经理每日实时库存上报服务"
|
||
assert snapshot["project_description"] == "首期聚焦总水位与TOP10缺货两个高价值信号。"
|
||
assert snapshot["task_titles"] == [
|
||
"每日固定时段生成并推送结构化摘要卡片",
|
||
"WMS库存快照自动拉取与清洗",
|
||
]
|
||
assert snapshot["tasks"][0]["description"] == "每日凌晨生成结构化文本卡片，并在上午10点前推送。"
|
||
assert snapshot["signal_titles"] == ["外部信号源 - WMS实时库存接口"]
|
||
assert snapshot["signals"][0]["summary"] == "提供各仓库每小时更新的库存快照，作为水位与缺货计算的原始数据源。"
|
||
assert snapshot["analysis_object_refs"] == {}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_load_analysis_project_snapshot_prefers_structural_signal_links_without_keyword_cues() -> None:
|
||
db = _FakeDB(
|
||
SimpleNamespace(
|
||
id=456,
|
||
tenant_id=9,
|
||
title="采购经理每日实时库存上报服务",
|
||
name="inventory-briefing",
|
||
description="首期聚焦总水位与TOP10缺货两个高价值信号。",
|
||
),
|
||
[
|
||
SimpleNamespace(
|
||
id=401,
|
||
title="任务A",
|
||
description="根据前序结果生成给采购经理的日报文本，并保留查看入口。",
|
||
status="todo",
|
||
),
|
||
SimpleNamespace(
|
||
id=402,
|
||
title="任务B",
|
||
description="从WMS取库存样本写入中间表，供后续核算使用。",
|
||
status="todo",
|
||
),
|
||
],
|
||
[
|
||
SimpleNamespace(
|
||
id=501,
|
||
title="信号一",
|
||
summary="WMS每小时更新库存样本，是总水位核算的原始来源。",
|
||
reasoning="这条来源直接支撑库存核算。",
|
||
status="待处理",
|
||
source="custom:wms",
|
||
),
|
||
],
|
||
)
|
||
|
||
snapshot = await load_analysis_project_snapshot(
|
||
db=db,
|
||
tenant_id=9,
|
||
project_id=456,
|
||
)
|
||
|
||
assert snapshot["analysis_object_refs"] == {}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_load_analysis_project_snapshot_builds_signal_refs_when_project_has_no_tasks() -> None:
|
||
db = _FakeDB(
|
||
SimpleNamespace(
|
||
id=789,
|
||
tenant_id=9,
|
||
title="直播复盘辅助工具（速记版）",
|
||
name="live-review-quick",
|
||
description="目标是在极短时间内完成直播复盘速记。",
|
||
),
|
||
[],
|
||
[
|
||
SimpleNamespace(
|
||
id=801,
|
||
title="输出规范 - 三栏速记清单格式",
|
||
summary="输出纯文本文件，含高光、掉点、执行偏差三栏。",
|
||
reasoning="由 signal_source_registry 直接物化为正式 Signal。",
|
||
status="待处理",
|
||
source="custom:output",
|
||
),
|
||
SimpleNamespace(
|
||
id=802,
|
||
title="输入适配 - 双源时间对齐能力",
|
||
summary="支持手动校准录屏与评论的时间偏移，并在此基础上完成全链路切片分析。",
|
||
reasoning="由 signal_source_registry 直接物化为正式 Signal。",
|
||
status="待处理",
|
||
source="custom:input",
|
||
),
|
||
SimpleNamespace(
|
||
id=803,
|
||
title="外部信号源 - TikTok Shop 直播开放API（评论流",
|
||
summary="获取带毫秒级时间戳的原始评论序列，用于与录屏帧对齐。",
|
||
reasoning="由 signal_source_registry 直接物化为正式 Signal。",
|
||
status="待处理",
|
||
source="custom:tiktok",
|
||
),
|
||
],
|
||
)
|
||
|
||
snapshot = await load_analysis_project_snapshot(
|
||
db=db,
|
||
tenant_id=9,
|
||
project_id=789,
|
||
)
|
||
|
||
assert snapshot["task_titles"] == []
|
||
assert snapshot["analysis_object_refs"] == {}
|