Close workflow smoke template drift

This commit is contained in:
2026-04-18 13:16:54 +02:00
parent 9f1ccec8f0
commit 98455ee579
3 changed files with 507 additions and 0 deletions
+121
View File
@@ -494,6 +494,96 @@ def choose_template_backed_output_type(
raise RuntimeError("No active template-backed still-image output type was found")
WORKFLOW_GOLDEN_TEMPLATE_BY_KEY = {
"still_legacy": "BlenderStudio",
"still_graph": "BlenderStudio",
"still_shadow": "BlenderStudio",
"turntable_graph": "Blender_Studio_Schadowcatcher_Anim",
"blend_graph": "BlenderStudio",
}
def workflow_golden_template_name(case: dict) -> str | None:
return WORKFLOW_GOLDEN_TEMPLATE_BY_KEY.get(str(case.get("key") or "").strip())
def _template_output_type_ids(template: dict) -> list[str]:
linked_output_type_ids = [
str(candidate_id)
for candidate_id in (template.get("output_type_ids") or [])
if candidate_id is not None
]
fallback_output_type_id = template.get("output_type_id")
if fallback_output_type_id and str(fallback_output_type_id) not in linked_output_type_ids:
linked_output_type_ids.append(str(fallback_output_type_id))
return linked_output_type_ids
def ensure_output_type_render_template_binding(
client: APIClient,
*,
output_type: dict,
preferred_template_name: str | None,
) -> list[dict]:
if not preferred_template_name:
return []
render_templates = get_render_templates(client)
preferred_template = find_named(render_templates, preferred_template_name)
if preferred_template is None:
raise RuntimeError(
f"Required render template '{preferred_template_name}' is missing in /admin"
)
if not preferred_template.get("is_active", True):
raise RuntimeError(
f"Required render template '{preferred_template_name}' is inactive in /admin"
)
output_type_id = str(output_type["id"])
changed = False
for template in render_templates:
current_ids = _template_output_type_ids(template)
desired_ids = [candidate_id for candidate_id in current_ids if candidate_id != output_type_id]
if template["id"] == preferred_template["id"] and output_type_id not in desired_ids:
desired_ids.append(output_type_id)
if desired_ids == current_ids:
continue
resp = client.patch(
f"/render-templates/{template['id']}",
json={"output_type_ids": desired_ids},
)
if resp.status_code != 200:
raise RuntimeError(
"Golden render-template link update failed "
f"({template.get('name')}{output_type.get('name')}): "
f"{resp.status_code} {resp.text[:400]}"
)
changed = True
if changed:
info(
"Ensured golden template binding: "
f"{output_type.get('name')} -> {preferred_template_name}"
)
refreshed_templates = get_render_templates(client)
matches = render_template_candidates_for_output_type(
refreshed_templates,
output_type_id,
active_only=False,
)
if not any(template.get("name") == preferred_template_name for template in matches):
raise RuntimeError(
"Golden render-template binding verification failed for "
f"{output_type.get('name')} -> {preferred_template_name}"
)
return matches
def build_output_type_workflow_snapshot(output_type: dict) -> dict:
return {
"workflow_definition_id": output_type.get("workflow_definition_id"),
@@ -510,6 +600,13 @@ def smoke_workflow_name(execution_mode: str) -> str:
return f"[Workflow Smoke] Canonical Still {execution_mode.title()}"
def workflow_smoke_template_name(execution_mode: str) -> str | None:
normalized_mode = str(execution_mode or "").strip().lower()
if normalized_mode in {"legacy", "graph", "shadow"}:
return "BlenderStudio"
return None
def build_workflow_golden_cases() -> list[dict]:
still_invocation = {
"width": 1024,
@@ -688,6 +785,18 @@ def ensure_workflow_still_smoke_resources(
output_type = resp.json()
info(f"Reusing smoke output type: {output_type_name}")
preferred_template_name = workflow_smoke_template_name(execution_mode)
if preferred_template_name:
bound_templates = ensure_output_type_render_template_binding(
client,
output_type=output_type,
preferred_template_name=preferred_template_name,
)
info(
"Smoke template candidates: "
+ ", ".join(template.get("name") or "<unnamed>" for template in bound_templates)
)
workflow = None
if execution_mode != "legacy":
workflows = get_workflows(client)
@@ -790,6 +899,18 @@ def ensure_workflow_golden_resources(
output_type = resp.json()
info(f"Reusing golden output type: {case['output_type_name']}")
preferred_template_name = workflow_golden_template_name(case)
if preferred_template_name:
bound_templates = ensure_output_type_render_template_binding(
client,
output_type=output_type,
preferred_template_name=preferred_template_name,
)
info(
"Golden template candidates: "
+ ", ".join(template.get("name") or "<unnamed>" for template in bound_templates)
)
workflow = None
workflow_name = case.get("workflow_name")
workflow_config = case.get("workflow_config")