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