Hartmut
1321ef2bd4
The queue handles far more than thumbnails: OCC tessellation, USD master generation, GLB production, order line renders, and workflow renders. asset_pipeline better reflects its role as the render-worker's primary queue. Updated all references in: task decorators, celery_app.py, beat_tasks.py, docker-compose.yml worker command, worker.py MONITORED_QUEUES, admin.py, CLAUDE.md, LEARNINGS.md, Dockerfile, helpTexts.ts, test files, and all .claude/commands/*.md skill files. 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 == "asset_pipeline"
|
|
|
|
|
|
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 == "asset_pipeline"
|
|
|
|
|
|
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 == "asset_pipeline"
|