Files
Van/攻城狮鹿鸣君 4cfe6c4e40 prepare a verifiable first Registry release
Establish the immutable mediahub-cloud identity, public documentation, branding, contribution contract, CI, guarded publishing workflow, and a live-smoke regression fix before the first public release.

Constraint: Registry versions and node IDs are immutable after publication

Rejected: Automatic publish on pyproject.toml changes | too easy to publish an unverified immutable version

Confidence: high

Scope-risk: moderate

Directive: Run paid provider-family live verification before triggering the manual Registry workflow

Tested: 221 pytest tests; JavaScript syntax; Python compilation; workflow JSON validation; restarted ComfyUI smoke; clean editable install; comfy node validate; comfy node pack inspection

Not-tested: Paid live API calls for every advertised provider family
2026-06-13 22:34:55 +08:00

124 lines
5.4 KiB
Python

import json
import subprocess
import sys
from pathlib import Path
import torch
import nodes
from mediahub.core.contracts import ImageResult, MediaJob, MediaRequest
def test_provider_config_nodes_are_separate_from_generic_execution_nodes():
assert "provider" not in nodes.MiniMaxCNT2IConfig.INPUT_TYPES()["required"]
assert "provider" not in nodes.AgnesImage21T2IConfig.INPUT_TYPES()["required"]
assert nodes.MiniMaxCNT2IConfig.RETURN_TYPES == ("MEDIAHUB_IMAGE_REQUEST",)
assert nodes.MediaHubGenerateImage.INPUT_TYPES()["required"]["request"][0] == "MEDIAHUB_IMAGE_REQUEST"
assert nodes.MediaHubSubmitVideo.RETURN_TYPES == ("MEDIAHUB_JOB", "STRING")
assert nodes.MediaHubResolveVideo.INPUT_TYPES()["required"]["job"][0] == "MEDIAHUB_JOB"
def test_minimax_config_builds_request_without_api_key():
request = nodes.MiniMaxCNT2IConfig().build(
prompt="cat", model="image-01", aspect_ratio="1:1", use_seed=False, seed=0,
n=1, response_format="url", prompt_optimizer=False, aigc_watermark=False,
use_style=False, style_type="漫画", style_weight=0.8,
size_mode="aspect_ratio", width=1024, height=1024,
)[0]
assert request.provider == "MiniMax CN"
assert request.operation == "image.generate"
assert "api_key" not in request.payload
def test_agnes_i2i_config_converts_comfy_image_to_data_uri():
request = nodes.AgnesImage21I2IConfig().build(
prompt="edit", width=1024, height=768, response_format="url",
image=torch.zeros((1, 4, 4, 3), dtype=torch.float32),
)[0]
assert request.provider == "Agnes"
assert request.payload["image"][0].startswith("data:image/jpeg;base64,")
def test_generic_image_node_uses_service(monkeypatch):
tensor = torch.zeros((1, 2, 2, 3), dtype=torch.float32)
monkeypatch.setattr(nodes, "generate_image", lambda key, request: ImageResult([("url", "x")], ["x"], {"ok": True}))
monkeypatch.setattr(nodes, "image_sources_to_tensor_batch", lambda sources: tensor)
result = nodes.MediaHubGenerateImage().generate("key", MediaRequest("Agnes", "image.generate", {}))
assert result[0] is tensor
assert json.loads(result[1]) == ["x"]
def test_generic_video_nodes_use_opaque_job(monkeypatch):
job = MediaJob("Agnes", "video.resolve", "task", {"video_id": "video"})
monkeypatch.setattr(nodes, "submit_video", lambda key, request: job)
result = nodes.MediaHubSubmitVideo().submit("key", MediaRequest("Agnes", "video.submit", {}))
assert result[0] is job
assert "api_key" not in json.loads(result[1])
def test_api_key_node_is_password_and_connectable():
assert nodes.MediaHubAPIKey.INPUT_TYPES()["required"]["api_key"][1]["password"] is True
assert nodes.MediaHubAPIKey().get_api_key("secret") == ("secret",)
def test_seed_volcengine_i2v_last_frame_is_optional():
input_types = nodes.SeedVolcengineI2VConfig.INPUT_TYPES()
assert "first_frame_image" in input_types["required"]
assert "last_frame_image" not in input_types["required"]
assert "last_frame_image" in input_types["optional"]
def test_old_nodes_are_removed_and_new_nodes_are_exported():
assert "MediaHubTextToImage" not in nodes.NODE_CLASS_MAPPINGS
assert "MediaHubT2VSubmit" not in nodes.NODE_CLASS_MAPPINGS
for name in (
"MiniMaxCNT2IConfig", "MiniMaxCNI2IConfig", "AgnesImage21T2IConfig",
"AgnesImage21I2IConfig", "MediaHubGenerateImage", "MiniMaxCNT2VConfig",
"MiniMaxCNI2VConfig", "AgnesVideo20T2VConfig", "AgnesVideo20I2VConfig",
"MediaHubSubmitVideo", "MediaHubResolveVideo", "MediaHubAPIKey",
):
assert name in nodes.NODE_CLASS_MAPPINGS
def test_provider_config_nodes_are_grouped_below_provider_category():
provider_nodes = {
nodes.MiniMaxCNT2IConfig: "MiniMax CN",
nodes.MiniMaxCNI2IConfig: "MiniMax CN",
nodes.MiniMaxCNT2VConfig: "MiniMax CN",
nodes.MiniMaxCNI2VConfig: "MiniMax CN",
nodes.AgnesImage21T2IConfig: "Agnes",
nodes.AgnesImage21I2IConfig: "Agnes",
nodes.AgnesVideo20T2VConfig: "Agnes",
nodes.AgnesVideo20I2VConfig: "Agnes",
}
for node, provider in provider_nodes.items():
assert node.CATEGORY.startswith(f"MediaHub/Provider/{provider}/")
def test_plugin_loads_as_isolated_comfyui_package(tmp_path):
root = Path(__file__).resolve().parents[1]
script = f"""
import importlib.util, sys
spec = importlib.util.spec_from_file_location("mediahub_test", {str(root / "__init__.py")!r}, submodule_search_locations=[{str(root)!r}])
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
assert "AgnesImage21T2IConfig" in module.NODE_CLASS_MAPPINGS
assert "MediaHubGenerateImage" in module.NODE_CLASS_MAPPINGS
"""
result = subprocess.run([sys.executable, "-c", script], cwd=tmp_path, capture_output=True, text=True)
assert result.returncode == 0, result.stderr
def test_frontend_keeps_password_dom_widget_and_new_config_classes():
source = (Path(__file__).resolve().parents[1] / "js" / "mediahub.js").read_text()
assert 'input.type = "password"' in source
assert "addDOMWidget" in source
assert "MiniMaxCNT2IConfig" in source
assert "AgnesVideo20T2VConfig" in source
assert "widget.hidden" not in source
assert 'container.className = "mediahub-api-key-row"' in source
assert 'toggle.title = "Show API Key"' in source
assert 'node.setSize([Math.max(node.size[0], 300), 72])' in source