from __future__ import annotations import importlib.util from pathlib import Path import sys import types def _load_render_pipeline_script(): candidates = [ Path(__file__).resolve().parents[3] / "scripts" / "test_render_pipeline.py", Path("/compose/scripts/test_render_pipeline.py"), ] script_path = next((candidate for candidate in candidates if candidate.exists()), None) assert script_path is not None if "requests" not in sys.modules: requests_stub = types.ModuleType("requests") requests_stub.Response = object requests_stub.Session = object requests_stub.exceptions = types.SimpleNamespace( ConnectionError=RuntimeError, ChunkedEncodingError=RuntimeError, ReadTimeout=RuntimeError, ) sys.modules["requests"] = requests_stub spec = importlib.util.spec_from_file_location("test_render_pipeline_script", script_path) assert spec is not None assert spec.loader is not None module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def test_build_output_type_workflow_link_payload_sets_graph_rollout_mode_explicitly(): module = _load_render_pipeline_script() payload = module.build_output_type_workflow_link_payload( workflow_definition_id="workflow-graph-123", execution_mode="graph", ) assert payload == { "workflow_definition_id": "workflow-graph-123", "workflow_rollout_mode": "graph", "is_active": True, } def test_build_output_type_workflow_link_payload_sets_shadow_rollout_mode_explicitly(): module = _load_render_pipeline_script() payload = module.build_output_type_workflow_link_payload( workflow_definition_id="workflow-shadow-123", execution_mode="shadow", ) assert payload == { "workflow_definition_id": "workflow-shadow-123", "workflow_rollout_mode": "shadow", "is_active": True, } def test_build_output_type_workflow_link_payload_keeps_legacy_rollout_implicit(): module = _load_render_pipeline_script() payload = module.build_output_type_workflow_link_payload( workflow_definition_id="workflow-legacy-123", execution_mode="legacy", ) assert payload == { "workflow_definition_id": "workflow-legacy-123", "is_active": True, } def test_build_graph_still_config_matches_canonical_still_graph_contract(): module = _load_render_pipeline_script() config = module.build_graph_still_config( execution_mode="shadow", render_params={ "resolution": [1920, 1080], "engine": "cycles", "samples": 128, }, ) assert config["ui"] == { "preset": "still_graph", "execution_mode": "shadow", "family": "order_line", } assert [node["id"] for node in config["nodes"]] == [ "setup", "template", "populate_materials", "bbox", "resolve_materials", "render", "output", "notify", ] assert config["edges"] == [ {"from": "setup", "to": "template"}, {"from": "setup", "to": "populate_materials"}, {"from": "setup", "to": "bbox"}, {"from": "template", "to": "resolve_materials"}, {"from": "populate_materials", "to": "resolve_materials"}, {"from": "resolve_materials", "to": "render"}, {"from": "bbox", "to": "render"}, {"from": "template", "to": "render"}, {"from": "render", "to": "output"}, {"from": "render", "to": "notify"}, ] render_node = next(node for node in config["nodes"] if node["id"] == "render") assert render_node["params"] == { "width": 1920, "height": 1080, "render_engine": "cycles", "samples": 128, "use_custom_render_settings": False, } def test_render_template_candidates_for_output_type_matches_m2m_and_legacy_fields(): module = _load_render_pipeline_script() templates = [ { "id": "template-active-m2m", "is_active": True, "output_type_ids": ["ot-1", "ot-2"], "output_type_id": None, }, { "id": "template-active-legacy", "is_active": True, "output_type_ids": [], "output_type_id": "ot-1", }, { "id": "template-inactive", "is_active": False, "output_type_ids": ["ot-1"], "output_type_id": None, }, ] matches = module.render_template_candidates_for_output_type(templates, "ot-1") assert [template["id"] for template in matches] == [ "template-active-m2m", "template-active-legacy", ] def test_build_graph_still_config_can_inherit_output_type_render_settings(): module = _load_render_pipeline_script() config = module.build_graph_still_config( execution_mode="shadow", use_custom_render_settings=False, ) render_node = next(node for node in config["nodes"] if node["id"] == "render") assert render_node["params"] == { "use_custom_render_settings": False, } def test_choose_template_backed_output_type_prefers_requested_name(): module = _load_render_pipeline_script() output_types = [ { "id": "ot-1", "name": "HQ-Blender-Alpha-HDR", "renderer": "blender", "artifact_kind": "still_image", "is_animation": False, }, { "id": "ot-2", "name": "Turntable", "renderer": "blender", "artifact_kind": "turntable_video", "is_animation": True, }, ] templates = [ { "id": "template-1", "is_active": True, "output_type_ids": ["ot-1"], "output_type_id": None, } ] output_type, matches = module.choose_template_backed_output_type( output_types, templates, preferred_name="HQ-Blender-Alpha-HDR", ) assert output_type["id"] == "ot-1" assert [template["id"] for template in matches] == ["template-1"] def test_workflow_golden_template_name_maps_cases_to_expected_templates(): module = _load_render_pipeline_script() assert module.workflow_golden_template_name({"key": "still_graph"}) == "BlenderStudio" assert module.workflow_golden_template_name({"key": "still_shadow"}) == "BlenderStudio" assert module.workflow_golden_template_name({"key": "turntable_graph"}) == "Blender_Studio_Schadowcatcher_Anim" assert module.workflow_golden_template_name({"key": "unknown"}) is None def test_workflow_smoke_template_name_maps_supported_modes_to_blenderstudio(): module = _load_render_pipeline_script() assert module.workflow_smoke_template_name("legacy") == "BlenderStudio" assert module.workflow_smoke_template_name("graph") == "BlenderStudio" assert module.workflow_smoke_template_name("shadow") == "BlenderStudio" assert module.workflow_smoke_template_name("turntable") is None def test_smoke_turntable_output_type_name_includes_variant_and_mode(): module = _load_render_pipeline_script() name = module.smoke_turntable_output_type_name("graph") assert "Turntable" in name assert "Graph" in name assert name.startswith("[Workflow Smoke]") def test_workflow_smoke_turntable_template_name_returns_animation_template_for_graph(): module = _load_render_pipeline_script() assert module.workflow_smoke_turntable_template_name("graph") == "Blender_Studio_Schadowcatcher_Anim" assert module.workflow_smoke_turntable_template_name("shadow") == "Blender_Studio_Schadowcatcher_Anim" assert module.workflow_smoke_turntable_template_name("legacy") is None def test_workflow_smoke_blend_template_name_returns_blenderstudio(): module = _load_render_pipeline_script() assert module.workflow_smoke_blend_template_name() == "BlenderStudio" def test_build_graph_turntable_config_accepts_render_params_override(): module = _load_render_pipeline_script() config = module.build_graph_turntable_config( execution_mode="graph", render_params={"width": 512, "height": 512, "fps": 12, "duration_s": 2, "samples": 16}, ) turntable_node = next(node for node in config["nodes"] if node["id"] == "turntable") assert turntable_node["params"]["width"] == 512 assert turntable_node["params"]["height"] == 512 assert turntable_node["params"]["fps"] == 12 assert turntable_node["params"]["duration_s"] == 2 assert turntable_node["params"]["samples"] == 16 def test_build_graph_turntable_config_defaults_unchanged_without_render_params(): module = _load_render_pipeline_script() config = module.build_graph_turntable_config(execution_mode="graph") turntable_node = next(node for node in config["nodes"] if node["id"] == "turntable") assert turntable_node["params"]["width"] == 768 assert turntable_node["params"]["height"] == 768 def test_smoke_blend_output_type_name_is_stable(): module = _load_render_pipeline_script() assert module.smoke_blend_output_type_name() == "[Workflow Smoke] Blend Export" assert module.smoke_blend_workflow_name() == "[Workflow Smoke] Blend Export" def test_ensure_output_type_render_template_binding_moves_output_type_to_preferred_template(): module = _load_render_pipeline_script() templates_state = [ { "id": "template-preferred", "name": "BlenderStudio", "is_active": True, "output_type_ids": ["ot-existing"], "output_type_id": "ot-existing", }, { "id": "template-other", "name": "Other Template", "is_active": True, "output_type_ids": ["ot-golden", "ot-other"], "output_type_id": "ot-golden", }, ] patch_calls: list[tuple[str, dict]] = [] class _Response: def __init__(self, payload: dict): self.status_code = 200 self._payload = payload self.text = "" def json(self): return self._payload class _Client: def patch(self, path: str, **kwargs): payload = kwargs["json"] patch_calls.append((path, payload)) template_id = path.rsplit("/", 1)[-1] template = next(item for item in templates_state if item["id"] == template_id) template["output_type_ids"] = list(payload["output_type_ids"]) template["output_type_id"] = payload["output_type_ids"][0] if payload["output_type_ids"] else None return _Response(template) client = _Client() original_get_render_templates = module.get_render_templates try: module.get_render_templates = lambda _client: [ { "id": item["id"], "name": item["name"], "is_active": item["is_active"], "output_type_ids": list(item["output_type_ids"]), "output_type_id": item["output_type_id"], } for item in templates_state ] matches = module.ensure_output_type_render_template_binding( client, output_type={"id": "ot-golden", "name": "[Workflow Golden] Canonical Still Graph"}, preferred_template_name="BlenderStudio", ) finally: module.get_render_templates = original_get_render_templates assert patch_calls == [ ("/render-templates/template-preferred", {"output_type_ids": ["ot-existing", "ot-golden"]}), ("/render-templates/template-other", {"output_type_ids": ["ot-other"]}), ] assert [template["name"] for template in matches] == ["BlenderStudio"] def test_build_output_type_workflow_snapshot_keeps_restore_contract(): module = _load_render_pipeline_script() snapshot = module.build_output_type_workflow_snapshot( { "workflow_definition_id": "workflow-123", "workflow_rollout_mode": "shadow", "is_active": False, } ) assert snapshot == { "workflow_definition_id": "workflow-123", "workflow_rollout_mode": "shadow", "is_active": False, }