import pytest from app.core.process_steps import StepName from app.domains.rendering.models import OutputType, WorkflowDefinition, WorkflowRun from app.domains.rendering.workflow_config_utils import build_preset_workflow_config from app.domains.rendering.workflow_graph_runtime import _STILL_TASK_KEYS, _TURNTABLE_TASK_KEYS 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", "shared"} for definition in definitions) assert all(definition.module_key for definition in definitions) assert all(definition.legacy_source for definition in definitions) def test_node_registry_module_keys_are_unique(): definitions = list_node_definitions() module_keys = [definition.module_key for definition in definitions] assert len(module_keys) == len(set(module_keys)) def test_node_registry_defaults_match_declared_fields(): definitions = list_node_definitions() for definition in definitions: field_keys = {field.key for field in definition.fields} default_keys = set(definition.defaults) assert default_keys <= field_keys def test_node_registry_contracts_have_valid_shape(): definitions = list_node_definitions() for definition in definitions: input_context = definition.input_contract.get("context") output_context = definition.output_contract.get("context") if definition.family == "shared": assert input_context is None assert output_context is None else: assert input_context == definition.family assert output_context == definition.family required = definition.input_contract.get("requires", []) required_any = definition.input_contract.get("requires_any", []) provides = definition.output_contract.get("provides", []) assert len(required) == len(set(required)) assert len(required_any) == len(set(required_any)) assert len(provides) == len(set(provides)) assert len(definition.artifact_roles_consumed) == len(set(definition.artifact_roles_consumed)) assert len(definition.artifact_roles_produced) == len(set(definition.artifact_roles_produced)) field_keys = [field.key for field in definition.fields] assert len(field_keys) == len(set(field_keys)) 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["frame_count"] == 120 assert definition.defaults["duration_s"] == 5 assert definition.defaults["turntable_degrees"] == 360 assert definition.defaults["turntable_axis"] == "world_z" assert definition.defaults["camera_orbit"] is True 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_graph_render_node_fields_are_supported_by_runtime_dispatch(): still_definition = get_node_definition(StepName.BLENDER_STILL) turntable_definition = get_node_definition(StepName.BLENDER_TURNTABLE) assert still_definition is not None assert turntable_definition is not None still_runtime_fields = {field.key for field in still_definition.fields if field.key != "use_custom_render_settings"} turntable_runtime_fields = { field.key for field in turntable_definition.fields if field.key != "use_custom_render_settings" } assert still_runtime_fields <= _STILL_TASK_KEYS assert turntable_runtime_fields <= _TURNTABLE_TASK_KEYS 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", "source_preference"} assert bbox.family == "shared" assert bbox.input_contract == {"requires": ["glb_preview"]} assert bbox.output_contract == {"provides": ["bbox"]} assert {field.key for field in template.fields} == { "template_id_override", "require_template", "material_library_path", "disable_materials", "target_collection", "material_replace_mode", "lighting_only_mode", "shadow_catcher_mode", "camera_orbit_mode", } assert {field.key for field in get_node_definition(StepName.MATERIAL_MAP_RESOLVE).fields} == { "disable_materials", "material_override", } assert {field.key for field in get_node_definition(StepName.AUTO_POPULATE_MATERIALS).fields} == { "persist_updates", "refresh_material_source", "include_populated_products", } 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 {field.key for field in output.fields} == { "expected_artifact_role", "require_upstream_artifact", } assert export_blend.defaults["output_name_suffix"] == "" assert {field.key for field in export_blend.fields} == {"output_name_suffix"} assert notify.defaults == {"channel": "audit_log", "require_armed_render": False} assert notify.input_contract["requires"] == ["order_line_context"] assert notify.input_contract["requires_any"] == [ "rendered_image", "rendered_frames", "rendered_video", "workflow_result", ] assert {field.key for field in notify.fields} == {"channel", "require_armed_render"} def test_cad_and_export_contract_nodes_only_expose_supported_settings(): occ_glb_export = get_node_definition(StepName.OCC_GLB_EXPORT) thumbnail_save = get_node_definition(StepName.THUMBNAIL_SAVE) export_blend = get_node_definition(StepName.EXPORT_BLEND) stl_cache_generate = get_node_definition(StepName.STL_CACHE_GENERATE) assert occ_glb_export is not None assert thumbnail_save is not None assert export_blend is not None assert stl_cache_generate is not None assert occ_glb_export.family == "cad_file" assert occ_glb_export.fields == [] assert occ_glb_export.defaults == {} assert occ_glb_export.input_contract == {"context": "cad_file", "requires": ["step_path"]} assert occ_glb_export.output_contract == {"context": "cad_file", "provides": ["glb_preview"]} assert occ_glb_export.artifact_roles_consumed == ["step_path"] assert occ_glb_export.artifact_roles_produced == ["glb_preview"] assert "does not expose per-node overrides yet" in occ_glb_export.description assert thumbnail_save.family == "cad_file" assert thumbnail_save.fields == [] assert thumbnail_save.defaults == {} assert thumbnail_save.input_contract == {"context": "cad_file", "requires": ["rendered_image"]} assert thumbnail_save.output_contract == {"context": "cad_file", "provides": ["cad_thumbnail_media"]} assert thumbnail_save.artifact_roles_consumed == ["rendered_image"] assert thumbnail_save.artifact_roles_produced == ["cad_thumbnail_media"] assert "connected upstream thumbnail request node" in thumbnail_save.description assert export_blend.family == "order_line" assert export_blend.defaults == {"output_name_suffix": ""} assert {field.key for field in export_blend.fields} == {"output_name_suffix"} assert export_blend.input_contract == { "context": "order_line", "requires": ["order_line_context", "render_template"], } assert export_blend.output_contract == {"context": "order_line", "provides": ["blend_asset"]} assert export_blend.artifact_roles_consumed == ["order_line_context", "render_template"] assert export_blend.artifact_roles_produced == ["blend_asset"] assert "Only the optional filename suffix is workflow-configurable today." in export_blend.description assert stl_cache_generate.family == "cad_file" assert stl_cache_generate.fields == [] assert stl_cache_generate.defaults == {} assert stl_cache_generate.input_contract == {"context": "cad_file", "requires": ["step_path"]} assert stl_cache_generate.output_contract == {"context": "cad_file", "provides": ["stl_cache"]} assert stl_cache_generate.artifact_roles_consumed == ["step_path"] assert stl_cache_generate.artifact_roles_produced == ["stl_cache"] assert "Compatibility node for legacy CAD flows." in stl_cache_generate.description @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", } blender_turntable = next( definition for definition in body["definitions"] if definition["step"] == StepName.BLENDER_TURNTABLE.value ) assert blender_turntable["defaults"]["fps"] == 24 assert blender_turntable["defaults"]["frame_count"] == 120 assert blender_turntable["defaults"]["duration_s"] == 5 assert blender_turntable["defaults"]["turntable_degrees"] == 360 assert blender_turntable["defaults"]["turntable_axis"] == "world_z" assert blender_turntable["defaults"]["camera_orbit"] is True 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": [], "allow_blank": True, "max_length": None, "text_format": "absolute_glb_path", }, { "key": "source_preference", "label": "Source Preference", "type": "select", "description": "Prefer a prepared GLB, force STEP fallback, or fail when no GLB artifact is available.", "section": "Inputs", "default": "auto", "min": None, "max": None, "step": None, "unit": None, "options": [ {"value": "auto", "label": "Auto"}, {"value": "step_only", "label": "STEP Only"}, {"value": "glb_only", "label": "GLB Only"}, ], "allow_blank": True, "max_length": None, "text_format": "plain", }, ] @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_workflow_crud_exposes_supported_artifact_kinds(client, auth_headers): create_response = await client.post( "/api/workflows", headers=auth_headers, json={ "name": "Still Workflow Contract", "config": build_preset_workflow_config("still_graph"), "is_active": True, }, ) assert create_response.status_code == 201, create_response.text created = create_response.json() assert created["family"] == "order_line" assert created["supported_artifact_kinds"] == ["still_image"] 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["supported_artifact_kinds"] == ["still_image"] @pytest.mark.asyncio async def test_workflow_crud_exposes_rollout_summary(client, db, auth_headers): workflow = WorkflowDefinition( name="Shadow Rollout Workflow", config=build_preset_workflow_config("still_graph") | { "ui": { **(build_preset_workflow_config("still_graph").get("ui") or {}), "execution_mode": "shadow", } }, is_active=True, ) db.add(workflow) await db.flush() output_type = OutputType( name="Shadow Still Output", workflow_definition_id=workflow.id, workflow_family="order_line", artifact_kind="still_image", workflow_rollout_mode="shadow", render_backend="celery", ) db.add(output_type) await db.flush() workflow_run = WorkflowRun( workflow_def_id=workflow.id, execution_mode="shadow", status="completed", ) db.add(workflow_run) await db.commit() response = await client.get(f"/api/workflows/{workflow.id}", headers=auth_headers) assert response.status_code == 200, response.text body = response.json() assert body["rollout_summary"]["linked_output_type_count"] == 1 assert body["rollout_summary"]["linked_output_type_names"] == ["Shadow Still Output"] assert body["rollout_summary"]["linked_output_types"] == [ { "id": str(output_type.id), "name": "Shadow Still Output", "is_active": True, "artifact_kind": "still_image", "workflow_rollout_mode": "shadow", } ] assert body["rollout_summary"]["rollout_modes"] == ["shadow"] assert body["rollout_summary"]["has_blocking_contracts"] is False assert body["rollout_summary"]["latest_shadow_run"]["workflow_run_id"] == str(workflow_run.id) assert body["rollout_summary"]["latest_shadow_run"]["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"