diff --git a/api/routers/export.py b/api/routers/export.py index 2a2f2bf3..1720fa83 100644 --- a/api/routers/export.py +++ b/api/routers/export.py @@ -4,7 +4,9 @@ from fastapi import APIRouter, HTTPException from fastapi.responses import Response, FileResponse -from services.generator_registry import WORKSPACE_DIR +# Import the module (not the name) so WORKSPACE_DIR is read at call time: the +# settings endpoint rebinds it when the user moves the workspace. +import services.generator_registry as registry router = APIRouter(tags=["export"]) @@ -16,8 +18,8 @@ def export_mesh(fmt: str, path: str): if fmt not in SUPPORTED: raise HTTPException(400, f"Unsupported format: {fmt}. Supported: {', '.join(SUPPORTED)}") - full_path = (WORKSPACE_DIR / path).resolve() - if not str(full_path).startswith(str(WORKSPACE_DIR.resolve())): + full_path = (registry.WORKSPACE_DIR / path).resolve() + if not str(full_path).startswith(str(registry.WORKSPACE_DIR.resolve())): raise HTTPException(400, "Invalid path") if not full_path.exists(): raise HTTPException(404, f"File not found: {path}") diff --git a/api/routers/optimize.py b/api/routers/optimize.py index 303def73..f969ecba 100644 --- a/api/routers/optimize.py +++ b/api/routers/optimize.py @@ -11,7 +11,9 @@ from urllib.parse import quote from pydantic import BaseModel, Field -from services.generator_registry import WORKSPACE_DIR +# Import the module (not the name) so WORKSPACE_DIR is read at call time: the +# settings endpoint rebinds it when the user moves the workspace. +import services.generator_registry as registry from services.mesh_ops import ( MeshOpContext, MeshOpExecutionError, @@ -52,8 +54,8 @@ def _resolve_input_path(raw_path: str) -> Path: raise HTTPException(404, f"File not found: {raw_path}") return resolved - resolved = (WORKSPACE_DIR / raw_path).resolve() - if not str(resolved).startswith(str(WORKSPACE_DIR.resolve())): + resolved = (registry.WORKSPACE_DIR / raw_path).resolve() + if not str(resolved).startswith(str(registry.WORKSPACE_DIR.resolve())): raise HTTPException(400, "Invalid path") if not resolved.exists(): raise HTTPException(404, f"File not found: {raw_path}") @@ -61,12 +63,12 @@ def _resolve_input_path(raw_path: str) -> Path: def _operation_output_path(input_path: Path, output_name: str) -> Path: - workspace = WORKSPACE_DIR.resolve() + workspace = registry.WORKSPACE_DIR.resolve() resolved_input = input_path.resolve() output_dir = ( input_path.parent if resolved_input == workspace or workspace in resolved_input.parents - else WORKSPACE_DIR / "Workflows" + else registry.WORKSPACE_DIR / "Workflows" ) output_dir.mkdir(parents=True, exist_ok=True) return output_dir / output_name @@ -80,7 +82,7 @@ def _run_operation( preserve_visuals: bool = False, ) -> MeshOpResult: context = MeshOpContext( - workspace_dir=WORKSPACE_DIR, + workspace_dir=registry.WORKSPACE_DIR, temp_dir=Path(tempfile.gettempdir()), output_path=output_path, preserve_visuals=preserve_visuals, @@ -100,7 +102,7 @@ def _run_operation( def _operation_response(result: MeshOpResult) -> dict[str, object]: output_path = result.file_path.resolve() try: - relative_path = output_path.relative_to(WORKSPACE_DIR.resolve()).as_posix() + relative_path = output_path.relative_to(registry.WORKSPACE_DIR.resolve()).as_posix() except ValueError: payload: dict[str, object] = {"path": str(output_path)} else: @@ -186,12 +188,12 @@ def transform_mesh(body: TransformRequest): stem = input_path.stem output_name = f"{stem}_xf_{uuid.uuid4().hex[:8]}.glb" - output_dir = input_path.parent if str(input_path).startswith(str(WORKSPACE_DIR.resolve())) else WORKSPACE_DIR / "Workflows" + output_dir = input_path.parent if str(input_path).startswith(str(registry.WORKSPACE_DIR.resolve())) else registry.WORKSPACE_DIR / "Workflows" output_dir.mkdir(parents=True, exist_ok=True) output_path = output_dir / output_name loaded.export(str(output_path)) - rel = output_path.relative_to(WORKSPACE_DIR).as_posix() + rel = output_path.relative_to(registry.WORKSPACE_DIR).as_posix() return {"url": f"/workspace/{rel}"} @@ -402,8 +404,8 @@ def export_mesh(path: str, format: str): if format not in ("obj", "stl", "ply"): raise HTTPException(400, "Supported formats: obj, stl, ply") - input_path = (WORKSPACE_DIR / path).resolve() - if not str(input_path).startswith(str(WORKSPACE_DIR.resolve())): + input_path = (registry.WORKSPACE_DIR / path).resolve() + if not str(input_path).startswith(str(registry.WORKSPACE_DIR.resolve())): raise HTTPException(400, "Invalid path") if not input_path.exists(): raise HTTPException(404, f"File not found: {path}") diff --git a/api/tests/test_mesh_routers_workspace.py b/api/tests/test_mesh_routers_workspace.py new file mode 100644 index 00000000..0febfb68 --- /dev/null +++ b/api/tests/test_mesh_routers_workspace.py @@ -0,0 +1,119 @@ +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +import trimesh +from fastapi import HTTPException + +import routers.export as export_router +import routers.optimize as optimize_router +import services.generator_registry as registry +from services.mesh_ops import MeshOpResult + +IDENTITY = [ + [1.0, 0.0, 0.0, 0.0], + [0.0, 1.0, 0.0, 0.0], + [0.0, 0.0, 1.0, 0.0], + [0.0, 0.0, 0.0, 1.0], +] + + +class MeshRoutersAfterWorkspaceMoveTests(unittest.TestCase): + """Export and mesh-edit endpoints must resolve paths against the workspace as + it is *now*. POST /settings/paths rebinds registry.WORKSPACE_DIR when the user + moves the workspace; a name captured at import keeps pointing at the old + folder, so a model generated after the move can't be exported or edited.""" + + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.root = Path(self._tmp.name) + self._prev_ws = registry.WORKSPACE_DIR + # The user moved the workspace: the registry global now points here. + registry.WORKSPACE_DIR = self.root / "new_workspace" + # Keep the test hermetic against import-time bindings: if a router still + # holds its own WORKSPACE_DIR name, point it at an empty "old" folder in + # the temp tree so the assertions -- not the real workspace -- catch it. + (self.root / "old_workspace").mkdir() + self._stale = [] + for module in (export_router, optimize_router): + if hasattr(module, "WORKSPACE_DIR"): + self._stale.append((module, module.WORKSPACE_DIR)) + module.WORKSPACE_DIR = self.root / "old_workspace" + + mesh_dir = registry.WORKSPACE_DIR / "MyColl" + mesh_dir.mkdir(parents=True) + trimesh.creation.box().export(mesh_dir / "mesh.glb") + + def tearDown(self) -> None: + registry.WORKSPACE_DIR = self._prev_ws + for module, value in self._stale: + module.WORKSPACE_DIR = value + self._tmp.cleanup() + + def test_export_router_converts_a_mesh_in_the_moved_workspace(self) -> None: + response = export_router.export_mesh("stl", "MyColl/mesh.glb") + self.assertEqual(response.status_code, 200) + self.assertGreater(len(response.body), 0) + + def test_optimize_export_converts_a_mesh_in_the_moved_workspace(self) -> None: + response = optimize_router.export_mesh(path="MyColl/mesh.glb", format="obj") + self.assertEqual(response.status_code, 200) + self.assertIn(b"v ", response.body) + + def test_transform_writes_its_result_into_the_moved_workspace(self) -> None: + result = optimize_router.transform_mesh( + optimize_router.TransformRequest(path="MyColl/mesh.glb", matrix=IDENTITY) + ) + self.assertTrue(result["url"].startswith("/workspace/MyColl/mesh_xf_")) + written = registry.WORKSPACE_DIR / result["url"].removeprefix("/workspace/") + self.assertTrue(written.is_file()) + + def test_decimate_and_smooth_read_their_input_from_the_moved_workspace(self) -> None: + # /optimize/mesh and /optimize/smooth resolve their input through this helper. + resolved = optimize_router._resolve_input_path("MyColl/mesh.glb") + self.assertEqual(resolved, (registry.WORKSPACE_DIR / "MyColl" / "mesh.glb").resolve()) + + def test_decimate_and_smooth_write_their_result_into_the_moved_workspace(self) -> None: + # The backends (meshoptimizer, pymeshlab) aren't available here, so stand in + # for the mesh-ops registry and check the router's own path handling: the + # workspace it hands the operation, where the output goes, and the URL. + class _RecordingRegistry: + def __init__(self) -> None: + self.contexts = [] + + def run(self, operation_id, input_path, params, context): + self.contexts.append(context) + context.output_path.touch() + return MeshOpResult(context.output_path, {"face_count": 12}) + + ops = _RecordingRegistry() + with patch.object(optimize_router, "mesh_ops_registry", ops): + decimated = optimize_router.optimize_mesh( + optimize_router.OptimizeRequest(path="MyColl/mesh.glb", target_faces=500) + ) + smoothed = optimize_router.smooth_mesh( + optimize_router.SmoothRequest(path="MyColl/mesh.glb", iterations=2) + ) + + self.assertEqual(decimated["url"], "/workspace/MyColl/mesh_opt500.glb") + self.assertEqual(smoothed["url"], "/workspace/MyColl/mesh_smooth2.glb") + for context in ops.contexts: + self.assertEqual(context.workspace_dir, registry.WORKSPACE_DIR) + self.assertEqual(context.output_path.parent, registry.WORKSPACE_DIR / "MyColl") + + def test_a_path_leaving_the_workspace_is_still_refused(self) -> None: + # Reading the live workspace must not loosen the containment check. + trimesh.creation.box().export(self.root / "outside.glb") + calls = ( + lambda: export_router.export_mesh("stl", "../outside.glb"), + lambda: optimize_router.export_mesh(path="../outside.glb", format="obj"), + ) + for call in calls: + with self.assertRaises(HTTPException) as raised: + call() + self.assertEqual(raised.exception.status_code, 400) + + +if __name__ == "__main__": + unittest.main() diff --git a/api/tests/test_optimize_mesh_ops.py b/api/tests/test_optimize_mesh_ops.py index a1f0c5fe..6c702ddb 100644 --- a/api/tests/test_optimize_mesh_ops.py +++ b/api/tests/test_optimize_mesh_ops.py @@ -39,7 +39,7 @@ def test_generic_list_and_run_routes_use_the_shared_registry(self) -> None: registry = _FakeRegistry(output_path) with ( - patch.object(optimize, "WORKSPACE_DIR", workspace), + patch.object(optimize.registry, "WORKSPACE_DIR", workspace), patch.object(optimize, "mesh_ops_registry", registry), ): descriptions = optimize.list_mesh_operations() @@ -75,7 +75,7 @@ def test_legacy_routes_delegate_with_their_existing_clamps_and_names(self) -> No registry = _FakeRegistry(fallback_output) with ( - patch.object(optimize, "WORKSPACE_DIR", workspace), + patch.object(optimize.registry, "WORKSPACE_DIR", workspace), patch.object(optimize, "mesh_ops_registry", registry), ): optimize_response = optimize.optimize_mesh( @@ -115,7 +115,7 @@ def run(self, operation_id, input_path, params, context): input_path = workspace / "input.glb" input_path.touch() with ( - patch.object(optimize, "WORKSPACE_DIR", workspace), + patch.object(optimize.registry, "WORKSPACE_DIR", workspace), patch.object(optimize, "mesh_ops_registry", MissingRegistry()), self.assertRaises(HTTPException) as raised, ): @@ -158,7 +158,7 @@ def operation(input_path, params, context): input_path = workspace / "input.glb" input_path.touch() with ( - patch.object(optimize, "WORKSPACE_DIR", workspace), + patch.object(optimize.registry, "WORKSPACE_DIR", workspace), patch.object(optimize, "mesh_ops_registry", registry), ): optimize.run_mesh_operation( @@ -191,7 +191,7 @@ def _assert_operation_error(self, registry, expected_status: int) -> None: input_path = workspace / "input.glb" input_path.touch() with ( - patch.object(optimize, "WORKSPACE_DIR", workspace), + patch.object(optimize.registry, "WORKSPACE_DIR", workspace), patch.object(optimize, "mesh_ops_registry", registry), self.assertRaises(HTTPException) as raised, ):