import pytest from app.core.process_steps import StepName from app.domains.rendering.models import WorkflowDefinition from app.domains.rendering.workflow_node_registry import ( get_node_definition, list_node_definitions, ) def test_node_registry_covers_all_step_names(): definitions = list_node_definitions() registered_steps = {definition.step for definition in definitions} expected_steps = {step.value for step in StepName} assert registered_steps == expected_steps assert all(definition.family in {"cad_file", "order_line"} for definition in definitions) assert all(definition.module_key for definition in definitions) assert all(definition.legacy_source for definition in definitions) def test_turntable_node_definition_exposes_expected_schema(): definition = get_node_definition(StepName.BLENDER_TURNTABLE) assert definition is not None assert definition.family == "order_line" assert definition.module_key == "render.production.turntable" assert definition.node_type == "renderFramesNode" assert definition.defaults["fps"] == 24 assert definition.defaults["duration_s"] == 5 assert definition.input_contract["context"] == "order_line" assert definition.output_contract["provides"] == ["rendered_frames", "rendered_video"] assert "material_assignments" in definition.artifact_roles_consumed assert "rendered_video" in definition.artifact_roles_produced assert definition.legacy_source == f"legacy_step:{StepName.BLENDER_TURNTABLE.value}" assert {field.key for field in definition.fields} >= { "render_engine", "cycles_device", "samples", "transparent_bg", "bg_color", "width", "height", "fps", "duration_s", "turntable_degrees", "turntable_axis", "camera_orbit", "rotation_x", "rotation_y", "rotation_z", "focal_length_mm", "sensor_width_mm", "material_override", } def test_order_line_setup_and_template_contracts_expose_runtime_outputs(): setup = get_node_definition(StepName.ORDER_LINE_SETUP) template = get_node_definition(StepName.RESOLVE_TEMPLATE) bbox = get_node_definition(StepName.GLB_BBOX) output = get_node_definition(StepName.OUTPUT_SAVE) export_blend = get_node_definition(StepName.EXPORT_BLEND) notify = get_node_definition(StepName.NOTIFY) assert setup is not None assert template is not None assert bbox is not None assert output is not None assert export_blend is not None assert notify is not None assert set(setup.output_contract["provides"]) >= { "order_line_context", "cad_file_ref", "step_path", "cad_materials", "glb_preview", "bbox", } assert set(template.output_contract["provides"]) >= { "render_template", "output_profile", "template_path", "material_library", "material_map", "use_materials", "override_material", } assert {field.key for field in bbox.fields} == {"glb_path"} assert output.input_contract["requires"] == ["order_line_context"] assert output.input_contract["requires_any"] == ["rendered_image", "rendered_frames", "rendered_video"] assert set(output.output_contract["provides"]) >= {"media_asset", "workflow_result"} assert export_blend.defaults["output_name_suffix"] == "" assert {field.key for field in export_blend.fields} == {"output_name_suffix"} assert notify.input_contract["requires"] == ["order_line_context"] assert notify.input_contract["requires_any"] == [ "rendered_image", "rendered_frames", "rendered_video", "workflow_result", ] @pytest.mark.asyncio async def test_node_definitions_endpoint_returns_registry(client, auth_headers): response = await client.get("/api/workflows/node-definitions", headers=auth_headers) assert response.status_code == 200 body = response.json() assert len(body["definitions"]) == len(StepName) blender_still = next( definition for definition in body["definitions"] if definition["step"] == StepName.BLENDER_STILL.value ) assert blender_still["family"] == "order_line" assert blender_still["module_key"] == "render.production.still" assert blender_still["node_type"] == "renderNode" assert blender_still["defaults"]["use_custom_render_settings"] is False assert blender_still["input_contract"]["context"] == "order_line" assert "bbox" in blender_still["artifact_roles_consumed"] assert blender_still["output_contract"]["provides"] == ["rendered_image"] assert blender_still["legacy_source"] == f"legacy_step:{StepName.BLENDER_STILL.value}" assert {field["key"] for field in blender_still["fields"]} >= { "use_custom_render_settings", "render_engine", "cycles_device", "samples", "width", "height", "transparent_bg", "noise_threshold", "denoiser", "denoising_input_passes", "denoising_prefilter", "denoising_quality", "denoising_use_gpu", "target_collection", "lighting_only", "shadow_catcher", "rotation_x", "rotation_y", "rotation_z", "focal_length_mm", "sensor_width_mm", "material_override", } glb_bbox = next( definition for definition in body["definitions"] if definition["step"] == StepName.GLB_BBOX.value ) assert glb_bbox["fields"] == [ { "key": "glb_path", "label": "GLB Path Override", "type": "text", "description": "Optional absolute path to a specific GLB file. Leave empty to reuse the prepared preview/export artifact automatically.", "section": "Inputs", "default": "", "min": None, "max": None, "step": None, "unit": None, "options": [], } ] @pytest.mark.asyncio async def test_workflow_crud_roundtrip_preserves_execution_mode(client, auth_headers): create_response = await client.post( "/api/workflows", headers=auth_headers, json={ "name": "Shadow Workflow", "config": { "version": 1, "ui": { "preset": "custom", "execution_mode": "shadow", }, "nodes": [ { "id": "setup", "step": StepName.ORDER_LINE_SETUP.value, "params": {}, } ], "edges": [], }, "is_active": True, }, ) assert create_response.status_code == 201 created = create_response.json() assert created["config"]["ui"]["execution_mode"] == "shadow" get_response = await client.get(f"/api/workflows/{created['id']}", headers=auth_headers) assert get_response.status_code == 200 fetched = get_response.json() assert fetched["config"]["ui"]["execution_mode"] == "shadow" @pytest.mark.asyncio async def test_admin_backfill_workflows_rewrites_legacy_configs(client, db, auth_headers): legacy = WorkflowDefinition( name="Legacy Still Workflow", config={ "type": "still", "params": {"width": 1280, "height": 720}, "ui": {"execution_mode": "graph"}, }, is_active=True, ) canonical = WorkflowDefinition( name="Canonical Workflow", config={ "version": 1, "ui": {"preset": "custom", "execution_mode": "legacy"}, "nodes": [ { "id": "setup", "step": StepName.ORDER_LINE_SETUP.value, "params": {}, } ], "edges": [], }, is_active=True, ) db.add_all([legacy, canonical]) await db.commit() response = await client.post("/api/admin/settings/backfill-workflows", headers=auth_headers) assert response.status_code == 200 body = response.json() assert body["scanned"] == 2 assert body["updated"] == 1 assert body["invalid"] == [] assert body["workflows"] == [{"id": str(legacy.id), "name": "Legacy Still Workflow"}] await db.refresh(legacy) await db.refresh(canonical) assert legacy.config["version"] == 1 assert legacy.config["ui"]["preset"] == "still" assert legacy.config["ui"]["execution_mode"] == "graph" assert any(node["step"] == StepName.BLENDER_STILL.value for node in legacy.config["nodes"]) assert canonical.config["ui"]["preset"] == "custom"