Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions api/routers/export.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"])

Expand All @@ -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}")
Expand Down
24 changes: 13 additions & 11 deletions api/routers/optimize.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -52,21 +54,21 @@ 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}")
return resolved


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
Expand All @@ -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,
Expand All @@ -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:
Expand Down Expand Up @@ -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}"}


Expand Down Expand Up @@ -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}")
Expand Down
119 changes: 119 additions & 0 deletions api/tests/test_mesh_routers_workspace.py
Original file line number Diff line number Diff line change
@@ -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()
10 changes: 5 additions & 5 deletions api/tests/test_optimize_mesh_ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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,
):
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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,
):
Expand Down