Hartmut
a70cb55d01
- workflow_builder.py: fix broken stubs, add render_order_line_still_task
(resolves step_path from DB instead of passing order_line_id as step_path)
- domains/rendering/tasks.py: add render_order_line_still_task,
export_gltf_for_order_line_task, export_blend_for_order_line_task,
generate_gltf_geometry_task (trimesh STL→GLB, no Blender needed)
- tasks/step_tasks.py: add generate_gltf_geometry_task for CadFile GLB export
- cad router: POST /{id}/generate-gltf-geometry endpoint (admin/PM)
- worker router: GET /celery-workers + POST /scale (docker compose subprocess)
- Dockerfile: pip install -e "[dev]" to enable pytest
- docker-compose.yml: docker socket + compose file mount on backend
- ThreeDViewer.tsx: mode toggle (geometry/production), wireframe, env presets,
download buttons (GLB + .blend)
- CadPreview.tsx: load gltf_geometry/gltf_production/blend_production assets
from MediaAsset table and pass URLs to ThreeDViewer
- ProductDetail.tsx: "View 3D" button → /cad/:id, "Generate GLB" button
- media router/service: cad_file_id filter on GET /api/media
- WorkerManagement.tsx: new page with worker status, queue depth, scale controls
- App.tsx + Layout.tsx: /workers route + sidebar link (admin/PM)
- tests: test_rendering_service.py, test_orders_service.py (backend)
- tests: WorkerActivity.test.tsx, WorkerManagement.test.tsx (frontend)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
113 lines
4.6 KiB
Python
113 lines
4.6 KiB
Python
"""Tests for rendering domain — workflow builder + task helpers."""
|
|
import uuid
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# workflow_builder unit tests (no DB required)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_dispatch_workflow_unknown_type_raises():
|
|
from app.domains.rendering.workflow_builder import dispatch_workflow
|
|
with pytest.raises(ValueError, match="Unknown workflow type"):
|
|
dispatch_workflow("nonexistent_type", str(uuid.uuid4()))
|
|
|
|
|
|
def test_build_still_returns_chain():
|
|
"""_build_still returns a Celery chain wrapping render_order_line_still_task."""
|
|
from celery import chain
|
|
from app.domains.rendering.workflow_builder import _build_still
|
|
canvas = _build_still(str(uuid.uuid4()), {})
|
|
# A single-task chain is still a Celery Signature, not a plain chain, but
|
|
# it should be callable / have apply_async
|
|
assert hasattr(canvas, "apply_async")
|
|
|
|
|
|
def test_build_multi_angle_creates_group():
|
|
"""_build_multi_angle returns a Celery group with one sig per angle."""
|
|
from celery import group
|
|
from app.domains.rendering.workflow_builder import _build_multi_angle
|
|
order_line_id = str(uuid.uuid4())
|
|
canvas = _build_multi_angle(order_line_id, {"angles": [0, 90, 180]})
|
|
# group has tasks attribute
|
|
assert hasattr(canvas, "tasks")
|
|
assert len(canvas.tasks) == 3
|
|
|
|
|
|
def test_build_still_with_exports_is_chain():
|
|
"""_build_still_with_exports returns a chain."""
|
|
from app.domains.rendering.workflow_builder import _build_still_with_exports
|
|
canvas = _build_still_with_exports(str(uuid.uuid4()), {})
|
|
assert hasattr(canvas, "apply_async")
|
|
|
|
|
|
def test_build_turntable_raises_without_step_path():
|
|
"""_build_turntable raises ValueError if step_path missing in params."""
|
|
from app.domains.rendering.workflow_builder import _build_turntable
|
|
with pytest.raises(ValueError, match="step_path"):
|
|
_build_turntable(str(uuid.uuid4()), {})
|
|
|
|
|
|
def test_build_turntable_raises_without_output_dir():
|
|
from app.domains.rendering.workflow_builder import _build_turntable
|
|
with pytest.raises(ValueError, match="output_dir"):
|
|
_build_turntable(str(uuid.uuid4()), {"step_path": "/tmp/test.stp"})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _resolve_step_path_for_order_line — unit-tests with DB (integration)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_step_path_returns_none_for_missing_line(db):
|
|
"""Returns (None, None) for a line_id that doesn't exist."""
|
|
from app.domains.rendering.tasks import _resolve_step_path_for_order_line
|
|
import asyncio
|
|
|
|
result = _resolve_step_path_for_order_line(str(uuid.uuid4()))
|
|
assert result == (None, None)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# publish_asset (unit test with mocked DB)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_publish_asset_signature():
|
|
"""publish_asset is importable and is a bound Celery task."""
|
|
from app.domains.rendering.tasks import publish_asset
|
|
assert callable(publish_asset)
|
|
assert hasattr(publish_asset, "delay")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# generate_gltf_geometry_task — smoke test (unit)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_generate_gltf_geometry_task_importable():
|
|
from app.tasks.step_tasks import generate_gltf_geometry_task
|
|
assert callable(generate_gltf_geometry_task)
|
|
assert hasattr(generate_gltf_geometry_task, "delay")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# New order-line tasks are importable and correctly registered
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_render_order_line_still_task_importable():
|
|
from app.domains.rendering.tasks import render_order_line_still_task
|
|
assert render_order_line_still_task.name == "app.domains.rendering.tasks.render_order_line_still_task"
|
|
assert render_order_line_still_task.queue == "thumbnail_rendering"
|
|
|
|
|
|
def test_export_gltf_for_order_line_task_importable():
|
|
from app.domains.rendering.tasks import export_gltf_for_order_line_task
|
|
assert export_gltf_for_order_line_task.queue == "thumbnail_rendering"
|
|
|
|
|
|
def test_export_blend_for_order_line_task_importable():
|
|
from app.domains.rendering.tasks import export_blend_for_order_line_task
|
|
assert export_blend_for_order_line_task.queue == "thumbnail_rendering"
|