"""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"] == []