beige/backend/tests/test_flywheel_wiring.py

"""
test_flywheel_wiring.py — 飞轮 Point1+Point2 验证
Point1: 5种信号正确写入 preference_events（选文案/选图/审核 approve/reject/regenerate）
Point2: aggregator 返回明确 prompt_fragment（含偏好角度、打回原因，workspace 严格过滤）
无 Docker — 全程纯 Python 单元测试
Point3+Point4 见 test_flywheel_wiring_p34.py
"""

from unittest.mock import MagicMock

SIGNAL_WEIGHTS_MAP = {
    "text_select": 3,
    "image_select": 3,
    "approve": 5,
    "reject_with_reason": -3,
    "regenerate": -1,
}

_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": ""},
]


# ══════════════════════════════════════════════════════════════
# Point1: 五种信号写入结构验证
# ══════════════════════════════════════════════════════════════
class TestPoint1SignalTypes:
    """验证 5 种信号都有对应 weight，且字段完整。"""

    def test_all_signal_types_have_weights(self):
        from app.constants.enums import SIGNAL_WEIGHTS
        required = {"text_select", "image_select", "approve", "reject_with_reason", "regenerate"}
        missing = required - set(SIGNAL_WEIGHTS.keys())
        assert not missing, f"SIGNAL_WEIGHTS 缺少: {missing}"

    def test_positive_signals_weight_positive(self):
        from app.constants.enums import SIGNAL_WEIGHTS
        for sig in ("text_select", "image_select", "approve"):
            assert SIGNAL_WEIGHTS[sig] > 0, f"{sig} weight 应 > 0，实际={SIGNAL_WEIGHTS[sig]}"

    def test_negative_signals_weight_negative(self):
        from app.constants.enums import SIGNAL_WEIGHTS
        for sig in ("reject_with_reason", "regenerate"):
            assert SIGNAL_WEIGHTS[sig] < 0, f"{sig} weight 应 < 0，实际={SIGNAL_WEIGHTS[sig]}"

    def test_collect_preference_event_returns_complete_fields(self):
        from app.services.ai_engine.preference_aggregator import collect_preference_event
        event = collect_preference_event(
            signal_type="text_select",
            user_id=7,
            workspace_id=1,
            product_id=10,
            angle_label="成分党",
        )
        required_keys = {"signal_type", "signal_weight", "user_id",
                         "workspace_id", "product_id", "angle_label",
                         "reason", "data_ownership"}
        assert required_keys.issubset(event.keys()), f"缺字段: {required_keys - event.keys()}"

    def test_collect_event_weight_matches_signal(self):
        from app.services.ai_engine.preference_aggregator import collect_preference_event
        ev = collect_preference_event("approve", 1, 1, 10)
        assert ev["signal_weight"] == SIGNAL_WEIGHTS_MAP["approve"]


# ══════════════════════════════════════════════════════════════
# Point2: aggregator 返回 prompt_fragment + workspace 严格过滤
# ══════════════════════════════════════════════════════════════
class TestPoint2AggregatorOutput:
    """验证 aggregate_preference_context 返回结构与 workspace/product 隔离。"""

    def test_returns_required_keys(self):
        from app.services.ai_engine.preference_aggregator import aggregate_preference_context
        ctx = aggregate_preference_context(_EVENTS_ENOUGH, _PRODUCT, 1, 10)
        for k in ("prompt_fragment", "recent_preference", "reject_reasons", "injected_count"):
            assert k in ctx, f"返回值缺字段: {k}"

    def test_prompt_fragment_not_empty_with_signals(self):
        from app.services.ai_engine.preference_aggregator import aggregate_preference_context
        ctx = aggregate_preference_context(_EVENTS_ENOUGH, _PRODUCT, 1, 10)
        assert ctx["prompt_fragment"].strip(), "有足够信号时 prompt_fragment 不应为空"

    def test_prompt_fragment_contains_top_angle(self):
        from app.services.ai_engine.preference_aggregator import aggregate_preference_context
        ctx = aggregate_preference_context(_EVENTS_ENOUGH, _PRODUCT, 1, 10)
        assert "成分党" in ctx["prompt_fragment"]  # 权重最高角度

    def test_reject_reasons_in_fragment(self):
        from app.services.ai_engine.preference_aggregator import aggregate_preference_context
        ctx = aggregate_preference_context(_EVENTS_ENOUGH, _PRODUCT, 1, 10)
        assert "太像广告" in ctx["prompt_fragment"]

    def test_workspace_isolation_filters_other_workspace(self):
        from app.services.ai_engine.preference_aggregator import aggregate_preference_context
        mixed = _EVENTS_ENOUGH + [
            {"signal_type": "text_select", "workspace_id": 2, "product_id": 10,
             "angle_label": "workspace2专有角度", "signal_weight": 100, "reason": ""},
        ]
        ctx = aggregate_preference_context(mixed, _PRODUCT, 1, 10)
        assert "workspace2专有角度" not in ctx["prompt_fragment"]

    def test_product_isolation_filters_other_product(self):
        from app.services.ai_engine.preference_aggregator import aggregate_preference_context
        mixed = _EVENTS_ENOUGH + [
            {"signal_type": "text_select", "workspace_id": 1, "product_id": 99,
             "angle_label": "product99专有角度", "signal_weight": 100, "reason": ""},
        ]
        ctx = aggregate_preference_context(mixed, _PRODUCT, 1, 10)
        assert "product99专有角度" not in ctx["prompt_fragment"]

    def test_cold_start_when_insufficient_signals(self):
        from app.services.ai_engine.preference_aggregator import aggregate_preference_context
        from app.services.ai_engine.constants import FLYWHEEL_COLD_START
        few = _EVENTS_ENOUGH[:max(0, FLYWHEEL_COLD_START - 1)]
        ctx = aggregate_preference_context(few, _PRODUCT, 1, 10)
        assert "recent_preference" in ctx
        assert ctx["injected_count"] == len(few)