417 lines
16 KiB
Python
417 lines
16 KiB
Python
|
"""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"] == []
|