zlei6/ComfyUI
1
0
Fork 0
mirror of https://github.com/Comfy-Org/ComfyUI.git synced 2026-08-11 04:22:27 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
comfy-cloud-partner
ComfyUI/tests-unit/comfy_api_nodes_test/comfy_cloud_test.py
Hunter Senft-Grupp 83dcd5b714 Fix Comfy Cloud node safety 
Amp-Thread-ID: https://ampcode.com/threads/T-019fd9e0-653f-74cb-a744-9d35f9264778
Co-authored-by: Amp <amp@ampcode.com>
2026-08-08 17:32:46 +00:00

1078 lines
48 KiB
Python
Raw Permalink Blame History

import asyncio
from io import BytesIO
from typing import get_args
from unittest.mock import AsyncMock, Mock
import aiohttp
import pytest
import torch
from comfy.cli_args import args
if not torch.cuda.is_available():
args.cpu = True
from comfy_api_nodes.apis.comfy_cloud import (
ComfyCloudGenerateRequest,
ComfyCloudGenerateResponse,
ComfyCloudStatusResponse,
ComfyCloudWorkflow,
ComfyCloudWorkflowInputs,
)
from comfy_api_nodes import nodes_comfy_cloud
from comfy_api_nodes.util import conversions, download_helpers
@pytest.mark.parametrize(
("node", "workflow", "returns_video", "requires_image"),
[
(nodes_comfy_cloud.ComfyCloudTextToImageNode, "text-to-image", False, False),
(nodes_comfy_cloud.ComfyCloudTextToVideoNode, "text-to-video", True, False),
(nodes_comfy_cloud.ComfyCloudImageToVideoNode, "image-to-video", True, True),
(nodes_comfy_cloud.ComfyCloudImageEditNode, "image-edit", False, True),
],
)
def test_workflow_submission_polling_and_download(monkeypatch, node, workflow, returns_video, requires_image):
sync = AsyncMock(
return_value=ComfyCloudGenerateResponse(
task_id="task-1",
status="queued",
polling_url="/proxy/comfy-cloud/workflow/tasks/task-1",
cancel_url="/proxy/comfy-cloud/workflow/tasks/task-1/cancel",
)
)
poll = AsyncMock(
return_value=ComfyCloudStatusResponse(
task_id="task-1",
status="completed",
progress=100,
output_url="/proxy/comfy-cloud/results/task-1/output",
)
)
upload = AsyncMock(return_value="https://example.com/input.png")
image_download = AsyncMock(return_value="image-output")
video_download = AsyncMock(return_value="video-output")
monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync)
monkeypatch.setattr(nodes_comfy_cloud, "poll_op", poll)
monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload)
monkeypatch.setattr(nodes_comfy_cloud, "download_url_to_image_tensor", image_download)
monkeypatch.setattr(nodes_comfy_cloud, "download_url_to_video_output", video_download)
monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1)
image = object() if requires_image else None
output = asyncio.run(node.execute("A tiny fennec fox", image))
endpoint = sync.call_args.args[1]
request = sync.call_args.kwargs["data"]
assert endpoint.path == "/proxy/comfy-cloud/workflow/generate"
assert endpoint.method == "POST"
assert request == ComfyCloudGenerateRequest(
workflow=workflow,
inputs=ComfyCloudWorkflowInputs(
prompt="A tiny fennec fox",
image_url="https://example.com/input.png" if requires_image else None,
),
)
assert upload.await_count == int(requires_image)
poll_endpoint = poll.call_args.args[1]
cancel_endpoint = poll.call_args.kwargs["cancel_endpoint"]
assert poll_endpoint.path == "/proxy/comfy-cloud/workflow/tasks/task-1"
assert cancel_endpoint.path == "/proxy/comfy-cloud/workflow/tasks/task-1/cancel"
assert cancel_endpoint.method == "POST"
assert output[0] == ("video-output" if returns_video else "image-output")
@pytest.mark.parametrize(
"node",
[nodes_comfy_cloud.ComfyCloudImageToVideoNode, nodes_comfy_cloud.ComfyCloudImageEditNode],
)
def test_image_workflows_reject_batches(monkeypatch, node):
upload = AsyncMock()
monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 2)
monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload)
with pytest.raises(ValueError, match="Exactly one input image"):
asyncio.run(node.execute("Animate this", object()))
upload.assert_not_awaited()
def test_contract_omits_optional_status_fields():
request = ComfyCloudGenerateRequest(
workflow="text-to-image",
inputs=ComfyCloudWorkflowInputs(prompt="A lighthouse"),
)
status = ComfyCloudStatusResponse(task_id="task-1", status="queued")
assert request.model_dump(exclude_none=True) == {
"workflow": "text-to-image",
"inputs": {"prompt": "A lighthouse"},
}
assert status.model_dump(exclude_none=True) == {"task_id": "task-1", "status": "queued"}
def test_status_progress_is_clamped_for_display():
assert nodes_comfy_cloud._progress(ComfyCloudStatusResponse(task_id="task-1", status="running", progress=100.5)) == 100
assert nodes_comfy_cloud._progress(ComfyCloudStatusResponse(task_id="task-1", status="running", progress=-1)) == 0
def test_poll_failure_cancels_submitted_task(monkeypatch):
poll = AsyncMock(side_effect=ValueError("invalid status response"))
cancel = AsyncMock(return_value={"status": "cancellation_requested"})
monkeypatch.setattr(nodes_comfy_cloud, "poll_op", poll)
monkeypatch.setattr(nodes_comfy_cloud, "sync_op_raw", cancel)
with pytest.raises(ValueError, match="invalid status response"):
asyncio.run(nodes_comfy_cloud._poll_task(nodes_comfy_cloud.ComfyCloudTextToImageNode, "task/1"))
cancel.assert_awaited_once()
assert cancel.call_args.args[1].path == "/proxy/comfy-cloud/workflow/tasks/task%2F1/cancel"
assert cancel.call_args.kwargs["max_retries"] == 0
@pytest.mark.parametrize("response_model", [ComfyCloudGenerateResponse, ComfyCloudStatusResponse])
@pytest.mark.parametrize("task_id", ["", " "])
def test_contract_rejects_empty_task_ids(response_model, task_id):
values = {"task_id": task_id, "status": "queued"}
if response_model is ComfyCloudGenerateResponse:
values.update(polling_url="/poll", cancel_url="/cancel")
with pytest.raises(ValueError, match="task_id"):
response_model(**values)
@pytest.mark.parametrize(
"url",
[
"http://example.com/output.png",
"http://127.0.0.1/output.png",
"//169.254.169.254/latest/meta-data",
"/unrelated/path/output.png",
"https://user@example.com/output.png",
"/proxy/comfy-cloud/../../v1/users/me",
"/proxy/comfy-cloud/%2e%2e/%2e%2e/v1/users/me",
"https://127.0.0.1/output.png",
"https://169.254.169.254/latest/meta-data",
"https://attacker.example/output.png",
],
)
def test_cloud_workflows_reject_untrusted_output_urls(monkeypatch, url):
sync = AsyncMock(
return_value=ComfyCloudGenerateResponse(
task_id="task-1",
status="queued",
polling_url="/poll",
cancel_url="/cancel",
)
)
poll = AsyncMock(
return_value=ComfyCloudStatusResponse(task_id="task-1", status="completed", output_url=url)
)
download = AsyncMock()
monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync)
monkeypatch.setattr(nodes_comfy_cloud, "poll_op", poll)
monkeypatch.setattr(nodes_comfy_cloud, "download_url_to_image_tensor", download)
with pytest.raises(RuntimeError, match="invalid output URL"):
asyncio.run(nodes_comfy_cloud.ComfyCloudTextToImageNode.execute("prompt"))
download.assert_not_awaited()
def test_cloud_workflows_accept_signed_https_output_urls(monkeypatch):
sync = AsyncMock(
return_value=ComfyCloudGenerateResponse(
task_id="task-1",
status="queued",
polling_url="/poll",
cancel_url="/cancel",
)
)
poll = AsyncMock(
return_value=ComfyCloudStatusResponse(
task_id="task-1",
status="completed",
output_url="https://storage.googleapis.com/comfy-cloud/output.png?signature=example",
)
)
download = AsyncMock(return_value="image-output")
monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync)
monkeypatch.setattr(nodes_comfy_cloud, "poll_op", poll)
monkeypatch.setattr(nodes_comfy_cloud, "download_url_to_image_tensor", download)
output = asyncio.run(nodes_comfy_cloud.ComfyCloudTextToImageNode.execute("prompt"))
assert output[0] == "image-output"
download.assert_awaited_once_with(
"https://storage.googleapis.com/comfy-cloud/output.png?signature=example",
timeout=nodes_comfy_cloud._OUTPUT_DOWNLOAD_TIMEOUT,
cls=nodes_comfy_cloud.ComfyCloudTextToImageNode,
allow_redirects=False,
)
@pytest.mark.parametrize(
"node",
[
nodes_comfy_cloud.ComfyCloudTextToImageNode,
nodes_comfy_cloud.ComfyCloudTextToVideoNode,
nodes_comfy_cloud.ComfyCloudImageToVideoNode,
nodes_comfy_cloud.ComfyCloudImageEditNode,
],
)
def test_legacy_nodes_reject_oversized_prompts(monkeypatch, node):
sync = AsyncMock()
monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync)
with pytest.raises(Exception, match="4096"):
asyncio.run(node.execute("x" * 4097, object()))
sync.assert_not_awaited()
def test_legacy_nodes_strip_prompts_before_submission(monkeypatch):
run = AsyncMock(return_value=("output",))
monkeypatch.setattr(nodes_comfy_cloud.ComfyCloudTextToImageNode, "_run", run)
asyncio.run(nodes_comfy_cloud.ComfyCloudTextToImageNode.execute(" prompt "))
assert run.call_args.args[0].prompt == "prompt"
def test_poc_nodes_strip_prompt_fields_before_submission(monkeypatch):
run = AsyncMock(return_value=("output",))
monkeypatch.setattr(nodes_comfy_cloud.ComfyCloudMageFlowImageNode, "_run", run)
asyncio.run(
nodes_comfy_cloud.ComfyCloudMageFlowImageNode.execute(
" prompt ", " avoid this ", "1:1", 0
)
)
inputs = run.call_args.args[0]
assert inputs.prompt == "prompt"
assert inputs.negative_prompt == "avoid this"
def test_task_routes_ignore_response_urls_and_errors_hide_task_token(monkeypatch):
sync = AsyncMock(
return_value=ComfyCloudGenerateResponse(
task_id="secret/task-token",
status="queued",
polling_url="https://attacker.example/poll",
cancel_url="https://attacker.example/cancel",
)
)
poll = AsyncMock(
return_value=ComfyCloudStatusResponse(
task_id="secret/task-token",
status="completed",
error="provider details with secret/task-token",
)
)
monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync)
monkeypatch.setattr(nodes_comfy_cloud, "poll_op", poll)
with pytest.raises(RuntimeError) as error:
asyncio.run(nodes_comfy_cloud.ComfyCloudTextToVideoNode.execute("A prompt"))
assert poll.call_args.args[1].path == "/proxy/comfy-cloud/workflow/tasks/secret%2Ftask-token"
assert poll.call_args.kwargs["cancel_endpoint"].path == "/proxy/comfy-cloud/workflow/tasks/secret%2Ftask-token/cancel"
assert "task-token" not in str(error.value)
assert "provider details" not in str(error.value)
IMAGE_POC_NODES = [
(
nodes_comfy_cloud.ComfyCloudIdeogram4DesignNode,
"image.ideogram-4-design.v1",
["prompt", "aspect_ratio", "quality_mode", "seed"],
{
"prompt": "A geometric fox logo",
"aspect_ratio": "21:9",
"quality_mode": "fast",
"seed": 11,
},
),
(
nodes_comfy_cloud.ComfyCloudKrea2CreativeImageNode,
"image.krea-2-creative-image.v1",
["prompt", "prompt_enhance", "aspect_ratio", "seed"],
{"prompt": "A glass forest", "prompt_enhance": False, "aspect_ratio": "16:9", "seed": 12},
),
(
nodes_comfy_cloud.ComfyCloudMageFlowImageNode,
"image.mage-flow-image.v1",
["prompt", "negative_prompt", "aspect_ratio", "seed"],
{"prompt": "A moonlit lake", "negative_prompt": "fog", "aspect_ratio": "3:2", "seed": 13},
),
(
nodes_comfy_cloud.ComfyCloudFlux2ReferenceEditNode,
"image.flux-2-reference-edit.v1",
["image", "instruction", "guidance", "quality_mode", "seed"],
{"image": object(), "instruction": "Make it winter", "guidance": 5.5, "quality_mode": "fast", "seed": 14},
),
(
nodes_comfy_cloud.ComfyCloudQwenImageEdit2511Node,
"image.qwen-image-edit-2511.v1",
["image", "instruction", "quality_mode", "seed"],
{"image": object(), "instruction": "Remove the sign", "quality_mode": "fast", "seed": 15},
),
(
nodes_comfy_cloud.ComfyCloudSeedVR2ImageUpscaleNode,
"image.seedvr2-image-upscale.v1",
["image", "scale"],
{"image": object(), "scale": "2x"},
),
]
@pytest.mark.parametrize(("node", "workflow", "input_names", "arguments"), IMAGE_POC_NODES)
def test_image_poc_node_schema_and_request_mapping(monkeypatch, node, workflow, input_names, arguments):
sync = AsyncMock(
return_value=ComfyCloudGenerateResponse(
task_id="task-poc",
status="queued",
polling_url="/tasks/task-poc",
cancel_url="/tasks/task-poc/cancel",
)
)
poll = AsyncMock(
return_value=ComfyCloudStatusResponse(
task_id="task-poc",
status="completed",
output_url="/proxy/comfy-cloud/results/task-poc/image.png",
)
)
upload = AsyncMock(return_value="/uploads/input.png")
download = AsyncMock(return_value="image-output")
monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync)
monkeypatch.setattr(nodes_comfy_cloud, "poll_op", poll)
monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload)
monkeypatch.setattr(nodes_comfy_cloud, "download_url_to_image_tensor", download)
monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1)
schema = node.define_schema()
assert schema.node_id == node.node_id
assert schema.display_name == node.display_name
assert schema.is_api_node is True
assert [input.id for input in schema.inputs] == input_names
assert len(schema.outputs) == 1
assert schema.outputs[0].get_io_type() == "IMAGE"
output = asyncio.run(node.execute(**arguments))
request = sync.call_args.kwargs["data"]
expected_inputs = {key: value for key, value in arguments.items() if key != "image"}
if "image" in arguments:
expected_inputs["assets"] = {"image": {"type": "IMAGE", "url": "/uploads/input.png"}}
assert request.workflow == workflow
assert request.inputs.model_dump(exclude_none=True) == expected_inputs
assert "asset_id" not in request.model_dump_json()
assert '"id"' not in request.model_dump_json()
assert upload.await_count == int("image" in arguments)
if "image" in arguments:
assert upload.call_args.kwargs == {"total_pixels": None}
download.assert_awaited_once_with(
"/proxy/comfy-cloud/results/task-poc/image.png",
timeout=30 * 60,
cls=node,
allow_redirects=False,
)
assert output[0] == "image-output"
def test_image_poc_schema_defaults_ranges_and_enums():
schemas = {
node.workflow: {input.id: input for input in node.define_schema().inputs}
for node, _, _, _ in IMAGE_POC_NODES
}
aspect_ratios = ["1:1", "3:4", "2:3", "3:2", "4:3", "16:9", "9:16", "21:9"]
for workflow in [
"image.ideogram-4-design.v1",
"image.krea-2-creative-image.v1",
"image.mage-flow-image.v1",
]:
assert schemas[workflow]["aspect_ratio"].options == aspect_ratios
assert schemas[workflow]["aspect_ratio"].default == "1:1"
seed = schemas[workflow]["seed"]
assert (seed.default, seed.min, seed.max) == (0, 0, 0xFFFFFFFFFFFFFFFF)
assert schemas["image.ideogram-4-design.v1"]["quality_mode"].options == ["quality", "balanced", "fast"]
assert schemas["image.ideogram-4-design.v1"]["quality_mode"].default == "balanced"
assert schemas["image.krea-2-creative-image.v1"]["prompt_enhance"].default is True
assert schemas["image.mage-flow-image.v1"]["negative_prompt"].default == ""
guidance = schemas["image.flux-2-reference-edit.v1"]["guidance"]
assert (guidance.default, guidance.min, guidance.max, guidance.step) == (4.0, 1.0, 10.0, 0.1)
for workflow in ["image.flux-2-reference-edit.v1", "image.qwen-image-edit-2511.v1"]:
assert schemas[workflow]["quality_mode"].options == ["quality", "fast"]
assert schemas[workflow]["quality_mode"].default == "quality"
seed = schemas[workflow]["seed"]
assert (seed.default, seed.min, seed.max) == (0, 0, 0xFFFFFFFFFFFFFFFF)
scale = schemas["image.seedvr2-image-upscale.v1"]["scale"]
assert scale.options == ["2x", "4x"]
assert scale.default == "4x"
def test_image_poc_api_declarations_and_extension_registration():
workflows = {workflow for _, workflow, _, _ in IMAGE_POC_NODES}
registered = set(asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list()))
assert workflows <= set(get_args(ComfyCloudWorkflow))
assert {node for node, _, _, _ in IMAGE_POC_NODES} <= registered
def test_cloud_workflow_controls_have_connection_sockets():
nodes = asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list())
for node in nodes:
for input_spec in node.define_schema().inputs:
if isinstance(input_spec, nodes_comfy_cloud.IO.WidgetInput):
assert input_spec.socketless is False, f"{node.__name__}.{input_spec.id}"
def test_cloud_workflow_schemas_have_descriptions():
nodes = asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list())
for node in nodes:
assert node.define_schema().description.strip(), node.__name__
def test_cloud_workflow_schemas_share_exact_estimated_rate_metadata():
nodes = asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list())
assert nodes_comfy_cloud.COMFY_CLOUD_GPU_SECOND_USD == 0.001295
assert nodes_comfy_cloud.COMFY_CLOUD_CREDITS_PER_USD == 211
assert nodes_comfy_cloud.COMFY_CLOUD_GPU_SECOND_CREDITS == pytest.approx(0.273245)
assert nodes_comfy_cloud.COMFY_CLOUD_GPU_HOUR_USD == pytest.approx(4.662)
assert nodes_comfy_cloud.COMFY_CLOUD_GPU_HOUR_CREDITS == pytest.approx(983.682)
for node in nodes:
schema = node.define_schema()
badge = schema.price_badge.as_dict(schema.inputs)
assert badge["expr"] == (
'{"type":"usd","usd":0.001295,"format":{"suffix":"/GPU-second","approximate":true}}'
)
assert "Estimated compute rate" in schema.description
assert "Actual final cost depends on GPU runtime" in schema.description
def test_all_linkable_widget_constraints_are_validated():
nodes = asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list())
for node in nodes:
for input_spec in node.define_schema().inputs:
if not isinstance(input_spec, nodes_comfy_cloud.IO.WidgetInput):
continue
io_type = input_spec.get_io_type()
if io_type == "COMBO":
invalid = "not-an-option"
elif io_type == "BOOLEAN":
invalid = "true"
elif io_type == "INT":
invalid = (input_spec.min - 1) if input_spec.min is not None else 1.5
elif io_type == "FLOAT":
invalid = float("nan")
elif io_type == "STRING" and nodes_comfy_cloud._TEXT_LIMITS.get(input_spec.id, (0,))[0]:
invalid = " "
else:
continue
with pytest.raises((ValueError, Exception), match=input_spec.id):
nodes_comfy_cloud._validate_node_inputs(node, {input_spec.id: invalid})
def test_linked_values_are_validated_before_upload(monkeypatch):
upload = AsyncMock()
monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload)
monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1)
with pytest.raises(ValueError, match="guidance"):
asyncio.run(
nodes_comfy_cloud.ComfyCloudFlux2ReferenceEditNode.execute(
object(), "instruction", float("nan"), "quality", 0
)
)
upload.assert_not_awaited()
def test_upload_inputs_have_decoded_resource_limits():
oversized_image = torch.empty((1, 8193, 1, 3), device="meta")
oversized_audio = {
"waveform": torch.empty((1, 2, nodes_comfy_cloud._MAX_DECODED_AUDIO_BYTES // 8 + 1), device="meta"),
"sample_rate": 48000,
}
with pytest.raises(ValueError, match="32-megapixel"):
nodes_comfy_cloud._validate_image_upload(oversized_image)
with pytest.raises(ValueError, match="256 MiB"):
nodes_comfy_cloud._validate_audio_upload(oversized_audio)
@pytest.mark.parametrize(
("node", "input_names"),
[
(nodes_comfy_cloud.ComfyCloudMiniMaxH3TextSoundNode, ["prompt", "aspect_ratio", "duration_seconds", "seed"]),
(nodes_comfy_cloud.ComfyCloudMiniMaxH3ImageSoundNode, ["image", "prompt", "aspect_ratio", "duration_seconds", "seed"]),
(nodes_comfy_cloud.ComfyCloudLTX23ImageAudioPerformanceNode, ["image", "audio", "prompt", "enhance_prompt", "duration_seconds", "seed"]),
(nodes_comfy_cloud.ComfyCloudLTX23FirstLastFrameNode, ["first_frame", "last_frame", "prompt", "duration_seconds", "seed"]),
(nodes_comfy_cloud.ComfyCloudWan22FirstLastFrameNode, ["first_frame", "last_frame", "prompt", "negative_prompt", "duration_seconds", "seed"]),
(nodes_comfy_cloud.ComfyCloudSCAIL2CharacterReplacementNode, ["reference_character", "driving_video", "scene_prompt", "driving_subject", "reference_subject", "seed"]),
],
)
def test_video_node_schemas_expose_only_manifest_inputs(node, input_names):
schema = node.define_schema()
assert schema.is_api_node
assert [input.id for input in schema.inputs] == input_names
assert len(schema.outputs) == 1
assert schema.outputs[0].get_io_type() == "VIDEO"
def test_ltx_performance_stages_image_and_audio(monkeypatch):
run = AsyncMock(return_value=("video-output",))
image_upload = AsyncMock(return_value="https://example.com/image.png")
audio_upload = AsyncMock(return_value="https://example.com/audio.mp4")
monkeypatch.setattr(nodes_comfy_cloud, "_run_video_workflow", run)
monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", image_upload)
monkeypatch.setattr(nodes_comfy_cloud, "upload_audio_to_comfyapi", audio_upload)
monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1)
audio = {"waveform": torch.zeros(1, 1, 480000), "sample_rate": 48000}
asyncio.run(nodes_comfy_cloud.ComfyCloudLTX23ImageAudioPerformanceNode.execute(object(), audio, "sing", True, 9, 7))
inputs = run.call_args.args[2]
assert inputs.image_url == "https://example.com/image.png"
assert inputs.audio_url == "https://example.com/audio.mp4"
assert inputs.duration_seconds == 9
def test_scail_stages_reference_image_and_driving_video(monkeypatch):
run = AsyncMock(return_value=("video-output",))
image_upload = AsyncMock(return_value="https://example.com/character.png")
video_upload = AsyncMock(return_value="https://example.com/driving.mp4")
video = Mock()
video.get_frame_count.return_value = 100
monkeypatch.setattr(nodes_comfy_cloud, "_run_video_workflow", run)
monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", image_upload)
monkeypatch.setattr(nodes_comfy_cloud, "upload_video_to_comfyapi", video_upload)
monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1)
asyncio.run(nodes_comfy_cloud.ComfyCloudSCAIL2CharacterReplacementNode.execute(object(), video, "park", "woman", "human", 1))
inputs = run.call_args.args[2]
assert inputs.reference_character_url == "https://example.com/character.png"
assert inputs.driving_video_url == "https://example.com/driving.mp4"
video.get_frame_count.assert_called_once()
def test_scail_defaults_and_frame_count_fail_closed_before_upload(monkeypatch):
schema = {input.id: input for input in nodes_comfy_cloud.ComfyCloudSCAIL2CharacterReplacementNode.define_schema().inputs}
image_upload = AsyncMock()
video_upload = AsyncMock()
video = Mock()
video.get_frame_count.side_effect = RuntimeError("decode failed")
monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", image_upload)
monkeypatch.setattr(nodes_comfy_cloud, "upload_video_to_comfyapi", video_upload)
monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1)
assert schema["driving_subject"].default == "human"
with pytest.raises(ValueError, match="Unable to determine video frame count"):
asyncio.run(nodes_comfy_cloud.ComfyCloudSCAIL2CharacterReplacementNode.execute(object(), video, "park", "human", "human", 1))
image_upload.assert_not_awaited()
video_upload.assert_not_awaited()
def test_scail_rejects_missing_frame_count_metadata_before_upload(monkeypatch):
upload = AsyncMock()
video = Mock()
video.get_frame_count.return_value = None
monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload)
monkeypatch.setattr(nodes_comfy_cloud, "upload_video_to_comfyapi", upload)
monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1)
with pytest.raises(ValueError, match="Unable to determine video frame count"):
asyncio.run(
nodes_comfy_cloud.ComfyCloudSCAIL2CharacterReplacementNode.execute(
object(), video, "park", "human", "human", 1
)
)
upload.assert_not_awaited()
@pytest.mark.parametrize("frame_count", [80, 158])
def test_scail_rejects_out_of_range_frames_before_upload(monkeypatch, frame_count):
upload = AsyncMock()
video = Mock()
video.get_frame_count.return_value = frame_count
monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload)
monkeypatch.setattr(nodes_comfy_cloud, "upload_video_to_comfyapi", upload)
monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1)
with pytest.raises(ValueError, match="frame count"):
asyncio.run(nodes_comfy_cloud.ComfyCloudSCAIL2CharacterReplacementNode.execute(object(), video, "park", "human", "human", 1))
upload.assert_not_awaited()
def test_in_memory_download_resets_retry_and_enforces_stream_limit(monkeypatch):
class Content:
def __init__(self, chunks):
self.chunks = iter(chunks)
self.finished = False
async def read(self, size):
chunk = next(self.chunks)
if isinstance(chunk, Exception):
raise chunk
if not chunk:
self.finished = True
return chunk
def at_eof(self):
return self.finished
class Response:
status = 200
headers = {}
def __init__(self, chunks, content_length=None):
self.content = Content(chunks)
self.content_length = content_length
async def __aenter__(self):
return self
async def __aexit__(self, *args):
return False
responses = [Response([b"partial", aiohttp.ClientPayloadError("retry")]), Response([b"final", b""])]
class Session:
def __init__(self, timeout):
pass
async def get(self, url, headers, allow_redirects=True):
return responses.pop(0)
async def close(self):
pass
monkeypatch.setattr(download_helpers.aiohttp, "ClientSession", Session)
monkeypatch.setattr(download_helpers, "sleep_with_interrupt", AsyncMock())
destination = BytesIO()
asyncio.run(download_helpers.download_url_to_bytesio("https://example.com/result", destination))
assert destination.read() == b"final"
monkeypatch.setattr(download_helpers, "_MAX_IN_MEMORY_DOWNLOAD_BYTES", 4)
responses.append(Response([b"12345", b""]))
with pytest.raises(ValueError, match="in-memory limit"):
asyncio.run(download_helpers.download_url_to_bytesio("https://example.com/result", BytesIO()))
responses.append(Response([], content_length=5))
with pytest.raises(ValueError, match="in-memory limit"):
asyncio.run(download_helpers.download_url_to_bytesio("https://example.com/result", BytesIO()))
def test_file_object_download_resets_retry_and_enforces_stream_limit(monkeypatch, tmp_path):
destination = (tmp_path / "result.bin").open("w+b")
destination.write(b"stale")
class Content:
def __init__(self, chunks):
self.chunks = iter(chunks)
self.finished = False
async def read(self, size):
chunk = next(self.chunks)
if isinstance(chunk, Exception):
raise chunk
if not chunk:
self.finished = True
return chunk
def at_eof(self):
return self.finished
class Response:
status = 200
headers = {}
content_length = None
def __init__(self, chunks):
self.content = Content(chunks)
async def __aenter__(self):
return self
async def __aexit__(self, *args):
return False
responses = [Response([b"partial", aiohttp.ClientPayloadError("retry")]), Response([b"done", b""])]
class Session:
def __init__(self, timeout):
pass
async def get(self, url, headers, allow_redirects=True):
return responses.pop(0)
async def close(self):
pass
monkeypatch.setattr(download_helpers.aiohttp, "ClientSession", Session)
monkeypatch.setattr(download_helpers, "sleep_with_interrupt", AsyncMock())
monkeypatch.setattr(download_helpers, "_MAX_IN_MEMORY_DOWNLOAD_BYTES", 10)
asyncio.run(download_helpers.download_url_to_bytesio("https://example.com/result", destination))
assert destination.read() == b"done"
monkeypatch.setattr(download_helpers, "_MAX_IN_MEMORY_DOWNLOAD_BYTES", 4)
responses.append(Response([b"12345", b""]))
with pytest.raises(ValueError, match="in-memory limit"):
asyncio.run(download_helpers.download_url_to_bytesio("https://example.com/result", destination))
destination.close()
def test_download_cloud_audio_url_to_audio_input(monkeypatch):
node = nodes_comfy_cloud.ComfyCloudTextToImageNode
downloaded = b"encoded audio"
expected = {"waveform": torch.ones(1, 2, 3), "sample_rate": 48000}
download_call = Mock()
async def download(url, dest, **kwargs):
download_call(url=url, dest=dest, **kwargs)
dest.write(downloaded)
dest.seek(0)
audio_decode = Mock(return_value=expected)
monkeypatch.setattr(download_helpers, "download_url_to_bytesio", download)
monkeypatch.setattr(download_helpers, "audio_bytes_to_audio_input", audio_decode)
output = asyncio.run(
download_helpers.download_url_to_audio_input(
"/proxy/comfy-cloud/results/task-1/audio.flac",
timeout=30,
max_retries=2,
cls=node,
)
)
assert output is expected
download_call.assert_called_once()
assert download_call.call_args.kwargs["url"] == "/proxy/comfy-cloud/results/task-1/audio.flac"
assert isinstance(download_call.call_args.kwargs["dest"], BytesIO)
assert download_call.call_args.kwargs["timeout"] == 30
assert download_call.call_args.kwargs["max_retries"] == 2
audio_decode.assert_called_once()
assert isinstance(audio_decode.call_args.args[0], BytesIO)
assert audio_decode.call_args.kwargs == {}
assert download_call.call_args.kwargs["cls"] is node
assert download_call.call_args.kwargs["allow_redirects"] is True
def test_audio_decode_stops_before_exceeding_budget(monkeypatch):
class Frame:
def to_ndarray(self):
return torch.ones(2, 8).numpy()
stream = type(
"Stream",
(),
{"codec_context": type("Codec", (), {"sample_rate": 48000})(), "channels": 2, "index": 0},
)()
class AudioFile:
streams = type("Streams", (), {"audio": [stream]})()
def __enter__(self):
return self
def __exit__(self, *args):
return False
def decode(self, streams):
yield Frame()
yield Frame()
monkeypatch.setattr(conversions.av, "open", lambda source: AudioFile())
monkeypatch.setattr(conversions, "_MAX_DECODED_AUDIO_BYTES", 64)
with pytest.raises(ValueError, match="Decoded audio exceeds"):
conversions.audio_bytes_to_audio_input(BytesIO(b"encoded"))
AUDIO_POC_NODES = [
(nodes_comfy_cloud.ComfyCloudACEStep15XLTurboNode, "audio.ace-step-1-5-xl-turbo.v1", ["style_prompt", "lyrics", "duration_seconds", "seed", "bpm", "time_signature", "language", "key"]),
(nodes_comfy_cloud.ComfyCloudStableAudio3MediumNode, "audio.stable-audio-3-medium.v1", ["prompt", "duration_seconds", "seed", "expand_prompt", "category"]),
(nodes_comfy_cloud.ComfyCloudChatterboxMultilingualVoiceCloneNode, "audio.chatterbox-multilingual-voice-clone.v1", ["text", "voice_reference", "language", "exaggeration", "cfg_weight", "temperature", "seed"]),
(nodes_comfy_cloud.ComfyCloudChatterboxDialogueNode, "audio.chatterbox-dialogue.v1", ["script", "speaker_a_reference", "speaker_b_reference", "exaggeration", "cfg_weight", "temperature", "seed"]),
(nodes_comfy_cloud.ComfyCloudChatterboxVoiceConversionNode, "audio.chatterbox-voice-conversion.v1", ["source_audio", "target_voice_reference", "seed"]),
(nodes_comfy_cloud.ComfyCloudMelBandRoFormerStemSeparationNode, "audio.melbandroformer-stem-separation.v1", ["audio"]),
]
@pytest.mark.parametrize(("node", "workflow", "input_names"), AUDIO_POC_NODES)
def test_audio_poc_schemas_and_registration(node, workflow, input_names):
schema = node.define_schema()
registered = asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list())
assert schema.is_api_node
assert schema.category == "partner/audio/Comfy Cloud"
assert [input.id for input in schema.inputs] == input_names
assert all(output.get_io_type() == "AUDIO" for output in schema.outputs)
assert workflow in get_args(ComfyCloudWorkflow)
assert node in registered
def test_audio_poc_schema_defaults_ranges_and_enums():
schemas = {workflow: {input.id: input for input in node.define_schema().inputs} for node, workflow, _ in AUDIO_POC_NODES}
ace = schemas["audio.ace-step-1-5-xl-turbo.v1"]
assert (ace["duration_seconds"].default, ace["duration_seconds"].min, ace["duration_seconds"].max, ace["duration_seconds"].step) == (120, 10, 300, 0.1)
assert (ace["bpm"].default, ace["bpm"].min, ace["bpm"].max) == (120, 10, 300)
assert ace["time_signature"].options == ["2", "3", "4", "6"]
assert ace["language"].default == "en"
assert ace["key"].default == "E minor"
stable = schemas["audio.stable-audio-3-medium.v1"]
assert stable["category"].options == ["Music", "Instrument", "SFX", "One-shot"]
assert stable["expand_prompt"].default is True
for workflow in [
"audio.chatterbox-multilingual-voice-clone.v1",
"audio.chatterbox-dialogue.v1",
"audio.chatterbox-voice-conversion.v1",
]:
assert schemas[workflow]["seed"].max == 0xFFFFFFFF
mel_schema = nodes_comfy_cloud.ComfyCloudMelBandRoFormerStemSeparationNode.define_schema()
assert [output.id for output in mel_schema.outputs] == ["vocals", "instruments"]
def test_audio_poc_request_mapping_and_named_result_decoding(monkeypatch):
sync = AsyncMock(return_value=ComfyCloudGenerateResponse(task_id="task-audio", status="queued", polling_url="/tasks/task-audio", cancel_url="/tasks/task-audio/cancel"))
poll = AsyncMock(return_value=ComfyCloudStatusResponse(task_id="task-audio", status="completed", output_urls={"vocals": "/proxy/comfy-cloud/results/vocals.mp3", "instruments": "/proxy/comfy-cloud/results/instruments.mp3"}))
upload = AsyncMock(return_value="/uploads/song.m4a")
download = AsyncMock(side_effect=["vocals-audio", "instruments-audio"])
monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync)
monkeypatch.setattr(nodes_comfy_cloud, "poll_op", poll)
monkeypatch.setattr(nodes_comfy_cloud, "upload_audio_to_comfyapi", upload)
monkeypatch.setattr(nodes_comfy_cloud, "download_url_to_audio_input", download)
audio = {"waveform": torch.zeros(1, 2, 48000), "sample_rate": 48000}
output = asyncio.run(nodes_comfy_cloud.ComfyCloudMelBandRoFormerStemSeparationNode.execute(audio))
request = sync.call_args.kwargs["data"]
assert request.workflow == "audio.melbandroformer-stem-separation.v1"
assert request.inputs.model_dump(exclude_none=True) == {"assets": {"audio": {"type": "AUDIO", "url": "/uploads/song.m4a"}}}
assert [call.args[0] for call in download.await_args_list] == [
"/proxy/comfy-cloud/results/vocals.mp3",
"/proxy/comfy-cloud/results/instruments.mp3",
]
assert all(call.kwargs["timeout"] == 30 * 60 for call in download.await_args_list)
assert all(call.kwargs["allow_redirects"] is False for call in download.await_args_list)
assert poll.call_args.kwargs["cancel_endpoint"].path == "/proxy/comfy-cloud/workflow/tasks/task-audio/cancel"
assert tuple(output) == ("vocals-audio", "instruments-audio")
def test_chatterbox_audio_inputs_use_named_staged_assets(monkeypatch):
run = AsyncMock(return_value=("audio-output",))
upload = AsyncMock(side_effect=["/uploads/source.m4a", "/uploads/target.m4a"])
monkeypatch.setattr(nodes_comfy_cloud, "_run_audio_workflow", run)
monkeypatch.setattr(nodes_comfy_cloud, "upload_audio_to_comfyapi", upload)
source = {"waveform": torch.zeros(1, 1, 48000), "sample_rate": 48000}
target = {"waveform": torch.zeros(1, 1, 96000), "sample_rate": 48000}
asyncio.run(nodes_comfy_cloud.ComfyCloudChatterboxVoiceConversionNode.execute(source, target, 7))
inputs = run.call_args.args[2]
assert inputs.model_dump(exclude_none=True) == {
"assets": {
"source_audio": {"type": "AUDIO", "url": "/uploads/source.m4a"},
"target_voice_reference": {"type": "AUDIO", "url": "/uploads/target.m4a"},
},
"seed": 7,
}
assert "audio_url" not in inputs.model_dump(exclude_none=True)
def test_chatterbox_dialogue_rejects_invalid_speaker_labels(monkeypatch):
upload = AsyncMock()
monkeypatch.setattr(nodes_comfy_cloud, "upload_audio_to_comfyapi", upload)
audio = {"waveform": torch.zeros(1, 1, 48000), "sample_rate": 48000}
with pytest.raises(ValueError, match="only speakers A and B"):
asyncio.run(nodes_comfy_cloud.ComfyCloudChatterboxDialogueNode.execute("NARRATOR: Hello", audio, audio, 0.5, 0.5, 0.8, 0))
upload.assert_not_awaited()
def test_chatterbox_dialogue_normalizes_labels_and_continuations(monkeypatch):
run = AsyncMock(return_value=("audio-output",))
monkeypatch.setattr(nodes_comfy_cloud, "_run_audio_workflow", run)
monkeypatch.setattr(nodes_comfy_cloud, "upload_audio_to_comfyapi", AsyncMock(side_effect=["/a", "/b"]))
audio = {"waveform": torch.zeros(1, 1, 48000), "sample_rate": 48000}
asyncio.run(nodes_comfy_cloud.ComfyCloudChatterboxDialogueNode.execute("a: Hello\ncontinued\n\nSpeaker B: Hi", audio, audio, 0.5, 0.5, 0.8, 0))
assert run.call_args.args[2].script == "SPEAKER A: Hello continued\nSPEAKER B: Hi"
def test_chatterbox_dialogue_accepts_colons_and_label_only_lines():
assert nodes_comfy_cloud._normalize_dialogue("SPEAKER A :\nMeet at 10:30\nhttps://example.com\nB: Done") == (
"SPEAKER A: Meet at 10:30 https://example.com\nSPEAKER B: Done"
)
@pytest.mark.parametrize("script", ["SPEAKER A:", "just a continuation", " "])
def test_chatterbox_dialogue_rejects_blank_or_unattributed_text(monkeypatch, script):
upload = AsyncMock()
monkeypatch.setattr(nodes_comfy_cloud, "upload_audio_to_comfyapi", upload)
audio = {"waveform": torch.zeros(1, 1, 48000), "sample_rate": 48000}
with pytest.raises(Exception):
asyncio.run(nodes_comfy_cloud.ComfyCloudChatterboxDialogueNode.execute(script, audio, audio, 0.5, 0.5, 0.8, 0))
upload.assert_not_awaited()
def test_audio_duration_tolerates_one_sample_and_rejects_invalid_sample_rate():
one_sample_over = {"waveform": torch.zeros(1, 1, 48001), "sample_rate": 48000}
nodes_comfy_cloud._validate_audio_duration("Audio", one_sample_over, 0.5, 1)
with pytest.raises(ValueError, match="sample rate"):
nodes_comfy_cloud._validate_audio_duration("Audio", {"waveform": torch.zeros(1, 1, 1), "sample_rate": 0}, 0.5, 1)
with pytest.raises(ValueError, match="between"):
nodes_comfy_cloud._validate_audio_duration("Audio", {"waveform": torch.zeros(1, 1, 31), "sample_rate": 1}, 1, 30)
@pytest.mark.parametrize(("file_format", "expected_format"), [(".GLB", "glb"), ("SPZ", "spz")])
def test_download_cloud_3d_url_to_file_3d(monkeypatch, file_format, expected_format):
node = nodes_comfy_cloud.ComfyCloudTextToImageNode
downloaded = b"3d result"
calls = []
async def download(url, dest, **kwargs):
calls.append((url, dest, kwargs))
dest.write(downloaded)
dest.seek(0)
monkeypatch.setattr(download_helpers, "download_url_to_bytesio", download)
output = asyncio.run(
download_helpers.download_url_to_file_3d(
f"/proxy/comfy-cloud/results/task-1/model.{expected_format}",
file_format,
timeout=45,
max_retries=3,
cls=node,
)
)
assert output.format == expected_format
assert output.get_bytes() == downloaded
assert calls[0][0] == f"/proxy/comfy-cloud/results/task-1/model.{expected_format}"
assert isinstance(calls[0][1], BytesIO)
assert calls[0][2] == {"timeout": 45, "max_retries": 3, "cls": node, "allow_redirects": True}
THREE_D_POC_NODES = [
(nodes_comfy_cloud.ComfyCloudTripoSplatImageToGaussianSplatNode, "3d.triposplat-image-to-gaussian-splat.v1", ["image", "remove_background", "seed", "gaussian_count"], {"image": object(), "remove_background": False, "seed": 7, "gaussian_count": 32768}, "FILE_3D_SPZ", "spz"),
(nodes_comfy_cloud.ComfyCloudHunyuan3D21ImageTo3DNode, "3d.hunyuan3d-2-1-image-to-3d.v1", ["image", "seed"], {"image": object(), "seed": 8}, "FILE_3D_GLB", "glb"),
(nodes_comfy_cloud.ComfyCloudHunyuan3DMultiViewTo3DNode, "3d.hunyuan3d-multiview-to-3d.v1", ["front_image", "back_image", "seed"], {"front_image": object(), "back_image": object(), "seed": 9}, "FILE_3D_GLB", "glb"),
(nodes_comfy_cloud.ComfyCloudMoGe2PhotoToTexturedMeshNode, "3d.moge-2-photo-to-textured-mesh.v1", ["image", "fov_degrees", "detail", "mesh_decimation", "gap_threshold", "texture"], {"image": object(), "fov_degrees": 45.5, "detail": 8, "mesh_decimation": 2, "gap_threshold": 0.05, "texture": False}, "FILE_3D_GLB", "glb"),
(nodes_comfy_cloud.ComfyCloudMoGe2PanoramaTo3DSceneNode, "3d.moge-2-panorama-to-3d-scene.v1", ["panorama", "detail", "split_resolution", "merge_resolution", "mesh_decimation", "gap_threshold", "texture"], {"panorama": object(), "detail": 6, "split_resolution": 768, "merge_resolution": 2048, "mesh_decimation": 3, "gap_threshold": 0.06, "texture": False}, "FILE_3D_GLB", "glb"),
]
@pytest.mark.parametrize(("node", "workflow", "input_names", "arguments", "output_type", "file_format"), THREE_D_POC_NODES)
def test_3d_poc_node_schema_request_mapping_and_registration(monkeypatch, node, workflow, input_names, arguments, output_type, file_format):
run = AsyncMock(return_value=("3d-output",))
upload = AsyncMock(side_effect=["/uploads/front.png", "/uploads/back.png"])
monkeypatch.setattr(nodes_comfy_cloud, "_run_3d_workflow", run)
monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload)
monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: 1)
schema = node.define_schema()
assert schema.is_api_node
assert schema.category == "partner/3d/Comfy Cloud"
assert [input.id for input in schema.inputs] == input_names
assert schema.outputs[0].get_io_type() == output_type
assert workflow in get_args(ComfyCloudWorkflow)
assert node in asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list())
output = asyncio.run(node.execute(**arguments))
assert output[0] == "3d-output"
assert run.call_args.args[1] == workflow
assert run.call_args.args[3] == file_format
request_inputs = run.call_args.args[2].model_dump(exclude_none=True)
image_names = [name for name in ("image", "front_image", "back_image", "panorama") if name in arguments]
expected_inputs = {name: value for name, value in arguments.items() if name not in image_names}
expected_inputs["assets"] = {
name: {"type": "IMAGE", "url": f"/uploads/{'front' if index == 0 else 'back'}.png"}
for index, name in enumerate(image_names)
}
assert request_inputs == expected_inputs
assert '"id"' not in run.call_args.args[2].model_dump_json()
def test_3d_poc_schema_defaults_and_ranges():
schemas = {workflow: {input.id: input for input in node.define_schema().inputs} for node, workflow, _, _, _, _ in THREE_D_POC_NODES}
tripo = schemas["3d.triposplat-image-to-gaussian-splat.v1"]
assert tripo["remove_background"].default is True
assert (tripo["seed"].default, tripo["seed"].min, tripo["seed"].max) == (46, 0, 0xFFFFFFFFFFFFFFFF)
assert (tripo["gaussian_count"].default, tripo["gaussian_count"].min, tripo["gaussian_count"].max) == (262144, 32768, 262144)
assert "application/octet-stream" in nodes_comfy_cloud.ComfyCloudTripoSplatImageToGaussianSplatNode.define_schema().outputs[0].tooltip
assert schemas["3d.hunyuan3d-2-1-image-to-3d.v1"]["seed"].default == 952805179515179
assert schemas["3d.hunyuan3d-multiview-to-3d.v1"]["seed"].default == 502126049100058
photo = schemas["3d.moge-2-photo-to-textured-mesh.v1"]
assert (photo["fov_degrees"].default, photo["fov_degrees"].min, photo["fov_degrees"].max, photo["fov_degrees"].step) == (0, 0, 170, 0.1)
assert (photo["detail"].default, photo["detail"].min, photo["detail"].max) == (9, 0, 9)
panorama = schemas["3d.moge-2-panorama-to-3d-scene.v1"]
assert (panorama["split_resolution"].default, panorama["split_resolution"].min, panorama["split_resolution"].max) == (512, 256, 1024)
assert (panorama["merge_resolution"].default, panorama["merge_resolution"].min, panorama["merge_resolution"].max) == (1024, 256, 8192)
def test_hunyuan_multiview_validates_both_images_before_upload(monkeypatch):
upload = AsyncMock()
monkeypatch.setattr(nodes_comfy_cloud, "upload_image_to_comfyapi", upload)
monkeypatch.setattr(nodes_comfy_cloud, "get_number_of_images", lambda image: image)
with pytest.raises(ValueError, match="Exactly one front image and one back image"):
asyncio.run(nodes_comfy_cloud.ComfyCloudHunyuan3DMultiViewTo3DNode.execute(1, 2, 9))
upload.assert_not_awaited()
def test_extension_preserves_all_23_poc_node_registrations():
legacy_nodes = {
nodes_comfy_cloud.ComfyCloudTextToImageNode,
nodes_comfy_cloud.ComfyCloudTextToVideoNode,
nodes_comfy_cloud.ComfyCloudImageToVideoNode,
nodes_comfy_cloud.ComfyCloudImageEditNode,
}
registered = set(asyncio.run(nodes_comfy_cloud.ComfyCloudExtension().get_node_list()))
assert len(registered - legacy_nodes) == 23
def test_3d_workflow_submission_polling_cancel_and_download(monkeypatch):
sync = AsyncMock(return_value=ComfyCloudGenerateResponse(task_id="task-3d", status="queued", polling_url="/tasks/task-3d", cancel_url="/tasks/task-3d/cancel"))
poll = AsyncMock(return_value=ComfyCloudStatusResponse(task_id="task-3d", status="completed", output_url="/proxy/comfy-cloud/results/model.spz"))
download = AsyncMock(return_value="spz-output")
monkeypatch.setattr(nodes_comfy_cloud, "sync_op", sync)
monkeypatch.setattr(nodes_comfy_cloud, "poll_op", poll)
monkeypatch.setattr(nodes_comfy_cloud, "download_url_to_file_3d", download)
output = asyncio.run(nodes_comfy_cloud._run_3d_workflow(nodes_comfy_cloud.ComfyCloudTripoSplatImageToGaussianSplatNode, "3d.triposplat-image-to-gaussian-splat.v1", ComfyCloudWorkflowInputs(seed=46), "spz"))
request = sync.call_args.kwargs["data"]
assert request.workflow == "3d.triposplat-image-to-gaussian-splat.v1"
assert request.inputs.model_dump(exclude_none=True) == {"seed": 46}
assert poll.call_args.args[1].path == "/proxy/comfy-cloud/workflow/tasks/task-3d"
assert poll.call_args.kwargs["cancel_endpoint"].path == "/proxy/comfy-cloud/workflow/tasks/task-3d/cancel"
assert poll.call_args.kwargs["cancel_endpoint"].method == "POST"
download.assert_awaited_once_with(
"/proxy/comfy-cloud/results/model.spz",
"spz",
timeout=30 * 60,
cls=nodes_comfy_cloud.ComfyCloudTripoSplatImageToGaussianSplatNode,
allow_redirects=False,
)
assert output[0] == "spz-output"
Reference in New Issue View Git Blame Copy Permalink