"""Tests for rendering domain — workflow builder + task helpers.""" import uuid from pathlib import Path from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest from app.core.render_paths import build_order_line_export_path, build_order_line_step_render_path # --------------------------------------------------------------------------- # 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") def test_build_ffmpeg_cmd_prefers_prefixed_frame_sequence(tmp_path): from app.domains.rendering.tasks import _build_ffmpeg_cmd frames_dir = tmp_path / "frames" frames_dir.mkdir() (frames_dir / "frame_0001.png").write_text("png", encoding="utf-8") cmd = _build_ffmpeg_cmd(frames_dir, tmp_path / "turntable.mp4") assert any("frame_%04d.png" in part for part in cmd) def test_build_ffmpeg_cmd_falls_back_to_legacy_frame_sequence(tmp_path): from app.domains.rendering.tasks import _build_ffmpeg_cmd frames_dir = tmp_path / "frames" frames_dir.mkdir() cmd = _build_ffmpeg_cmd(frames_dir, tmp_path / "turntable.mp4") assert any(part.endswith("%04d.png") for part in cmd) assert not any("frame_%04d.png" in part for part in cmd) def test_build_ffmpeg_cmd_limits_bg_color_overlay_to_frame_sequence(tmp_path): from app.domains.rendering.tasks import _build_ffmpeg_cmd frames_dir = tmp_path / "frames" frames_dir.mkdir() (frames_dir / "frame_0001.png").write_text("png", encoding="utf-8") cmd = _build_ffmpeg_cmd( frames_dir, tmp_path / "turntable.mp4", fps=24, bg_color="#ffffff", width=512, height=512, ) assert "-filter_complex" in cmd filter_index = cmd.index("-filter_complex") + 1 assert cmd[filter_index] == "[1:v][0:v]overlay=0:0:shortest=1" assert "color=c=0xffffff:size=512x512:rate=24" in cmd assert "-crf" in cmd assert cmd[cmd.index("-crf") + 1] == "18" # --------------------------------------------------------------------------- # 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_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" def test_normalize_order_line_still_params_maps_legacy_editor_fields(): from app.domains.rendering.tasks import _normalize_order_line_still_params normalized = _normalize_order_line_still_params( { "render_engine": "cycles", "samples": 256, "resolution": [1920, 1080], "rotation_z": 45, "usd_path": "/app/uploads/step_files/example_master.usd", } ) assert normalized["engine"] == "cycles" assert "render_engine" not in normalized assert normalized["width"] == 1920 assert normalized["height"] == 1080 assert "resolution" not in normalized assert normalized["rotation_z"] == 45 assert str(normalized["usd_path"]) == "/app/uploads/step_files/example_master.usd" def test_normalize_order_line_still_params_drops_graph_only_override_flag(): from app.domains.rendering.tasks import _normalize_order_line_still_params normalized = _normalize_order_line_still_params( { "use_custom_render_settings": True, "width": 640, "height": 640, } ) assert "use_custom_render_settings" not in normalized assert normalized["width"] == 640 assert normalized["height"] == 640 def test_normalize_order_line_still_params_drops_graph_control_params(): from app.domains.rendering.tasks import _normalize_order_line_still_params normalized = _normalize_order_line_still_params( { "width": 640, "graph_notify_node_ids": ["notify"], "graph_output_node_ids": ["output"], "workflow_run_id": "run-1", "workflow_node_id": "render", "emit_legacy_notifications": True, } ) assert normalized == {"width": 640} def test_resolve_order_line_still_output_extension_uses_output_type_contract(monkeypatch): from app.domains.rendering.tasks import _resolve_order_line_still_output_extension line = SimpleNamespace(output_type=SimpleNamespace(output_format="webp"), render_overrides=None) db = MagicMock() db.execute.return_value.scalar_one_or_none.return_value = line class _SessionCtx: def __enter__(self): return db def __exit__(self, exc_type, exc, tb): return False monkeypatch.setattr("app.core.db_utils.get_sync_session", lambda: _SessionCtx()) assert _resolve_order_line_still_output_extension("line-1") == "webp" def test_render_order_line_still_task_finalizes_non_png_outputs(tmp_path, monkeypatch): from app.domains.rendering.tasks import render_order_line_still_task step_path = tmp_path / "cad" / "bearing.step" step_path.parent.mkdir(parents=True, exist_ok=True) step_path.write_text("STEP", encoding="utf-8") task_self = SimpleNamespace(request=SimpleNamespace(id="task-webp-save")) render_calls: list[dict] = [] def _fake_render_to_file(step_path, output_path, **kwargs): render_calls.append( { "step_path": step_path, "output_path": output_path, **kwargs, } ) Path(output_path).parent.mkdir(parents=True, exist_ok=True) Path(output_path).write_text("WEBP", encoding="utf-8") return True, {"renderer": "blender", "engine_used": "cycles", "total_duration_s": 0.5} monkeypatch.setattr( "app.domains.rendering.tasks._resolve_step_path_for_order_line", lambda order_line_id: (str(step_path), "cad-1"), ) monkeypatch.setattr( "app.domains.rendering.tasks._resolve_order_line_still_output_extension", lambda order_line_id, params=None: "webp", ) monkeypatch.setattr("app.services.step_processor.render_to_file", _fake_render_to_file) monkeypatch.setattr("app.domains.rendering.tasks._update_workflow_run_status", lambda *args, **kwargs: None) monkeypatch.setattr("app.domains.rendering.tasks._finalize_graph_still_output", lambda *args, **kwargs: None) result = render_order_line_still_task.run.__func__( task_self, "line-webp", job_document_enabled=False, emit_events=False, publish_asset_enabled=False, ) expected_output = build_order_line_step_render_path(step_path, "line-webp", "line_line-webp.webp") assert render_calls == [ { "step_path": str(step_path), "output_path": str(expected_output), "order_line_id": "line-webp", } ] assert result["output_path"] == str(expected_output) def test_finalise_image_converts_png_to_jpg(tmp_path): from PIL import Image from app.services.step_processor import _finalise_image src = tmp_path / "source.png" dst = tmp_path / "final.jpg" Image.new("RGBA", (4, 4), (10, 20, 30, 255)).save(src, "PNG") result = _finalise_image(src, dst) assert result == dst assert dst.exists() assert not src.exists() with Image.open(dst) as img: assert img.format == "JPEG" assert img.mode == "RGB" def test_finalise_image_flattens_transparency_for_jpg(tmp_path): from PIL import Image from app.services.step_processor import _finalise_image src = tmp_path / "source.png" dst = tmp_path / "final.jpg" Image.new("RGBA", (2, 2), (0, 0, 0, 0)).save(src, "PNG") result = _finalise_image(src, dst) assert result == dst with Image.open(dst) as img: assert img.format == "JPEG" assert img.getpixel((0, 0)) == (255, 255, 255) assert img.getpixel((1, 1)) == (255, 255, 255) def test_render_order_line_still_task_uses_graph_authoritative_output_handoff(tmp_path, monkeypatch): from app.domains.rendering.tasks import publish_asset, render_order_line_still_task step_path = tmp_path / "cad" / "bearing.step" step_path.parent.mkdir(parents=True, exist_ok=True) step_path.write_text("STEP", encoding="utf-8") task_self = SimpleNamespace(request=SimpleNamespace(id="task-graph-save")) finalize_calls: list[dict] = [] notify_finalize_calls: list[dict] = [] notify_emit_calls: list[dict] = [] publish_calls: list[tuple] = [] render_calls: list[dict] = [] def _fake_render_to_file(step_path, output_path, **kwargs): render_calls.append( { "step_path": step_path, "output_path": output_path, **kwargs, } ) Path(output_path).parent.mkdir(parents=True, exist_ok=True) Path(output_path).write_text("PNG", encoding="utf-8") return True, {"renderer": "blender", "engine_used": "cycles", "total_duration_s": 0.5} monkeypatch.setattr( "app.domains.rendering.tasks._resolve_step_path_for_order_line", lambda order_line_id: (str(step_path), "cad-1"), ) monkeypatch.setattr( "app.domains.rendering.tasks._resolve_order_line_still_output_extension", lambda order_line_id, params=None: "png", ) monkeypatch.setattr("app.services.step_processor.render_to_file", _fake_render_to_file) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_graph_still_output", lambda order_line_id, **kwargs: finalize_calls.append( {"order_line_id": order_line_id, **kwargs} ), ) monkeypatch.setattr( "app.domains.rendering.tasks._update_workflow_run_status", lambda *args, **kwargs: None, ) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_graph_notify_nodes", lambda **kwargs: notify_finalize_calls.append(kwargs), ) monkeypatch.setattr( "app.domains.rendering.tasks._emit_graph_render_notifications", lambda order_line_id, **kwargs: notify_emit_calls.append( {"order_line_id": order_line_id, **kwargs} ), ) monkeypatch.setattr( publish_asset, "delay", lambda *args, **kwargs: publish_calls.append((args, kwargs)), ) result = render_order_line_still_task.run.__func__( task_self, "line-1", workflow_run_id="run-1", workflow_node_id="render", publish_asset_enabled=False, graph_authoritative_output_enabled=True, graph_output_node_ids=["output"], graph_notify_node_ids=["notify"], job_document_enabled=False, emit_events=False, emit_legacy_notifications=True, width=640, height=480, ) expected_output = build_order_line_step_render_path(step_path, "line-1", "line_line-1.png") assert result["renderer"] == "blender" assert publish_calls == [] assert render_calls == [ { "step_path": str(step_path), "output_path": str(expected_output), "order_line_id": "line-1", "width": 640, "height": 480, } ] assert finalize_calls == [ { "order_line_id": "line-1", "success": True, "output_path": str(expected_output), "render_log": result, "workflow_run_id": "run-1", "output_node_ids": ["output"], "render_node_id": "render", } ] assert notify_emit_calls == [ { "order_line_id": "line-1", "success": True, "render_log": result, } ] assert notify_finalize_calls == [ { "workflow_run_id": "run-1", "notify_node_ids": ["notify"], "success": True, "render_node_id": "render", } ] def test_render_order_line_still_task_uses_shadow_observer_output_handoff(tmp_path, monkeypatch): from app.domains.rendering.tasks import publish_asset, render_order_line_still_task step_path = tmp_path / "cad" / "bearing.step" step_path.parent.mkdir(parents=True, exist_ok=True) step_path.write_text("STEP", encoding="utf-8") task_self = SimpleNamespace(request=SimpleNamespace(id="task-shadow-save")) observer_finalize_calls: list[dict] = [] publish_calls: list[tuple] = [] def _fake_render_to_file(step_path, output_path, **kwargs): Path(output_path).parent.mkdir(parents=True, exist_ok=True) Path(output_path).write_text("PNG", encoding="utf-8") return True, {"renderer": "blender", "engine_used": "cycles", "total_duration_s": 0.5} monkeypatch.setattr( "app.domains.rendering.tasks._resolve_step_path_for_order_line", lambda order_line_id: (str(step_path), "cad-1"), ) monkeypatch.setattr( "app.domains.rendering.tasks._resolve_order_line_still_output_extension", lambda order_line_id, params=None: "png", ) monkeypatch.setattr("app.services.step_processor.render_to_file", _fake_render_to_file) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_shadow_still_output", lambda order_line_id, **kwargs: observer_finalize_calls.append( {"order_line_id": order_line_id, **kwargs} ), ) monkeypatch.setattr( "app.domains.rendering.tasks._update_workflow_run_status", lambda *args, **kwargs: None, ) monkeypatch.setattr( publish_asset, "delay", lambda *args, **kwargs: publish_calls.append((args, kwargs)), ) result = render_order_line_still_task.run.__func__( task_self, "line-shadow", workflow_run_id="run-shadow", workflow_node_id="render", publish_asset_enabled=False, observer_output_enabled=True, graph_output_node_ids=["output"], job_document_enabled=False, emit_events=False, ) expected_output = build_order_line_step_render_path(step_path, "line-shadow", "line_line-shadow.png") assert result["renderer"] == "blender" assert publish_calls == [] assert observer_finalize_calls == [ { "order_line_id": "line-shadow", "success": True, "output_path": str(expected_output), "render_log": result, "workflow_run_id": "run-shadow", "output_node_ids": ["output"], "render_node_id": "render", } ] def test_render_order_line_still_task_publishes_asset_without_graph_authoritative_handoff( tmp_path, monkeypatch, ): from app.domains.rendering.tasks import publish_asset, render_order_line_still_task step_path = tmp_path / "cad" / "bearing.step" step_path.parent.mkdir(parents=True, exist_ok=True) step_path.write_text("STEP", encoding="utf-8") task_self = SimpleNamespace(request=SimpleNamespace(id="task-publish")) finalize_calls: list[dict] = [] publish_calls: list[tuple] = [] def _fake_render_to_file(step_path, output_path, **kwargs): Path(output_path).parent.mkdir(parents=True, exist_ok=True) Path(output_path).write_text("PNG", encoding="utf-8") return True, { "renderer": "blender", "engine_used": "cycles", "total_duration_s": 0.5, } monkeypatch.setattr( "app.domains.rendering.tasks._resolve_step_path_for_order_line", lambda order_line_id: (str(step_path), "cad-1"), ) monkeypatch.setattr("app.services.step_processor.render_to_file", _fake_render_to_file) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_graph_still_output", lambda *args, **kwargs: finalize_calls.append(kwargs), ) monkeypatch.setattr( "app.domains.rendering.tasks._update_workflow_run_status", lambda *args, **kwargs: None, ) monkeypatch.setattr( publish_asset, "delay", lambda *args, **kwargs: publish_calls.append((args, kwargs)), ) render_order_line_still_task.run.__func__( task_self, "line-2", workflow_run_id="run-2", workflow_node_id="render", publish_asset_enabled=True, graph_authoritative_output_enabled=False, job_document_enabled=False, emit_events=False, width=640, height=480, ) expected_output = build_order_line_step_render_path(step_path, "line-2", "line_line-2.png") assert finalize_calls == [] assert publish_calls == [ ( ( "line-2", "still", str(expected_output), ), { "render_config": { "renderer": "blender", "engine_used": "cycles", "total_duration_s": 0.5, "output_path": str(expected_output), }, "workflow_run_id": "run-2", }, ) ] def test_export_blend_for_order_line_task_uses_graph_authoritative_output_handoff( tmp_path, monkeypatch, ): from app.config import settings from app.domains.rendering.tasks import export_blend_for_order_line_task, publish_asset step_path = tmp_path / "cad" / "bearing.step" step_path.parent.mkdir(parents=True, exist_ok=True) step_path.write_text("STEP", encoding="utf-8") glb_path = step_path.parent / "bearing_render.glb" glb_path.write_text("GLB", encoding="utf-8") task_self = SimpleNamespace(request=SimpleNamespace(id="task-blend-graph")) finalize_calls: list[dict] = [] notify_finalize_calls: list[dict] = [] notify_emit_calls: list[dict] = [] publish_calls: list[tuple] = [] monkeypatch.setattr(settings, "upload_dir", str(tmp_path / "uploads")) monkeypatch.setattr( "app.domains.rendering.tasks._resolve_step_path_for_order_line", lambda order_line_id: (str(step_path), "cad-1"), ) monkeypatch.setattr( "app.services.render_blender.find_blender", lambda: "/usr/bin/blender", ) monkeypatch.setattr( "app.services.render_blender.build_tessellated_glb_path", lambda *args, **kwargs: glb_path, ) monkeypatch.setattr( "subprocess.run", lambda *args, **kwargs: SimpleNamespace(returncode=0, stderr=""), ) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_graph_blend_output", lambda order_line_id, **kwargs: finalize_calls.append( {"order_line_id": order_line_id, **kwargs} ), ) monkeypatch.setattr( "app.domains.rendering.tasks._update_workflow_run_status", lambda *args, **kwargs: None, ) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_graph_notify_nodes", lambda **kwargs: notify_finalize_calls.append(kwargs), ) monkeypatch.setattr( "app.domains.rendering.tasks._emit_graph_render_notifications", lambda order_line_id, **kwargs: notify_emit_calls.append( {"order_line_id": order_line_id, **kwargs} ), ) monkeypatch.setattr( publish_asset, "delay", lambda *args, **kwargs: publish_calls.append((args, kwargs)), ) result = export_blend_for_order_line_task.run.__func__( task_self, "line-3", workflow_run_id="run-3", workflow_node_id="blend", publish_asset_enabled=False, graph_authoritative_output_enabled=True, graph_output_node_ids=["output"], graph_notify_node_ids=["notify"], emit_legacy_notifications=True, ) expected_blend_path = build_order_line_export_path("line-3", "bearing_production.blend") assert result == { "blend_path": str(expected_blend_path), "artifact_type": "blend_production", } assert publish_calls == [] assert finalize_calls == [ { "order_line_id": "line-3", "success": True, "output_path": str(expected_blend_path), "render_log": result, "workflow_run_id": "run-3", "output_node_ids": ["output"], "render_node_id": "blend", } ] assert notify_emit_calls == [ { "order_line_id": "line-3", "success": True, "render_log": result, } ] assert notify_finalize_calls == [ { "workflow_run_id": "run-3", "notify_node_ids": ["notify"], "success": True, "render_node_id": "blend", } ] def test_export_blend_for_order_line_task_publishes_without_graph_authoritative_handoff( tmp_path, monkeypatch, ): from app.config import settings from app.domains.rendering.tasks import export_blend_for_order_line_task, publish_asset step_path = tmp_path / "cad" / "bearing.step" step_path.parent.mkdir(parents=True, exist_ok=True) step_path.write_text("STEP", encoding="utf-8") glb_path = step_path.parent / "bearing_render.glb" glb_path.write_text("GLB", encoding="utf-8") task_self = SimpleNamespace(request=SimpleNamespace(id="task-blend-publish")) finalize_calls: list[dict] = [] publish_calls: list[tuple] = [] monkeypatch.setattr(settings, "upload_dir", str(tmp_path / "uploads")) monkeypatch.setattr( "app.domains.rendering.tasks._resolve_step_path_for_order_line", lambda order_line_id: (str(step_path), "cad-1"), ) monkeypatch.setattr( "app.services.render_blender.find_blender", lambda: "/usr/bin/blender", ) monkeypatch.setattr( "app.services.render_blender.build_tessellated_glb_path", lambda *args, **kwargs: glb_path, ) monkeypatch.setattr( "subprocess.run", lambda *args, **kwargs: SimpleNamespace(returncode=0, stderr=""), ) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_graph_blend_output", lambda *args, **kwargs: finalize_calls.append(kwargs), ) monkeypatch.setattr( "app.domains.rendering.tasks._update_workflow_run_status", lambda *args, **kwargs: None, ) monkeypatch.setattr( publish_asset, "delay", lambda *args, **kwargs: publish_calls.append((args, kwargs)), ) export_blend_for_order_line_task.run.__func__( task_self, "line-4", workflow_run_id="run-4", workflow_node_id="blend", publish_asset_enabled=True, graph_authoritative_output_enabled=False, ) expected_blend_path = build_order_line_export_path("line-4", "bearing_production.blend") assert finalize_calls == [] assert publish_calls == [ ( ( "line-4", "blend_production", str(expected_blend_path), ), { "workflow_run_id": "run-4", }, ) ] def test_render_turntable_task_uses_graph_authoritative_output_handoff( tmp_path, monkeypatch, ): from app.domains.rendering.tasks import publish_asset, render_turntable_task step_path = tmp_path / "cad" / "bearing.step" step_path.parent.mkdir(parents=True, exist_ok=True) step_path.write_text("STEP", encoding="utf-8") task_self = SimpleNamespace(request=SimpleNamespace(id="task-turntable-graph")) expected_output_mp4 = build_order_line_step_render_path(step_path, "line-5", "preview.mp4") finalize_calls: list[dict] = [] notify_finalize_calls: list[dict] = [] notify_emit_calls: list[dict] = [] publish_calls: list[tuple] = [] workflow_status_calls: list[tuple] = [] blender_calls: list[list[str]] = [] def _fake_subprocess_run(cmd, *args, **kwargs): cmd_text = " ".join(str(part) for part in cmd) if "export_step_to_gltf.py" in cmd_text: glb_path = step_path.parent / "bearing_thumbnail.glb" glb_path.write_text("GLB", encoding="utf-8") elif "turntable_render.py" in cmd_text: blender_calls.append([str(part) for part in cmd]) frames_dir = Path(cmd[cmd.index("--") + 2]) frames_dir.mkdir(parents=True, exist_ok=True) (frames_dir / "frame_0001.png").write_text("PNG", encoding="utf-8") if "ffmpeg" in cmd_text: output_mp4 = expected_output_mp4 output_mp4.parent.mkdir(parents=True, exist_ok=True) output_mp4.write_text("MP4", encoding="utf-8") return SimpleNamespace(returncode=0, stdout="", stderr="") monkeypatch.setattr( "app.domains.rendering.tasks._resolve_step_path_for_order_line", lambda order_line_id: (str(step_path), "cad-1"), ) monkeypatch.setattr( "app.services.render_blender.find_blender", lambda: "/usr/bin/blender", ) monkeypatch.setattr( "app.domains.rendering.tasks._build_ffmpeg_cmd", lambda *args, **kwargs: ["ffmpeg", str(expected_output_mp4)], ) monkeypatch.setattr( "subprocess.run", _fake_subprocess_run, ) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_graph_turntable_output", lambda order_line_id, **kwargs: finalize_calls.append( {"order_line_id": order_line_id, **kwargs} ), ) monkeypatch.setattr( "app.domains.rendering.tasks._update_workflow_run_status", lambda *args, **kwargs: workflow_status_calls.append((args, kwargs)), ) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_graph_notify_nodes", lambda **kwargs: notify_finalize_calls.append(kwargs), ) monkeypatch.setattr( "app.domains.rendering.tasks._emit_graph_render_notifications", lambda order_line_id, **kwargs: notify_emit_calls.append( {"order_line_id": order_line_id, **kwargs} ), ) monkeypatch.setattr( publish_asset, "delay", lambda *args, **kwargs: publish_calls.append((args, kwargs)), ) result = render_turntable_task.run.__func__( task_self, "line-5", output_name="preview", workflow_run_id="run-5", workflow_node_id="turntable", publish_asset_enabled=False, graph_authoritative_output_enabled=True, graph_output_node_ids=["output"], graph_notify_node_ids=["notify"], emit_legacy_notifications=True, emit_events=False, ) assert result == { "output_mp4": str(expected_output_mp4), "frame_count": 120, "fps": 30, } assert publish_calls == [] assert finalize_calls == [ { "order_line_id": "line-5", "success": True, "output_path": str(expected_output_mp4), "render_log": result, "workflow_run_id": "run-5", "output_node_ids": ["output"], "render_node_id": "turntable", } ] assert workflow_status_calls == [ ( ("line-5", "completed"), {"workflow_run_id": "run-5", "workflow_node_id": "turntable"}, ) ] assert notify_emit_calls == [ { "order_line_id": "line-5", "success": True, "render_log": result, } ] assert notify_finalize_calls == [ { "workflow_run_id": "run-5", "notify_node_ids": ["notify"], "success": True, "render_node_id": "turntable", } ] assert len(blender_calls) == 1 args = blender_calls[0][blender_calls[0].index("--") + 1 :] assert Path(args[0]).parent == step_path.parent assert Path(args[0]).suffix == ".glb" expected_frames_dir = expected_output_mp4.parent / "_frames_preview" assert args[1:9] == [ str(expected_frames_dir), "120", "360", "1920", "1080", "cycles", "64", "{}", ] assert args[9:17] == [ "", "Product", "", "{}", "[]", "0", "gpu", "0", ] assert args[17:23] == [ "0.0", "0.0", "0.0", "world_z", "", "0", ] def test_render_turntable_task_uses_shadow_observer_output_handoff( tmp_path, monkeypatch, ): from app.domains.rendering.tasks import publish_asset, render_turntable_task step_path = tmp_path / "cad" / "bearing.step" step_path.parent.mkdir(parents=True, exist_ok=True) step_path.write_text("STEP", encoding="utf-8") task_self = SimpleNamespace(request=SimpleNamespace(id="task-turntable-shadow")) expected_output_mp4 = build_order_line_step_render_path( step_path, "line-shadow", "preview_shadow-abcd1234.mp4", ) observer_finalize_calls: list[dict] = [] publish_calls: list[tuple] = [] def _fake_subprocess_run(cmd, *args, **kwargs): cmd_text = " ".join(str(part) for part in cmd) if "export_step_to_gltf.py" in cmd_text: glb_path = step_path.parent / "bearing_thumbnail.glb" glb_path.write_text("GLB", encoding="utf-8") elif "turntable_render.py" in cmd_text: frames_dir = Path(cmd[cmd.index("--") + 2]) frames_dir.mkdir(parents=True, exist_ok=True) (frames_dir / "frame_0001.png").write_text("PNG", encoding="utf-8") if "ffmpeg" in cmd_text: output_mp4 = expected_output_mp4 output_mp4.parent.mkdir(parents=True, exist_ok=True) output_mp4.write_text("MP4", encoding="utf-8") return SimpleNamespace(returncode=0, stdout="", stderr="") monkeypatch.setattr( "app.domains.rendering.tasks._resolve_step_path_for_order_line", lambda order_line_id: (str(step_path), "cad-1"), ) monkeypatch.setattr( "app.services.render_blender.find_blender", lambda: "/usr/bin/blender", ) monkeypatch.setattr( "app.domains.rendering.tasks._build_ffmpeg_cmd", lambda *args, **kwargs: ["ffmpeg", str(expected_output_mp4)], ) monkeypatch.setattr("subprocess.run", _fake_subprocess_run) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_shadow_turntable_output", lambda order_line_id, **kwargs: observer_finalize_calls.append( {"order_line_id": order_line_id, **kwargs} ), ) monkeypatch.setattr( "app.domains.rendering.tasks._update_workflow_run_status", lambda *args, **kwargs: None, ) monkeypatch.setattr( publish_asset, "delay", lambda *args, **kwargs: publish_calls.append((args, kwargs)), ) result = render_turntable_task.run.__func__( task_self, "line-shadow", output_name="preview", output_name_suffix="shadow-abcd1234", workflow_run_id="run-shadow", workflow_node_id="turntable", publish_asset_enabled=False, observer_output_enabled=True, graph_output_node_ids=["output"], emit_events=False, ) assert result == { "output_mp4": str(expected_output_mp4), "frame_count": 120, "fps": 30, } assert publish_calls == [] assert observer_finalize_calls == [ { "order_line_id": "line-shadow", "success": True, "output_path": str(expected_output_mp4), "render_log": result, "workflow_run_id": "run-shadow", "output_node_ids": ["output"], "render_node_id": "turntable", } ] def test_render_turntable_task_uses_isolated_frames_dir_per_output_name( tmp_path, monkeypatch, ): from app.domains.rendering.tasks import render_turntable_task step_path = tmp_path / "cad" / "bearing.step" step_path.parent.mkdir(parents=True, exist_ok=True) step_path.write_text("STEP", encoding="utf-8") task_self = SimpleNamespace(request=SimpleNamespace(id="task-turntable-isolated-frames")) expected_output_mp4 = build_order_line_step_render_path( step_path, "line-shadow", "preview_shadow-run123.mp4", ) expected_frames_dir = expected_output_mp4.parent / "_frames_preview_shadow-run123" blender_calls: list[list[str]] = [] ffmpeg_calls: list[tuple[Path, Path, dict]] = [] def _fake_subprocess_run(cmd, *args, **kwargs): cmd_text = " ".join(str(part) for part in cmd) if "export_step_to_gltf.py" in cmd_text: glb_path = step_path.parent / "bearing_thumbnail.glb" glb_path.write_text("GLB", encoding="utf-8") elif "turntable_render.py" in cmd_text: blender_calls.append([str(part) for part in cmd]) frames_dir = Path(cmd[cmd.index("--") + 2]) frames_dir.mkdir(parents=True, exist_ok=True) (frames_dir / "frame_0001.png").write_text("fresh", encoding="utf-8") elif "ffmpeg" in cmd_text: output_mp4 = expected_output_mp4 output_mp4.parent.mkdir(parents=True, exist_ok=True) output_mp4.write_text("MP4", encoding="utf-8") return SimpleNamespace(returncode=0, stdout="", stderr="") def _fake_build_ffmpeg_cmd(frames_dir, output_mp4, **kwargs): ffmpeg_calls.append((frames_dir, output_mp4, kwargs)) return ["ffmpeg", str(output_mp4)] stale_frame_dir = expected_frames_dir stale_frame_dir.mkdir(parents=True, exist_ok=True) (stale_frame_dir / "frame_0001.png").write_text("stale", encoding="utf-8") monkeypatch.setattr( "app.domains.rendering.tasks._resolve_step_path_for_order_line", lambda order_line_id: (str(step_path), "cad-1"), ) monkeypatch.setattr( "app.services.render_blender.find_blender", lambda: "/usr/bin/blender", ) monkeypatch.setattr("app.domains.rendering.tasks._build_ffmpeg_cmd", _fake_build_ffmpeg_cmd) monkeypatch.setattr("subprocess.run", _fake_subprocess_run) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_shadow_turntable_output", lambda *args, **kwargs: None, ) monkeypatch.setattr( "app.domains.rendering.tasks._update_workflow_run_status", lambda *args, **kwargs: None, ) render_turntable_task.run.__func__( task_self, "line-shadow", output_name="preview", output_name_suffix="shadow-run123", workflow_run_id="run-shadow", workflow_node_id="turntable", publish_asset_enabled=False, observer_output_enabled=True, emit_events=False, ) assert len(blender_calls) == 1 args = blender_calls[0][blender_calls[0].index("--") + 1 :] assert args[1] == str(stale_frame_dir) assert stale_frame_dir.exists() assert list(stale_frame_dir.iterdir()) == [stale_frame_dir / "frame_0001.png"] assert (stale_frame_dir / "frame_0001.png").read_text(encoding="utf-8") == "fresh" assert ffmpeg_calls == [ ( stale_frame_dir, expected_output_mp4, {"fps": 30, "bg_color": "", "width": 1920, "height": 1080}, ) ] def test_render_turntable_task_preserves_legacy_step_path_signature( tmp_path, monkeypatch, ): from app.domains.rendering.tasks import publish_asset, render_turntable_task step_path = tmp_path / "cad" / "bearing.step" output_dir = step_path.parent / "legacy-renders" step_path.parent.mkdir(parents=True, exist_ok=True) step_path.write_text("STEP", encoding="utf-8") task_self = SimpleNamespace(request=SimpleNamespace(id="task-turntable-legacy")) finalize_calls: list[dict] = [] publish_calls: list[tuple] = [] workflow_status_calls: list[tuple] = [] blender_calls: list[list[str]] = [] def _fake_subprocess_run(cmd, *args, **kwargs): cmd_text = " ".join(str(part) for part in cmd) if "export_step_to_gltf.py" in cmd_text: glb_path = step_path.parent / "bearing_thumbnail.glb" glb_path.write_text("GLB", encoding="utf-8") elif "turntable_render.py" in cmd_text: blender_calls.append([str(part) for part in cmd]) frames_dir = Path(cmd[cmd.index("--") + 2]) frames_dir.mkdir(parents=True, exist_ok=True) (frames_dir / "frame_0001.png").write_text("PNG", encoding="utf-8") if "ffmpeg" in cmd_text: output_mp4 = output_dir / "turntable.mp4" output_mp4.parent.mkdir(parents=True, exist_ok=True) output_mp4.write_text("MP4", encoding="utf-8") return SimpleNamespace(returncode=0, stdout="", stderr="") monkeypatch.setattr( "app.services.render_blender.find_blender", lambda: "/usr/bin/blender", ) monkeypatch.setattr( "app.domains.rendering.tasks._build_ffmpeg_cmd", lambda *args, **kwargs: ["ffmpeg", str(output_dir / "turntable.mp4")], ) monkeypatch.setattr( "subprocess.run", _fake_subprocess_run, ) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_graph_turntable_output", lambda *args, **kwargs: finalize_calls.append(kwargs), ) monkeypatch.setattr( "app.domains.rendering.tasks._update_workflow_run_status", lambda *args, **kwargs: workflow_status_calls.append((args, kwargs)), ) monkeypatch.setattr( publish_asset, "delay", lambda *args, **kwargs: publish_calls.append((args, kwargs)), ) result = render_turntable_task.run.__func__( task_self, str(step_path), str(output_dir), emit_events=False, ) assert result == { "output_mp4": str(output_dir / "turntable.mp4"), "frame_count": 120, "fps": 30, } assert len(blender_calls) == 1 assert "--camera-orbit" in blender_calls[0] assert finalize_calls == [] assert publish_calls == [] assert workflow_status_calls == [] def test_render_order_line_still_task_finalizes_notify_handoff_on_failure(tmp_path, monkeypatch): from app.domains.rendering.tasks import render_order_line_still_task step_path = tmp_path / "cad" / "bearing.step" step_path.parent.mkdir(parents=True, exist_ok=True) step_path.write_text("STEP", encoding="utf-8") task_self = SimpleNamespace( request=SimpleNamespace(id="task-still-failure"), retry=lambda *, exc, countdown: (_ for _ in ()).throw(exc), ) notify_finalize_calls: list[dict] = [] notify_emit_calls: list[dict] = [] monkeypatch.setattr( "app.domains.rendering.tasks._resolve_step_path_for_order_line", lambda order_line_id: (str(step_path), "cad-1"), ) monkeypatch.setattr( "app.services.step_processor.render_to_file", lambda **kwargs: (_ for _ in ()).throw(RuntimeError("still boom")), ) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_graph_still_output", lambda *args, **kwargs: None, ) monkeypatch.setattr( "app.domains.rendering.tasks._update_workflow_run_status", lambda *args, **kwargs: None, ) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_graph_notify_nodes", lambda **kwargs: notify_finalize_calls.append(kwargs), ) monkeypatch.setattr( "app.domains.rendering.tasks._emit_graph_render_notifications", lambda order_line_id, **kwargs: notify_emit_calls.append( {"order_line_id": order_line_id, **kwargs} ), ) with pytest.raises(RuntimeError, match="still boom"): render_order_line_still_task.run.__func__( task_self, "line-still-failure", workflow_run_id="run-still-failure", workflow_node_id="render", graph_authoritative_output_enabled=True, graph_output_node_ids=["output"], graph_notify_node_ids=["notify"], emit_legacy_notifications=True, job_document_enabled=False, emit_events=False, ) assert notify_emit_calls == [ { "order_line_id": "line-still-failure", "success": False, "render_log": {"error": "still boom"}, } ] assert notify_finalize_calls == [ { "workflow_run_id": "run-still-failure", "notify_node_ids": ["notify"], "success": False, "render_node_id": "render", "error": "still boom", } ] def test_export_blend_for_order_line_task_finalizes_notify_handoff_on_failure( tmp_path, monkeypatch, ): from app.config import settings from app.domains.rendering.tasks import export_blend_for_order_line_task step_path = tmp_path / "cad" / "bearing.step" step_path.parent.mkdir(parents=True, exist_ok=True) step_path.write_text("STEP", encoding="utf-8") glb_path = step_path.parent / "bearing_render.glb" glb_path.write_text("GLB", encoding="utf-8") task_self = SimpleNamespace( request=SimpleNamespace(id="task-blend-failure"), retry=lambda *, exc, countdown: (_ for _ in ()).throw(exc), ) notify_finalize_calls: list[dict] = [] notify_emit_calls: list[dict] = [] monkeypatch.setattr(settings, "upload_dir", str(tmp_path / "uploads")) monkeypatch.setattr( "app.domains.rendering.tasks._resolve_step_path_for_order_line", lambda order_line_id: (str(step_path), "cad-1"), ) monkeypatch.setattr( "app.services.render_blender.find_blender", lambda: "/usr/bin/blender", ) monkeypatch.setattr( "app.services.render_blender.build_tessellated_glb_path", lambda *args, **kwargs: glb_path, ) monkeypatch.setattr( "subprocess.run", lambda *args, **kwargs: SimpleNamespace(returncode=1, stderr="blend boom"), ) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_graph_blend_output", lambda *args, **kwargs: None, ) monkeypatch.setattr( "app.domains.rendering.tasks._update_workflow_run_status", lambda *args, **kwargs: None, ) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_graph_notify_nodes", lambda **kwargs: notify_finalize_calls.append(kwargs), ) monkeypatch.setattr( "app.domains.rendering.tasks._emit_graph_render_notifications", lambda order_line_id, **kwargs: notify_emit_calls.append( {"order_line_id": order_line_id, **kwargs} ), ) with pytest.raises(RuntimeError, match="export_blend.py exited 1"): export_blend_for_order_line_task.run.__func__( task_self, "line-blend-failure", workflow_run_id="run-blend-failure", workflow_node_id="blend", graph_authoritative_output_enabled=True, graph_output_node_ids=["output"], graph_notify_node_ids=["notify"], emit_legacy_notifications=True, ) assert notify_emit_calls == [ { "order_line_id": "line-blend-failure", "success": False, "render_log": {"error": "export_blend.py exited 1:\nblend boom"}, } ] assert notify_finalize_calls == [ { "workflow_run_id": "run-blend-failure", "notify_node_ids": ["notify"], "success": False, "render_node_id": "blend", "error": "export_blend.py exited 1:\nblend boom", } ] def test_render_turntable_task_finalizes_notify_handoff_on_failure( tmp_path, monkeypatch, ): from app.domains.rendering.tasks import render_turntable_task step_path = tmp_path / "cad" / "bearing.step" step_path.parent.mkdir(parents=True, exist_ok=True) step_path.write_text("STEP", encoding="utf-8") task_self = SimpleNamespace( request=SimpleNamespace(id="task-turntable-failure"), retry=lambda *, exc, countdown: (_ for _ in ()).throw(exc), ) notify_finalize_calls: list[dict] = [] notify_emit_calls: list[dict] = [] def _fake_subprocess_run(cmd, *args, **kwargs): cmd_text = " ".join(str(part) for part in cmd) if "export_step_to_gltf.py" in cmd_text: glb_path = step_path.parent / "bearing_thumbnail.glb" glb_path.write_text("GLB", encoding="utf-8") return SimpleNamespace(returncode=0, stdout="", stderr="") if "turntable_render.py" in cmd_text: return SimpleNamespace(returncode=1, stdout="turntable boom", stderr="") return SimpleNamespace(returncode=0, stdout="", stderr="") monkeypatch.setattr( "app.domains.rendering.tasks._resolve_step_path_for_order_line", lambda order_line_id: (str(step_path), "cad-1"), ) monkeypatch.setattr( "app.services.render_blender.find_blender", lambda: "/usr/bin/blender", ) monkeypatch.setattr( "subprocess.run", _fake_subprocess_run, ) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_graph_turntable_output", lambda *args, **kwargs: None, ) monkeypatch.setattr( "app.domains.rendering.tasks._update_workflow_run_status", lambda *args, **kwargs: None, ) monkeypatch.setattr( "app.domains.rendering.tasks._finalize_graph_notify_nodes", lambda **kwargs: notify_finalize_calls.append(kwargs), ) monkeypatch.setattr( "app.domains.rendering.tasks._emit_graph_render_notifications", lambda order_line_id, **kwargs: notify_emit_calls.append( {"order_line_id": order_line_id, **kwargs} ), ) with pytest.raises(RuntimeError, match="Blender turntable exited 1"): render_turntable_task.run.__func__( task_self, "line-turntable-failure", workflow_run_id="run-turntable-failure", workflow_node_id="turntable", graph_authoritative_output_enabled=True, graph_output_node_ids=["output"], graph_notify_node_ids=["notify"], emit_legacy_notifications=True, emit_events=False, ) assert notify_emit_calls == [ { "order_line_id": "line-turntable-failure", "success": False, "render_log": {"error": "Blender turntable exited 1:\nturntable boom"}, } ] assert notify_finalize_calls == [ { "workflow_run_id": "run-turntable-failure", "notify_node_ids": ["notify"], "success": False, "render_node_id": "turntable", "error": "Blender turntable exited 1:\nturntable boom", } ]