Van/攻城狮鹿鸣君
e9394a1f2a
Two new providers on Alibaba Cloud DashScope platform, sharing the same
API endpoints but with different models and parameters.
Wan Bailian CN:
- Image: wan2.7-image-pro, wan2.7-image (sync, T2I/I2I)
- Video: wan2.7-t2v, wan2.7-i2v, wan2.7-r2v, wan2.7-videoedit (async)
- 6 config nodes: T2I, I2I, T2V, I2V + R2V/VideoEdit adapters ready
HappyHorse Bailian CN:
- Video: happyhorse-1.0-t2v, happyhorse-1.0-i2v, happyhorse-1.0-r2v,
happyhorse-1.0-video-edit (async)
- 2 config nodes: T2V, I2V + R2V/VideoEdit adapters ready
Shared infrastructure:
- DASHSCOPE_API_KEY Bearer token auth
- All video models share POST /services/aigc/video-generation/video-synthesis
- Query: GET /tasks/{task_id}
- X-DashScope-Async: enable header for all video requests
- Video URL in output.video_url
Implementation:
- 9 API documentation snapshots in docs/providers/bailian/
- 50 new tests (30 Wan + 20 HappyHorse), 220 total
- 6 UI workflow examples and 6 API smoke fixtures
- JS dynamic widgets for Wan image size tiers and HappyHorse duration
- Architecture isolation tests pass
84 lines
3.3 KiB
Python
84 lines
3.3 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
import nodes
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_every_api_fixture_has_a_user_workflow():
|
|
fixtures = {path.name.replace("_prompt.json", "") for path in (ROOT / "tests/fixtures").glob("*_prompt.json")}
|
|
examples = {path.name.replace("_preview.json", "") for path in (ROOT / "examples").glob("*_preview.json")}
|
|
assert fixtures == examples
|
|
assert len(fixtures) == 18
|
|
|
|
|
|
def test_user_workflow_widget_values_match_custom_node_contracts():
|
|
for path in (ROOT / "examples").glob("*_preview.json"):
|
|
workflow = json.loads(path.read_text())
|
|
for item in workflow["nodes"]:
|
|
cls = nodes.NODE_CLASS_MAPPINGS.get(item["type"])
|
|
if cls is None:
|
|
continue
|
|
input_types = cls.INPUT_TYPES()
|
|
widget_names = []
|
|
for section in ("required", "optional"):
|
|
for name, config in input_types.get(section, {}).items():
|
|
if not (len(config) > 1 and config[1].get("forceInput")):
|
|
widget_names.append(name)
|
|
assert len(item["widgets_values"]) == len(widget_names), (
|
|
path.name,
|
|
item["type"],
|
|
widget_names,
|
|
item["widgets_values"],
|
|
)
|
|
|
|
|
|
def test_all_provider_workflows_use_generic_execution_nodes():
|
|
for path in (ROOT / "tests/fixtures").glob("*_prompt.json"):
|
|
classes = {item["class_type"] for item in json.loads(path.read_text()).values()}
|
|
if "image" in path.name:
|
|
assert "MediaHubGenerateImage" in classes
|
|
if "video" in path.name or "t2v" in path.name or "i2v" in path.name:
|
|
assert {"MediaHubSubmitVideo", "MediaHubResolveVideo"} <= classes
|
|
|
|
|
|
def test_user_workflows_follow_three_region_layout():
|
|
for path in (ROOT / "examples").glob("*_preview.json"):
|
|
workflow = json.loads(path.read_text())
|
|
groups = {group["title"]: group for group in workflow["groups"]}
|
|
assert set(groups) == {"Inputs / 输入", "Provider Config / 配置", "Outputs / 输出"}
|
|
|
|
input_types = {"PrimitiveStringMultiline", "LoadImage"}
|
|
config_types = {
|
|
"MediaHubAPIKey",
|
|
"MediaHubGenerateImage",
|
|
"MediaHubSubmitVideo",
|
|
"MediaHubResolveVideo",
|
|
}
|
|
output_types = {"PreviewImage", "SaveVideo"}
|
|
|
|
for item in workflow["nodes"]:
|
|
node_type = item["type"]
|
|
if node_type in input_types:
|
|
assert item["pos"][0] < 580
|
|
elif node_type.endswith("Config") or node_type in config_types:
|
|
assert 580 <= item["pos"][0] < 1860
|
|
elif node_type in output_types:
|
|
assert item["pos"][0] >= 1860
|
|
else:
|
|
raise AssertionError((path.name, node_type))
|
|
|
|
api_key = next(item for item in workflow["nodes"] if item["type"] == "MediaHubAPIKey")
|
|
assert api_key["size"][1] <= 90
|
|
|
|
|
|
def test_user_workflow_links_flow_left_to_right():
|
|
for path in (ROOT / "examples").glob("*_preview.json"):
|
|
workflow = json.loads(path.read_text())
|
|
x_by_id = {item["id"]: item["pos"][0] for item in workflow["nodes"]}
|
|
for link in workflow["links"]:
|
|
_, origin_id, _, target_id, _, _ = link
|
|
assert x_by_id[origin_id] < x_by_id[target_id], (path.name, link)
|