6a2632da70
将 Clover 从上层产品包旧仓库中独立出来,建立专属版本控制。 当前状态=纵切片端到端已打通(登录→选品→出文出图→审核→下载包), M1文案质量去套路化已验收。此提交作为后续按核销清单逐条修复的基线。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
167 lines
8.2 KiB
Python
167 lines
8.2 KiB
Python
"""
|
||
test_flywheel_wiring_p34.py — 飞轮 Point3+Point4 验证
|
||
Point3: pipeline 把 prompt_fragment 注入 user_prompt（prompt trace 含片段）
|
||
Point4: UI 层 flywheel_injected SSE 事件含 recent_preference + reject_reasons
|
||
无 Docker — 全程纯 Python 单元测试
|
||
Point1+Point2 见 test_flywheel_wiring.py
|
||
"""
|
||
|
||
from unittest.mock import MagicMock
|
||
|
||
_PRODUCT = {
|
||
"name": "素颜霜",
|
||
"category": "护肤",
|
||
"selling_points": ["保湿", "天然"],
|
||
"style_tone": "素人分享风",
|
||
"text_angles": ["成分党", "懒人必备"],
|
||
"custom_prompt": "请保持真实素人口吻",
|
||
}
|
||
|
||
_EVENTS_ENOUGH = [
|
||
{"signal_type": "text_select", "workspace_id": 1, "product_id": 10,
|
||
"angle_label": "成分党", "signal_weight": 3, "reason": ""},
|
||
{"signal_type": "approve", "workspace_id": 1, "product_id": 10,
|
||
"angle_label": "成分党", "signal_weight": 5, "reason": ""},
|
||
{"signal_type": "text_select", "workspace_id": 1, "product_id": 10,
|
||
"angle_label": "懒人必备", "signal_weight": 3, "reason": ""},
|
||
{"signal_type": "reject_with_reason","workspace_id": 1, "product_id": 10,
|
||
"angle_label": "", "signal_weight": -3, "reason": "太像广告"},
|
||
{"signal_type": "regenerate", "workspace_id": 1, "product_id": 10,
|
||
"angle_label": "", "signal_weight": -1, "reason": ""},
|
||
]
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# Point3: pipeline 把 prompt_fragment 注入 user_prompt
|
||
# ══════════════════════════════════════════════════════════════
|
||
class TestPoint3PromptInjection:
|
||
"""验证 compose_variants 把 flywheel_context 正确注入 user_prompt。"""
|
||
|
||
def test_compose_variants_appends_flywheel_to_user_prompt(self):
|
||
from app.services.ai_engine.prompt_composer import compose_variants
|
||
fragment = "【偏好角度参考】历史选择偏好：成分党"
|
||
_, user_prompt = compose_variants(_PRODUCT, 3, flywheel_context=fragment)
|
||
assert fragment in user_prompt, "flywheel_context 应出现在 user_prompt 中"
|
||
|
||
def test_compose_variants_flywheel_not_in_system_prompt(self):
|
||
"""飞轮片段不应覆盖 system_prompt（COPY_SYSTEM 质量护栏不能被干扰）"""
|
||
from app.services.ai_engine.prompt_composer import compose_variants
|
||
fragment = "飞轮注入内容UNIQUE123"
|
||
system_prompt, _ = compose_variants(_PRODUCT, 2, flywheel_context=fragment)
|
||
assert fragment not in system_prompt, "飞轮片段不应出现在 system_prompt"
|
||
|
||
def test_compose_variants_empty_flywheel_still_valid(self):
|
||
"""无飞轮片段时 compose_variants 仍正常返回 (system, user)"""
|
||
from app.services.ai_engine.prompt_composer import compose_variants
|
||
system_p, user_p = compose_variants(_PRODUCT, 3, flywheel_context="")
|
||
assert isinstance(system_p, str) and len(system_p) > 0
|
||
assert isinstance(user_p, str) and len(user_p) > 0
|
||
|
||
def test_pipeline_fragment_contains_top_angle(self):
|
||
"""pipeline 重现：ORM events → aggregate → fragment 含最高权重角度"""
|
||
from app.services.ai_engine.preference_aggregator import aggregate_preference_context
|
||
|
||
def make_event(sig, angle, weight, reason):
|
||
e = MagicMock()
|
||
e.signal_type = sig
|
||
e.workspace_id = 1
|
||
e.product_id = 10
|
||
e.angle_label = angle
|
||
e.signal_weight = weight
|
||
e.reason = reason
|
||
return e
|
||
|
||
mock_events = [
|
||
make_event("text_select", "成分党", 3, ""),
|
||
make_event("approve", "成分党", 5, ""),
|
||
make_event("reject_with_reason", "", -3, "太像广告"),
|
||
make_event("text_select", "懒人必备", 3, ""),
|
||
make_event("regenerate", "", -1, ""),
|
||
]
|
||
events_dicts = [
|
||
{"signal_type": e.signal_type, "workspace_id": e.workspace_id,
|
||
"product_id": e.product_id, "angle_label": e.angle_label or "",
|
||
"signal_weight": e.signal_weight, "reason": e.reason or ""}
|
||
for e in mock_events
|
||
]
|
||
ctx = aggregate_preference_context(events_dicts, _PRODUCT, 1, 10)
|
||
fragment = ctx.get("prompt_fragment", "")
|
||
assert isinstance(fragment, str)
|
||
assert "成分党" in fragment
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════
|
||
# Point4: UI 层 flywheel_injected SSE 事件结构验证
|
||
# ══════════════════════════════════════════════════════════════
|
||
class TestPoint4SseFlywheelEvent:
|
||
"""验证 flywheel_injected SSE 事件包含 UI 展示所需字段。"""
|
||
|
||
# FE useSse.ts flywheel_injected 事件期望字段
|
||
FE_EXPECTED_FIELDS = {"recent_preference", "reject_reasons"}
|
||
|
||
def test_aggregator_output_matches_sse_fields(self):
|
||
from app.services.ai_engine.preference_aggregator import aggregate_preference_context
|
||
ctx = aggregate_preference_context(_EVENTS_ENOUGH, _PRODUCT, 1, 10)
|
||
missing = self.FE_EXPECTED_FIELDS - ctx.keys()
|
||
assert not missing, f"SSE flywheel_injected 缺少字段: {missing}"
|
||
|
||
def test_recent_preference_is_nonempty_string(self):
|
||
from app.services.ai_engine.preference_aggregator import aggregate_preference_context
|
||
ctx = aggregate_preference_context(_EVENTS_ENOUGH, _PRODUCT, 1, 10)
|
||
assert isinstance(ctx["recent_preference"], str) and ctx["recent_preference"].strip()
|
||
|
||
def test_reject_reasons_is_list_of_strings(self):
|
||
from app.services.ai_engine.preference_aggregator import aggregate_preference_context
|
||
ctx = aggregate_preference_context(_EVENTS_ENOUGH, _PRODUCT, 1, 10)
|
||
assert isinstance(ctx["reject_reasons"], list)
|
||
for r in ctx["reject_reasons"]:
|
||
assert isinstance(r, str)
|
||
|
||
def test_reject_reasons_max_3(self):
|
||
"""打回原因最多返回3条（FE 卡片布局限制）"""
|
||
from app.services.ai_engine.preference_aggregator import aggregate_preference_context
|
||
many_rejects = [
|
||
{"signal_type": "reject_with_reason", "workspace_id": 1,
|
||
"product_id": 10, "angle_label": "", "signal_weight": -3,
|
||
"reason": f"原因{i}"}
|
||
for i in range(10)
|
||
] + _EVENTS_ENOUGH
|
||
ctx = aggregate_preference_context(many_rejects, _PRODUCT, 1, 10)
|
||
assert len(ctx["reject_reasons"]) <= 3
|
||
|
||
def test_flywheel_service_get_context_returns_sse_compatible_shape(self):
|
||
"""flywheel_service.get_preference_context 返回结构与 FE PreferenceContext DTO 一致"""
|
||
from app.services.flywheel_service import get_preference_context
|
||
from app.constants.enums import SignalType
|
||
|
||
def make_orm(sig, angle, reason):
|
||
e = MagicMock()
|
||
e.signal_type = sig
|
||
e.angle_label = angle
|
||
e.reason = reason
|
||
return e
|
||
|
||
orm_events = [
|
||
make_orm(SignalType.TEXT_SELECT, "成分党", ""),
|
||
make_orm(SignalType.TEXT_SELECT, "成分党", ""),
|
||
make_orm(SignalType.APPROVE, "成分党", ""),
|
||
make_orm(SignalType.REJECT_WITH_REASON, "", "太像广告"),
|
||
make_orm(SignalType.TEXT_SELECT, "懒人必备", ""),
|
||
]
|
||
mock_db = MagicMock()
|
||
mq = MagicMock()
|
||
mock_db.query.return_value = mq
|
||
mq.filter.return_value = mq
|
||
mq.order_by.return_value = mq
|
||
mq.limit.return_value = mq
|
||
mq.all.return_value = orm_events
|
||
|
||
result = get_preference_context(mock_db, workspace_id=1, product_id=10)
|
||
|
||
fe_dto_keys = {"recent_preference", "reject_reasons", "injected_count"}
|
||
missing = fe_dto_keys - result.keys()
|
||
assert not missing, f"FE PreferenceContext DTO 缺字段: {missing}"
|
||
assert isinstance(result["recent_preference"], str)
|
||
assert isinstance(result["reject_reasons"], list)
|
||
assert isinstance(result["injected_count"], int)
|