Spaces:
Running on Zero
Running on Zero
feat(workflow): load_template returns fresh deep copy per mode
Browse files- tests/test_workflow.py +24 -0
- workflow.py +19 -0
tests/test_workflow.py
ADDED
|
@@ -0,0 +1,24 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
"""Unit tests for workflow.py — pure functions over JSON dicts."""
|
| 2 |
+
import pytest
|
| 3 |
+
|
| 4 |
+
import workflow
|
| 5 |
+
|
| 6 |
+
|
| 7 |
+
def test_load_template_returns_dict_for_valid_mode():
|
| 8 |
+
wf = workflow.load_template("t2v")
|
| 9 |
+
assert isinstance(wf, dict)
|
| 10 |
+
assert "nodes" in wf
|
| 11 |
+
assert len(wf["nodes"]) > 0
|
| 12 |
+
|
| 13 |
+
|
| 14 |
+
def test_load_template_raises_for_unknown_mode():
|
| 15 |
+
with pytest.raises(ValueError, match="unknown mode"):
|
| 16 |
+
workflow.load_template("nonexistent")
|
| 17 |
+
|
| 18 |
+
|
| 19 |
+
def test_load_template_returns_independent_copy():
|
| 20 |
+
"""Mutations to one returned dict must not affect later loads."""
|
| 21 |
+
a = workflow.load_template("t2v")
|
| 22 |
+
a["nodes"].append({"id": -999})
|
| 23 |
+
b = workflow.load_template("t2v")
|
| 24 |
+
assert {-999} & {n.get("id") for n in b["nodes"]} == set()
|
workflow.py
ADDED
|
@@ -0,0 +1,19 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
"""Pure functions over LTX 2.3 mode workflow JSON templates."""
|
| 2 |
+
from __future__ import annotations
|
| 3 |
+
|
| 4 |
+
import copy
|
| 5 |
+
import json
|
| 6 |
+
import pathlib
|
| 7 |
+
from typing import Any
|
| 8 |
+
|
| 9 |
+
WORKFLOWS_DIR = pathlib.Path(__file__).parent / "workflows"
|
| 10 |
+
|
| 11 |
+
VALID_MODES: tuple[str, ...] = ("t2v", "a2v", "i2v", "lipsync", "keyframe", "style")
|
| 12 |
+
|
| 13 |
+
|
| 14 |
+
def load_template(mode: str) -> dict[str, Any]:
|
| 15 |
+
"""Load a fresh, independent copy of the named mode's workflow template."""
|
| 16 |
+
if mode not in VALID_MODES:
|
| 17 |
+
raise ValueError(f"unknown mode {mode!r}; expected one of {VALID_MODES}")
|
| 18 |
+
path = WORKFLOWS_DIR / f"{mode}.json"
|
| 19 |
+
return copy.deepcopy(json.loads(path.read_text()))
|