geMoldInsight/tests/test_experience_feedback_algorithm.py

"""D17 Human-in-Loop 闭环：算法接缝回归测试。

覆盖：
- PartingCandidateGenerator：hints 注入 axis 优先级、method 标签
- PartingSchemeScorer：score_breakdown 新增 human_hint_bonus、total_score 加成
- MultiSchemeMoldPlanner：hints 透传、global_summary.applied_hints
- processing_service：OCC payload 装配 experience_hints（OCC-gated）

注：PartingCandidateGenerator / PartingSchemeScorer / MultiSchemeMoldPlanner 本身
不直接 import OCC（OCC shape 留 lazy 在 occ_worker），可在无 OCC 环境直接测试。
processing_service.py 通过 occ_process_pool 间接 import OCC，那两个测试 OCC-gated。
"""
from typing import Any, Dict, List
from unittest.mock import AsyncMock, MagicMock

import pytest

# OCC 条件探测（仅 processing_service 测试需要）
try:
    import OCC  # noqa: F401
    HAS_OCC = True
except ImportError:
    HAS_OCC = False

OCC_GATED = pytest.mark.skipif(
    not HAS_OCC,
    reason="D17 payload 测试依赖 processing_service（含 occ_process_pool），"
           "OCC 缺失时无法 import；项目硬规则 OCC-gated",
)


# ── PartingCandidateGenerator 测试 ──

def _make_analysis(dims=(80.0, 60.0, 40.0), volume=50000.0):
    """构造 PartingCandidateGenerator 期望的 analysis dict。"""
    return {
        "bounding_box": {
            "dimensions": list(dims),
            "min": [0.0, 0.0, 0.0],
            "max": list(dims),
            "center": [d / 2 for d in dims],
        },
        "volume": volume,
        "inertia_matrix": [[1000.0, 0.0, 0.0], [0.0, 800.0, 0.0], [0.0, 0.0, 600.0]],
        "axis_normal_stats": {"X": 35.0, "Y": 35.0, "Z": 30.0},
    }


def test_parting_candidate_generator_no_hints_default():
    """hints=None 应保持原有 3 轴评分（向后兼容）。"""
    from moldinsight.core.parting_candidate_generator import PartingCandidateGenerator

    gen = PartingCandidateGenerator()
    candidates = gen.generate_candidates(
        analysis=_make_analysis(),
        is_foam_material=False,
        max_candidates=3,
    )
    assert len(candidates) == 3
    # 没有 human_experience_primary 标签
    for c in candidates:
        assert c["method"] != "human_experience_primary"


def test_parting_candidate_generator_applies_hints_axis_weight():
    """hints={X: weight=0.9, sample_count=3} → X 轴 method 标签升级、priority_score +18。"""
    from moldinsight.core.parting_candidate_generator import PartingCandidateGenerator

    gen = PartingCandidateGenerator()
    candidates_no = gen.generate_candidates(
        analysis=_make_analysis(),
        is_foam_material=False,
        max_candidates=3,
        hints=None,
    )
    x_no = next(c for c in candidates_no if c["axis"] == "X")
    x_no_score = x_no["priority_score"]

    candidates_with = gen.generate_candidates(
        analysis=_make_analysis(),
        is_foam_material=False,
        max_candidates=3,
        hints={
            "X": {"weight": 0.9, "sample_count": 3, "adopted_count": 5, "rejected_count": 1},
        },
    )
    x_with = next(c for c in candidates_with if c["axis"] == "X")
    # priority_score 提升 18 分（0.9 × 20）
    assert abs(x_with["priority_score"] - (x_no_score + 18.0)) < 0.01
    # method 标签变为 human_experience_primary
    assert x_with["method"] == "human_experience_primary"


def test_parting_candidate_generator_low_sample_count_no_method_upgrade():
    """sample_count=1（信号不足）时 method 标签不升级。"""
    from moldinsight.core.parting_candidate_generator import PartingCandidateGenerator

    gen = PartingCandidateGenerator()
    candidates = gen.generate_candidates(
        analysis=_make_analysis(),
        is_foam_material=False,
        max_candidates=3,
        hints={
            "Y": {"weight": 0.8, "sample_count": 1, "adopted_count": 1, "rejected_count": 0},
        },
    )
    y = next(c for c in candidates if c["axis"] == "Y")
    # sample_count < 2 → method 不升级（但 priority_score 仍加成 16 分）
    assert y["method"] != "human_experience_primary"


# ── PartingSchemeScorer 测试 ──

def _make_scheme(axis: str = "X", method: str = "geometric_primary", score: float = 60.0):
    """构造 PartingSchemeScorer 期望的 scheme dict。"""
    return {
        "scheme_id": f"scheme_{axis}",
        "axis": axis,
        "parting": {"axis": axis},
        "method": method,
        "priority_score": score,
        "cavity_data": {
            "mold_cavities": {
                "cavity": {"vertex_count": 100},
                "core": {"vertex_count": 100},
            },
            "quality_checks": {
                "undercut_regions": [],
                "side_actions": {
                    "summary": {"total_mechanism_count": 0, "complexity": "simple"},
                    "slider_mechanisms": [],
                    "lifter_mechanisms": [],
                    "undercut_analysis": {"total_undercut_area": 0},
                },
            },
            "manufacturing_info": {
                "estimated_mold_size": {"length": 200, "width": 200, "height": 200},
                "estimated_clamping_force": "150-300 吨",
            },
        },
        "key_info": {
            "quality_considerations": {"warpage_risk": "low"},
            "geometric_characteristics": {"wall_thickness_range": "1.5 - 3.0 mm"},
        },
    }


def test_scheme_scorer_no_hints_no_bonus_field():
    """hints=None → score_breakdown 不含 human_hint_bonus（保持默认结构）。"""
    from moldinsight.core.parting_scheme_scorer import PartingSchemeScorer

    scorer = PartingSchemeScorer()
    scored = scorer.score_schemes([_make_scheme("X")])
    # hints=None 时 bonus=0，但仍写入 score_breakdown 以让前端 diff 稳定
    assert "human_hint_bonus" in scored[0]["score_breakdown"]
    assert scored[0]["score_breakdown"]["human_hint_bonus"] == 0.0


def test_scheme_scorer_human_hint_bonus_added():
    """hints={Y: weight=1.0, sample_count=5} → score_breakdown.human_hint_bonus == 12.0。"""
    from moldinsight.core.parting_scheme_scorer import PartingSchemeScorer

    scorer = PartingSchemeScorer()
    hints = {"Y": {"weight": 1.0, "sample_count": 5, "adopted_count": 5, "rejected_count": 0}}

    # 同一方案：有 hints vs 无 hints，total_score 差应等于 human_hint_bonus
    scored_with = scorer.score_schemes([_make_scheme("Y")], hints=hints)
    scored_without = scorer.score_schemes([_make_scheme("Y")], hints=None)

    assert scored_with[0]["score_breakdown"]["human_hint_bonus"] == 12.0
    delta = scored_with[0]["score"] - scored_without[0]["score"]
    assert abs(delta - 12.0) < 0.01


def test_scheme_scorer_low_sample_count_halves_bonus():
    """sample_count=1 → bonus ×0.5 = 6.0（信号不足折半）。"""
    from moldinsight.core.parting_scheme_scorer import PartingSchemeScorer

    scorer = PartingSchemeScorer()
    hints = {"Z": {"weight": 1.0, "sample_count": 1, "adopted_count": 1, "rejected_count": 0}}

    scored = scorer.score_schemes([_make_scheme("Z")], hints=hints)
    assert scored[0]["score_breakdown"]["human_hint_bonus"] == 6.0


def test_scheme_scorer_zero_weight_no_bonus():
    """weight=0 → bonus=0（既不加分也不扣分）。"""
    from moldinsight.core.parting_scheme_scorer import PartingSchemeScorer

    scorer = PartingSchemeScorer()
    hints = {"X": {"weight": 0.0, "sample_count": 3, "adopted_count": 0, "rejected_count": 3}}

    scored = scorer.score_schemes([_make_scheme("X")], hints=hints)
    assert scored[0]["score_breakdown"]["human_hint_bonus"] == 0.0


# ── MultiSchemeMoldPlanner 测试（OCC-gated：直接 import OCC）──

@OCC_GATED
def test_multi_scheme_planner_passes_hints_through(monkeypatch):
    """generate_plan(hints=...) 应透传到 candidate_generator 和 scheme_scorer。"""
    from moldinsight.core import multi_scheme_planner

    captured = {"candidate_hints": None, "scorer_hints": None}

    class FakeGenerator:
        def __init__(self):
            self.calls = []

        def set_material(self, *_):
            pass

        def apply_process_params(self, *_):
            pass

        def analyze_product_geometry(self, _shape):
            return {
                "bounding_box": {"dimensions": [80, 60, 40]},
                "volume": 50000,
                "inertia_matrix": [[1000, 0, 0], [0, 800, 0], [0, 0, 600]],
            }

    class FakePlanner:
        def generate_candidates(self, **kwargs):
            captured["candidate_hints"] = kwargs.get("hints")
            return [
                {
                    "scheme_id": "scheme_1",
                    "axis": "X",
                    "direction": [1, 0, 0],
                    "title": "推荐候选方向",
                    "method": "geometric_primary",
                    "priority_score": 80.0,
                    "opening_span_mm": 40.0,
                    "projected_area_cm2": 32.0,
                    "reason": "test",
                },
            ]

        def score_schemes(self, schemes, *, hints=None):
            captured["scorer_hints"] = hints
            for s in schemes:
                s["score"] = 80.0
                s["score_breakdown"] = {"human_hint_bonus": 0.0}
            return schemes

    planner_obj = multi_scheme_planner.MultiSchemeMoldPlanner.__new__(
        multi_scheme_planner.MultiSchemeMoldPlanner
    )
    planner_obj.candidate_generator = FakePlanner()
    planner_obj.scheme_scorer = FakePlanner()
    planner_obj.candidate_generator.generate_candidates = planner_obj.candidate_generator.generate_candidates
    planner_obj.scheme_scorer.score_schemes = planner_obj.scheme_scorer.score_schemes
    # 用 planner_obj.candidate_generator 与 scheme_scorer 是 FakePlanner 实例，所以
    # generator.generate_candidates 会调用 FakePlanner.generate_candidates —— 但因为同
    # 一实例两个方法都覆盖，下面显式覆写两次：
    planner_obj.candidate_generator = type("G", (), {
        "generate_candidates": lambda self, **kw: (
            captured.update({"candidate_hints": kw.get("hints")}) or
            [{"scheme_id": "scheme_1", "axis": "X", "direction": [1,0,0],
              "title": "推荐", "method": "geo", "priority_score": 80.0,
              "opening_span_mm": 40.0, "projected_area_cm2": 32.0, "reason": "test"}]
        )
    })()
    planner_obj.scheme_scorer = type("S", (), {
        "score_schemes": lambda self, schemes, *, hints=None: (
            captured.update({"scorer_hints": hints}) or
            [{**s, "score": 80.0, "score_breakdown": {"human_hint_bonus": 0.0}} for s in schemes]
        )
    })()

    fake_hints = {"X": {"weight": 0.9, "sample_count": 3, "adopted_count": 5, "rejected_count": 1}}
    result = planner_obj.generate_plan(
        shape=MagicMock(),
        material={"name": "ABS"},
        is_foam_material=False,
        hints=fake_hints,
    )

    assert captured["candidate_hints"] == fake_hints, "candidate_generator 未接收 hints"
    assert captured["scorer_hints"] == fake_hints, "scheme_scorer 未接收 hints"
    assert result["global_summary"]["applied_hints"] == fake_hints


@OCC_GATED
def test_multi_scheme_planner_applied_hints_default_empty():
    """generate_plan 不传 hints 时 global_summary.applied_hints 为空 dict。"""
    from moldinsight.core import multi_scheme_planner

    planner_obj = multi_scheme_planner.MultiSchemeMoldPlanner.__new__(
        multi_scheme_planner.MultiSchemeMoldPlanner
    )
    planner_obj.candidate_generator = type("G", (), {
        "generate_candidates": lambda self, **kw: [
            {"scheme_id": "scheme_1", "axis": "X", "direction": [1,0,0],
             "title": "推荐", "method": "geo", "priority_score": 80.0,
             "opening_span_mm": 40.0, "projected_area_cm2": 32.0, "reason": "test"}
        ]
    })()
    planner_obj.scheme_scorer = type("S", (), {
        "score_schemes": lambda self, schemes, *, hints=None: (
            [{**s, "score": 80.0, "score_breakdown": {"human_hint_bonus": 0.0}} for s in schemes]
        )
    })()

    result = planner_obj.generate_plan(
        shape=MagicMock(),
        material={"name": "ABS"},
        is_foam_material=False,
    )
    assert result["global_summary"]["applied_hints"] == {}


# ── processing_service payload 装配测试（OCC-gated）──

@OCC_GATED
def test_processing_service_step_generate_cavity_includes_experience_hints(monkeypatch):
    """_step_generate_cavity 应在 run_occ payload 中装入 experience_hints。"""
    import asyncio
    from moldinsight.services import processing_service

    # Mock experience_feedback_service
    fake_hints = [{"scheme_axis": "X", "weight": 0.8, "sample_count": 4,
                   "adopted_count": 4, "rejected_count": 0}]
    fake_ef_service = MagicMock()
    fake_ef_service.resolve_for_process_params = AsyncMock(return_value=fake_hints)

    monkeypatch.setattr(
        processing_service, "experience_feedback_service", fake_ef_service, raising=False
    )

    # Mock run_occ 拦截 payload
    captured_payload = {}
    async def fake_run_occ(self, op_name, payload, timeout):
        captured_payload["op_name"] = op_name
        captured_payload["payload"] = payload
        return {
            "plan_result": {"candidate_schemes": [], "best_scheme_id": None,
                            "global_summary": {"applied_hints": {}}},
            "export_manifest": None,
        }
    monkeypatch.setattr(
        processing_service.ProcessingService, "run_occ", fake_run_occ
    )

    # Mock cad_exporter
    monkeypatch.setattr(
        processing_service.ProcessingService, "__init__",
        lambda self: setattr(self, "cad_exporter", MagicMock(output_dir="/tmp"))
    )

    svc = processing_service.ProcessingService()
    svc.cad_exporter = MagicMock(output_dir="/tmp")

    async def run():
        await svc._step_generate_cavity(
            db_session=MagicMock(),
            file_path="/tmp/x.stp",
            selected_material={"name": "ABS"},
            is_foam_material=False,
            process_params={"material": "ABS", "draft_angle": 2.0,
                            "shrinkage_rate": 0.5, "parting_precision": 0.1,
                            "cavity_match": 95},
            task_id="task-test-1",
            timeout=60,
        )

    asyncio.run(run())

    assert captured_payload["payload"]["experience_hints"] == fake_hints


@OCC_GATED
def test_processing_service_step_generate_cavity_empty_hints_on_error(monkeypatch):
    """experience_feedback_service 抛异常时 hints 应回退到空 list（不阻塞主流程）。"""
    import asyncio
    from moldinsight.services import processing_service

    fake_ef_service = MagicMock()
    fake_ef_service.resolve_for_process_params = AsyncMock(
        side_effect=RuntimeError("DB down")
    )

    monkeypatch.setattr(
        processing_service, "experience_feedback_service", fake_ef_service, raising=False
    )

    captured_payload = {}
    async def fake_run_occ(self, op_name, payload, timeout):
        captured_payload["payload"] = payload
        return {
            "plan_result": {"candidate_schemes": [], "best_scheme_id": None,
                            "global_summary": {"applied_hints": {}}},
            "export_manifest": None,
        }
    monkeypatch.setattr(
        processing_service.ProcessingService, "run_occ", fake_run_occ
    )

    svc = processing_service.ProcessingService()
    svc.cad_exporter = MagicMock(output_dir="/tmp")

    async def run():
        await svc._step_generate_cavity(
            db_session=MagicMock(),
            file_path="/tmp/x.stp",
            selected_material={"name": "ABS"},
            is_foam_material=False,
            process_params={"material": "ABS"},
            task_id="task-test-1",
            timeout=60,
        )

    asyncio.run(run())

    # 抛异常时回退到空 list，主流程继续
    assert captured_payload["payload"]["experience_hints"] == []