From 6da39f941e825ddf2a9f3746fc0a26b2b820f6b9 Mon Sep 17 00:00:00 2001 From: kevin9327 <5299031+kevin9327@users.noreply.github.com> Date: Sun, 13 Sep 2026 06:50:52 +0900 Subject: [PATCH] fix(export,optimize): read WORKSPACE_DIR dynamically so edits work after the workspace moves export.py and optimize.py bound the workspace path with `from services.generator_registry import WORKSPACE_DIR`, capturing it at import time. POST /settings/paths rebinds that module global when the workspace is moved in Settings, but these routers kept resolving paths against the old folder. A model generated after the move could not be exported, decimated, smoothed or transformed ("File not found"), and results for inputs outside the workspace were written into the old folder's Workflows/. Read registry.WORKSPACE_DIR at call time instead, the same way generation.py and ply_to_splat already do. Co-Authored-By: Claude Opus 5 --- api/routers/export.py | 10 +- api/routers/optimize.py | 24 ++--- api/tests/test_export_router.py | 12 +-- api/tests/test_mesh_routers_workspace.py | 119 +++++++++++++++++++++++ api/tests/test_optimize_mesh_ops.py | 10 +- 5 files changed, 149 insertions(+), 26 deletions(-) create mode 100644 api/tests/test_mesh_routers_workspace.py diff --git a/api/routers/export.py b/api/routers/export.py index 6cd03d61..0d8e3429 100644 --- a/api/routers/export.py +++ b/api/routers/export.py @@ -9,7 +9,9 @@ from fastapi.responses import Response, FileResponse from services import imported_sources -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"]) @@ -122,7 +124,7 @@ def _resolve_slicer_source(token: str) -> tuple[Path, str]: # Containment check via ancestry, not string prefix: `startswith` would let a # sibling like `-other/...` slip through, and `..` escapes resolve # outside the workspace and fail this check. - workspace = WORKSPACE_DIR.resolve() + workspace = registry.WORKSPACE_DIR.resolve() full_path = (workspace / decoded).resolve() if full_path != workspace and workspace not in full_path.parents: raise HTTPException(400, "Invalid path") @@ -181,8 +183,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 7ae622c5..843bb596 100644 --- a/api/routers/optimize.py +++ b/api/routers/optimize.py @@ -12,7 +12,9 @@ from pydantic import BaseModel, Field from services import imported_sources -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, @@ -53,8 +55,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}") @@ -62,12 +64,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 @@ -81,7 +83,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, @@ -101,7 +103,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: @@ -187,12 +189,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}"} @@ -408,8 +410,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_export_router.py b/api/tests/test_export_router.py index 0972e836..2bed2a68 100644 --- a/api/tests/test_export_router.py +++ b/api/tests/test_export_router.py @@ -33,8 +33,8 @@ class ExportForSlicerTests(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() self.workspace = Path(self._tmp.name).resolve() - self._orig_workspace = export_router.WORKSPACE_DIR - export_router.WORKSPACE_DIR = self.workspace + self._orig_workspace = export_router.registry.WORKSPACE_DIR + export_router.registry.WORKSPACE_DIR = self.workspace # A box that is tallest along Y (glTF up-axis) and unit-sized, matching # what image-to-3D generators emit. Exported to GLB, it reloads as a # Scene so the flatten path is exercised too. @@ -44,7 +44,7 @@ def setUp(self) -> None: box.export(str(self.workspace / self.rel)) def tearDown(self) -> None: - export_router.WORKSPACE_DIR = self._orig_workspace + export_router.registry.WORKSPACE_DIR = self._orig_workspace self._tmp.cleanup() def test_converts_glb_to_stl_with_download_filename(self) -> None: @@ -154,17 +154,17 @@ class ImportedSourceSlicerTests(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() self.outside = Path(self._tmp.name).resolve() - self._orig_workspace = export_router.WORKSPACE_DIR + self._orig_workspace = export_router.registry.WORKSPACE_DIR # A workspace elsewhere, so nothing here is reachable as a relative path. self._ws_tmp = tempfile.TemporaryDirectory() - export_router.WORKSPACE_DIR = Path(self._ws_tmp.name).resolve() + export_router.registry.WORKSPACE_DIR = Path(self._ws_tmp.name).resolve() imported_sources.clear() # Unit-sized and tallest along Y, as a glTF export would be. self.mesh_path = self.outside / "imported.glb" trimesh.creation.box(extents=[0.3, 1.0, 0.3]).export(str(self.mesh_path)) def tearDown(self) -> None: - export_router.WORKSPACE_DIR = self._orig_workspace + export_router.registry.WORKSPACE_DIR = self._orig_workspace imported_sources.clear() self._tmp.cleanup() self._ws_tmp.cleanup() 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, ):