cjw/geMoldInsight
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

x

Browse Source
This commit is contained in:
chenjw28
2026-08-31 18:01:34 +08:00
parent 3ea59551db
commit bee439cf34
46 changed files with 1884 additions and 1898 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/temp_test_injection_p0.py
-265
View File
@@ -1,265 +0,0 @@
import io
import importlib.util
import sys
import types
from pathlib import Path
from types import SimpleNamespace
import pytest
from fastapi import FastAPI, UploadFile
from httpx import ASGITransport, AsyncClient
from services.calculation_service import CalculationService
from utils.file_handler import FileHandler
VALID_STEP_BYTES = (
b"ISO-10303-21;\n"
b"HEADER;\n"
b"FILE_DESCRIPTION(('STEP AP214'),'1');\n"
b"ENDSEC;\n"
b"DATA;\n"
b"ENDSEC;\n"
b"END-ISO-10303-21;\n"
)
def _load_module_from_path(module_name: str, file_path: str, stub_modules: dict[str, object]):
originals = {}
for name, module in stub_modules.items():
originals[name] = sys.modules.get(name)
sys.modules[name] = module
try:
spec = importlib.util.spec_from_file_location(module_name, file_path)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(module)
return module
finally:
for name, original in originals.items():
if original is None:
sys.modules.pop(name, None)
else:
sys.modules[name] = original
@pytest.mark.asyncio
async def test_file_handler_sanitizes_step_filename(tmp_path):
handler = FileHandler(upload_dir=str(tmp_path))
upload = UploadFile(filename="../../bad name?.step", file=io.BytesIO(VALID_STEP_BYTES))
file_path, file_size, meta = await handler.save_uploaded_file(upload)
assert file_path.exists()
assert file_size == len(VALID_STEP_BYTES)
assert file_path.parent == tmp_path
assert ".." not in file_path.name
assert meta["safe_original_name"] == "bad_name.step"
assert len(meta["sha256"]) == 64
@pytest.mark.asyncio
async def test_file_handler_rejects_invalid_step_content(tmp_path):
handler = FileHandler(upload_dir=str(tmp_path))
upload = UploadFile(filename="fake.step", file=io.BytesIO(b"not-a-step"))
with pytest.raises(ValueError, match="不是有效的 STP/STEP 数据"):
await handler.save_uploaded_file(upload)
@pytest.mark.asyncio
async def test_export_route_requires_cached_shapes(monkeypatch):
fake_processing_service = types.ModuleType("services.processing_service")
fake_processing_service.processing_service = SimpleNamespace(
get_export_shapes=lambda task_id, scheme_id=None: None,
)
fake_auth_service = types.ModuleType("services.auth_service")
async def fake_current_user():
return SimpleNamespace(id=1)
fake_auth_service.get_current_active_user = fake_current_user
fake_redis_task_manager = types.ModuleType("services.redis_task_manager")
fake_redis_task_manager.redis_task_manager = SimpleNamespace(get_task=None)
fake_models_database = types.ModuleType("models.database")
fake_models_database.User = SimpleNamespace
advanced_router = _load_module_from_path(
"temp_advanced_router",
"d:\\Project\\geMoldInsight\\src\\api\\v1\\advanced_router.py",
{
"services.processing_service": fake_processing_service,
"services.auth_service": fake_auth_service,
"services.redis_task_manager": fake_redis_task_manager,
"models.database": fake_models_database,
},
)
app = FastAPI()
app.include_router(advanced_router.router)
app.dependency_overrides[advanced_router.get_current_active_user] = lambda: SimpleNamespace(id=1)
async def fake_get_task_data(task_id):
return {
"task_id": task_id,
"filename": "demo.step",
"best_scheme_id": "scheme_1",
}
monkeypatch.setattr(advanced_router, "_get_task_data", fake_get_task_data)
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
response = await client.post(
"/export-mold",
json={"task_id": "task-1", "scheme_id": "scheme_1", "formats": ["step"]},
)
assert response.status_code == 409
assert "导出缓存已失效" in response.json()["detail"]
def test_calculation_service_attaches_injection_system_summary():
plan_result = {
"best_scheme_id": "scheme_1",
"candidate_schemes": [
{
"scheme_id": "scheme_1",
"cavity_data": {
"product_analysis": {
"bounding_box": {"dimensions": [100, 80, 30]}
},
"manufacturing_info": {
"estimated_mold_size": {"length": 200, "width": 180, "height": 110}
},
"mold_cavities": {"cavity_count": 1},
},
"key_info": {},
}
],
}
result = CalculationService.attach_injection_system_summaries(plan_result, "ABS")
best_scheme = result["candidate_schemes"][0]
assert "injection_system" in best_scheme["cavity_data"]
assert best_scheme["cavity_data"]["manufacturing_info"]["cooling_summary"]["channel_count"] >= 1
assert best_scheme["cavity_data"]["manufacturing_info"]["gating_summary"]["gate_type"] in {"auto", "side", "center", "submarine", "fan"}
assert "injection_system" in result
@pytest.mark.asyncio
async def test_upload_route_persists_process_parameters(monkeypatch, tmp_path):
captured = {}
class DummyStorageService:
async def save_stp_file(self, session, file_path, original_filename, user_id):
captured["saved_file"] = {
"file_path": str(file_path),
"original_filename": original_filename,
"user_id": user_id,
}
return SimpleNamespace(id=42)
async def create_processing_task(self, session, task_id, stp_file_id, task_type="stp_parsing", parameters=None):
captured["task"] = {
"task_id": task_id,
"stp_file_id": stp_file_id,
"task_type": task_type,
"parameters": parameters,
}
async def fake_save_uploaded_file(file):
target = Path(tmp_path) / "cached_demo.step"
target.write_bytes(VALID_STEP_BYTES)
return target, len(VALID_STEP_BYTES), {
"safe_original_name": "demo.step",
"sha256": "a" * 64,
"original_filename": "demo.step",
"stored_filename": target.name,
}
async def fake_set_task(task_id, task_info):
captured["redis"] = {"task_id": task_id, "task_info": task_info}
async def fake_process_file_with_storage(task_id, file_path, stp_file_id, process_params):
captured["background"] = {
"task_id": task_id,
"file_path": str(file_path),
"stp_file_id": stp_file_id,
"process_params": process_params,
}
fake_processing_service_module = types.ModuleType("services.processing_service")
fake_processing_service_module.processing_service = SimpleNamespace(
process_file_with_storage=fake_process_file_with_storage,
)
fake_storage_module = types.ModuleType("services.storage_integration_rustfs")
fake_storage_module.StorageIntegrationService = lambda: DummyStorageService()
fake_redis_task_manager = types.ModuleType("services.redis_task_manager")
fake_redis_task_manager.redis_task_manager = SimpleNamespace(
set_task=fake_set_task,
)
fake_database_module = types.ModuleType("database.database")
async def override_get_db_session():
yield object()
fake_database_module.get_db_session = override_get_db_session
fake_auth_service = types.ModuleType("services.auth_service")
async def override_get_current_user():
return SimpleNamespace(id=7, username="tester")
fake_auth_service.get_current_active_user = override_get_current_user
fake_models_database = types.ModuleType("models.database")
fake_models_database.User = SimpleNamespace
upload_router = _load_module_from_path(
"temp_upload_router",
"d:\\Project\\geMoldInsight\\src\\api\\v1\\upload_router.py",
{
"services.processing_service": fake_processing_service_module,
"services.storage_integration_rustfs": fake_storage_module,
"services.redis_task_manager": fake_redis_task_manager,
"database.database": fake_database_module,
"services.auth_service": fake_auth_service,
"models.database": fake_models_database,
},
)
monkeypatch.setattr(upload_router.file_handler, "save_uploaded_file", fake_save_uploaded_file)
app = FastAPI()
app.include_router(upload_router.router)
app.dependency_overrides[upload_router.get_db_session] = override_get_db_session
app.dependency_overrides[upload_router.get_current_active_user] = override_get_current_user
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
response = await client.post(
"/upload",
files={"file": ("demo.step", VALID_STEP_BYTES, "application/step")},
data={
"material": "ABS",
"draft_angle": "3.5",
"shrinkage_rate": "0.8",
"parting_precision": "0.05",
"cavity_match": "96",
},
)
assert response.status_code == 200
payload = response.json()
assert payload["parameters"] == {
"material": "ABS",
"draft_angle": 3.5,
"shrinkage_rate": 0.8,
"parting_precision": 0.05,
"cavity_match": 96,
}
assert captured["task"]["parameters"] == payload["parameters"]
assert captured["redis"]["task_info"]["parameters"] == payload["parameters"]
assert captured["background"]["process_params"] == payload["parameters"]
tests/test_moldinsight_regressions.py
+233
View File
@@ -0,0 +1,233 @@
import io
import pytest
from fastapi import UploadFile
from moldinsight.core.parting_candidate_generator import PartingCandidateGenerator
from moldinsight.core.parting_scheme_scorer import PartingSchemeScorer
from moldinsight.services.calculation_service import CalculationService
from moldinsight.services.cost_estimate_service import estimate_cost_by_rules
from moldinsight.services.material_service import MaterialService
from shared.services.redis_task_manager import RedisTaskManager
from shared.utils.file_handler import FileHandler
VALID_STEP_BYTES = (
b"ISO-10303-21;\n"
b"HEADER;\n"
b"FILE_DESCRIPTION(('STEP AP214'),'1');\n"
b"ENDSEC;\n"
b"DATA;\n"
b"ENDSEC;\n"
b"END-ISO-10303-21;\n"
)
@pytest.mark.asyncio
async def test_file_handler_sanitizes_step_filename(tmp_path):
handler = FileHandler(upload_dir=str(tmp_path))
upload = UploadFile(filename="../../bad name?.step", file=io.BytesIO(VALID_STEP_BYTES))
file_path, file_size, meta = await handler.save_uploaded_file(upload)
assert file_path.exists()
assert file_size == len(VALID_STEP_BYTES)
assert file_path.parent == tmp_path
assert ".." not in file_path.name
assert meta["safe_original_name"] == "bad_name.step"
assert len(meta["sha256"]) == 64
@pytest.mark.asyncio
async def test_file_handler_rejects_invalid_step_content(tmp_path):
handler = FileHandler(upload_dir=str(tmp_path))
upload = UploadFile(filename="fake.step", file=io.BytesIO(b"not-a-step"))
with pytest.raises(ValueError, match="不是有效的 STP/STEP 数据"):
await handler.save_uploaded_file(upload)
def test_calculation_service_attaches_injection_system_summary():
plan_result = {
"best_scheme_id": "scheme_1",
"candidate_schemes": [
{
"scheme_id": "scheme_1",
"cavity_data": {
"product_analysis": {
"bounding_box": {"dimensions": [100, 80, 30]}
},
"manufacturing_info": {
"estimated_mold_size": {"length": 200, "width": 180, "height": 110}
},
"mold_cavities": {"cavity_count": 1},
},
"key_info": {},
}
],
}
result = CalculationService.attach_injection_system_summaries(plan_result, "ABS")
best_scheme = result["candidate_schemes"][0]
assert "injection_system" in best_scheme["cavity_data"]
assert best_scheme["cavity_data"]["manufacturing_info"]["cooling_summary"]["channel_count"] >= 1
assert best_scheme["cavity_data"]["manufacturing_info"]["gating_summary"]["gate_type"] in {
"auto", "side", "center", "submarine", "fan"
}
assert "injection_system" in result
def test_material_service_falls_back_to_abs_for_unknown_material():
material = MaterialService.get_material("UNKNOWN")
assert material["name"] == "ABS"
assert MaterialService.resolve_material("UNKNOWN") == "ABS"
assert MaterialService.is_foam_material("UNKNOWN") is False
def test_parting_candidate_generator_prioritizes_z_for_foam_material():
generator = PartingCandidateGenerator()
analysis = {
"bounding_box": {"dimensions": [120, 80, 60]},
"inertia_matrix": [[10, 0, 0], [0, 12, 0], [0, 0, 8]],
"axis_normal_stats": {"X": 20.0, "Y": 30.0, "Z": 50.0},
}
candidates = generator.generate_candidates(analysis, is_foam_material=True)
assert candidates[0]["axis"] == "Z"
assert candidates[0]["method"] == "foam_axis_rule"
assert "泡沫模具优先上下开模" in candidates[0]["reason"]
def test_parting_scheme_scorer_prefers_scheme_without_undercuts():
scorer = PartingSchemeScorer()
schemes = [
{
"scheme_id": "clean",
"priority_score": 90,
"offset_ratio": 0.0,
"method": "geometric_primary",
"parting": {"line": [[0, 0, 0], [10, 0, 0]]},
"core_required": True,
"cavity_data": {
"mold_cavities": {
"cavity": {"vertex_count": 120},
"core": {"vertex_count": 120},
},
"quality_checks": {
"parting_line_smoothness": 92,
"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": 220, "width": 180, "height": 120},
"estimated_clamping_force": "180",
},
"metadata": {"draft_angle": 2.0},
},
"key_info": {
"quality_considerations": {"warpage_risk": "low"},
"geometric_characteristics": {"wall_thickness_range": "1.8-3.2mm"},
},
},
{
"scheme_id": "complex",
"priority_score": 85,
"offset_ratio": 0.1,
"method": "geometric_primary",
"parting": {"line": [[0, 0, 0], [100, 0, 0], [100, 50, 0]]},
"core_required": True,
"cavity_data": {
"mold_cavities": {
"cavity": {"vertex_count": 120},
"core": {"vertex_count": 120},
},
"quality_checks": {
"parting_line_smoothness": 75,
"undercut_regions": [{"id": 1}, {"id": 2}],
"side_actions": {
"summary": {"total_mechanism_count": 2, "complexity": "complex"},
"slider_mechanisms": [{"actuation": "pneumatic"}],
"lifter_mechanisms": [{"actuation": "mechanical"}],
"undercut_analysis": {"total_undercut_area": 1800},
},
},
"manufacturing_info": {
"estimated_mold_size": {"length": 420, "width": 360, "height": 180},
"estimated_clamping_force": "620",
},
"metadata": {"draft_angle": 2.0},
},
"key_info": {
"quality_considerations": {"warpage_risk": "high"},
"geometric_characteristics": {"wall_thickness_range": "0.8-7.0mm"},
},
},
]
ranked = scorer.score_schemes(schemes)
assert ranked[0]["scheme_id"] == "clean"
assert ranked[0]["rank"] == 1
assert ranked[0]["title"] == "推荐方案"
assert ranked[0]["score"] > ranked[1]["score"]
assert ranked[0]["dfm_violation_count"] <= ranked[1]["dfm_violation_count"]
def test_cost_estimate_rules_returns_expected_shape():
analysis_result = {
"geometry_data": {
"bounding_box": {"dimensions": [100, 80, 30]},
"volume": 24000,
}
}
detailed = {
"candidate_schemes": [
{
"cavity_data": {
"mold_cavities": {"cavity_count": 2},
"manufacturing_info": {
"mold_material": "P20",
"estimated_mold_size": {"length": 320, "width": 260, "height": 140},
"estimated_cycle_time": "35秒",
},
"metadata": {"selected_material": "ABS"},
"side_actions": {
"summary": {"total_slider_count": 1, "total_lifter_count": 0}
},
}
}
]
}
result = estimate_cost_by_rules(analysis_result, detailed)
assert result["source"] == "rules"
assert "mold_cost" in result
assert "part_cost" in result
assert result["confidence"] == 0.55
assert result["total_mold_cost"].startswith("¥")
assert result["cost_per_part"].startswith("¥")
def test_redis_task_manager_make_serializable_handles_enums_and_datetime():
from datetime import datetime
from shared.models.schemas import ProcessingStatus
payload = {
"status": ProcessingStatus.COMPLETED,
"completed_at": datetime(2026, 8, 31, 12, 0, 0),
"nested": {"flag": True},
}
serialized = RedisTaskManager._make_serializable(payload)
assert serialized["status"] == "completed"
assert serialized["completed_at"] == "2026-08-31T12:00:00"
assert serialized["nested"] == {"flag": True}