Merge branch 'master' into feature/custom-node-startup-errors

This commit is contained in:
Jedrzej Kosinski
2026-05-15 16:31:50 -07:00
committed by GitHub
301 changed files with 628628 additions and 2451 deletions

View File

@@ -52,7 +52,10 @@ def mock_provider(mock_releases):
@pytest.fixture(autouse=True)
def clear_cache():
import utils.install_util
import app.frontend_management
utils.install_util.PACKAGE_VERSIONS = {}
app.frontend_management.COMFY_PACKAGE_VERSIONS = []
def test_get_release(mock_provider, mock_releases):
@@ -147,7 +150,7 @@ def test_init_frontend_default_with_mocks():
# Act
with (
patch("app.frontend_management.check_frontend_version") as mock_check,
patch("app.frontend_management.check_comfy_packages_versions") as mock_check,
patch.object(
FrontendManager, "default_frontend_path", return_value="/mocked/path"
),
@@ -168,7 +171,7 @@ def test_init_frontend_fallback_on_error():
patch.object(
FrontendManager, "init_frontend_unsafe", side_effect=Exception("Test error")
),
patch("app.frontend_management.check_frontend_version") as mock_check,
patch("app.frontend_management.check_comfy_packages_versions") as mock_check,
patch.object(
FrontendManager, "default_frontend_path", return_value="/default/path"
),
@@ -277,7 +280,9 @@ def test_get_installed_templates_version():
def test_get_installed_templates_version_not_installed():
# Act
with patch("app.frontend_management.version", side_effect=Exception("Package not found")):
with patch(
"app.frontend_management.version", side_effect=Exception("Package not found")
):
version = FrontendManager.get_installed_templates_version()
# Assert

View File

@@ -0,0 +1,90 @@
"""Tests for NodeReplaceManager registration behavior."""
import importlib
import sys
import types
import pytest
@pytest.fixture
def NodeReplaceManager(monkeypatch):
"""Provide NodeReplaceManager with `nodes` stubbed.
`app.node_replace_manager` does `import nodes` at module level, which pulls in
torch + the full ComfyUI graph. register() doesn't actually need it, so we
stub `nodes` per-test (via monkeypatch so it's torn down) and reload the
module so it picks up the stub instead of any cached real import.
"""
fake_nodes = types.ModuleType("nodes")
fake_nodes.NODE_CLASS_MAPPINGS = {}
monkeypatch.setitem(sys.modules, "nodes", fake_nodes)
monkeypatch.delitem(sys.modules, "app.node_replace_manager", raising=False)
module = importlib.import_module("app.node_replace_manager")
yield module.NodeReplaceManager
# Drop the freshly-imported module so the next test (or a later real import
# of `nodes`) starts from a clean slate.
sys.modules.pop("app.node_replace_manager", None)
class FakeNodeReplace:
"""Lightweight stand-in for comfy_api.latest._io.NodeReplace."""
def __init__(self, new_node_id, old_node_id, old_widget_ids=None,
input_mapping=None, output_mapping=None):
self.new_node_id = new_node_id
self.old_node_id = old_node_id
self.old_widget_ids = old_widget_ids
self.input_mapping = input_mapping
self.output_mapping = output_mapping
def test_register_adds_replacement(NodeReplaceManager):
manager = NodeReplaceManager()
manager.register(FakeNodeReplace(new_node_id="NewNode", old_node_id="OldNode"))
assert manager.has_replacement("OldNode")
assert len(manager.get_replacement("OldNode")) == 1
def test_register_allows_multiple_alternatives_for_same_old_node(NodeReplaceManager):
"""Different new_node_ids for the same old_node_id should all be kept."""
manager = NodeReplaceManager()
manager.register(FakeNodeReplace(new_node_id="AltA", old_node_id="OldNode"))
manager.register(FakeNodeReplace(new_node_id="AltB", old_node_id="OldNode"))
replacements = manager.get_replacement("OldNode")
assert len(replacements) == 2
assert {r.new_node_id for r in replacements} == {"AltA", "AltB"}
def test_register_is_idempotent_for_duplicate_pair(NodeReplaceManager):
"""Re-registering the same (old_node_id, new_node_id) should be a no-op."""
manager = NodeReplaceManager()
manager.register(FakeNodeReplace(new_node_id="NewNode", old_node_id="OldNode"))
manager.register(FakeNodeReplace(new_node_id="NewNode", old_node_id="OldNode"))
manager.register(FakeNodeReplace(new_node_id="NewNode", old_node_id="OldNode"))
assert len(manager.get_replacement("OldNode")) == 1
def test_register_idempotent_preserves_first_registration(NodeReplaceManager):
"""First registration wins; later duplicates with different mappings are ignored."""
manager = NodeReplaceManager()
first = FakeNodeReplace(
new_node_id="NewNode", old_node_id="OldNode",
input_mapping=[{"new_id": "a", "old_id": "x"}],
)
second = FakeNodeReplace(
new_node_id="NewNode", old_node_id="OldNode",
input_mapping=[{"new_id": "b", "old_id": "y"}],
)
manager.register(first)
manager.register(second)
replacements = manager.get_replacement("OldNode")
assert len(replacements) == 1
assert replacements[0] is first
def test_register_dedupe_does_not_affect_other_old_nodes(NodeReplaceManager):
manager = NodeReplaceManager()
manager.register(FakeNodeReplace(new_node_id="NewA", old_node_id="OldA"))
manager.register(FakeNodeReplace(new_node_id="NewA", old_node_id="OldA"))
manager.register(FakeNodeReplace(new_node_id="NewB", old_node_id="OldB"))
assert len(manager.get_replacement("OldA")) == 1
assert len(manager.get_replacement("OldB")) == 1

View File

@@ -0,0 +1,81 @@
"""Tests for path_utils asset category resolution."""
import os
import tempfile
from pathlib import Path
from unittest.mock import patch
import pytest
from app.assets.services.path_utils import get_asset_category_and_relative_path
@pytest.fixture
def fake_dirs():
"""Create temporary input, output, and temp directories."""
with tempfile.TemporaryDirectory() as root:
root_path = Path(root)
input_dir = root_path / "input"
output_dir = root_path / "output"
temp_dir = root_path / "temp"
models_dir = root_path / "models" / "checkpoints"
for d in (input_dir, output_dir, temp_dir, models_dir):
d.mkdir(parents=True)
with patch("app.assets.services.path_utils.folder_paths") as mock_fp:
mock_fp.get_input_directory.return_value = str(input_dir)
mock_fp.get_output_directory.return_value = str(output_dir)
mock_fp.get_temp_directory.return_value = str(temp_dir)
with patch(
"app.assets.services.path_utils.get_comfy_models_folders",
return_value=[("checkpoints", [str(models_dir)])],
):
yield {
"input": input_dir,
"output": output_dir,
"temp": temp_dir,
"models": models_dir,
}
class TestGetAssetCategoryAndRelativePath:
def test_input_file(self, fake_dirs):
f = fake_dirs["input"] / "photo.png"
f.touch()
cat, rel = get_asset_category_and_relative_path(str(f))
assert cat == "input"
assert rel == "photo.png"
def test_output_file(self, fake_dirs):
f = fake_dirs["output"] / "result.png"
f.touch()
cat, rel = get_asset_category_and_relative_path(str(f))
assert cat == "output"
assert rel == "result.png"
def test_temp_file(self, fake_dirs):
"""Regression: temp files must be categorised, not raise ValueError."""
f = fake_dirs["temp"] / "GLSLShader_output_00004_.png"
f.touch()
cat, rel = get_asset_category_and_relative_path(str(f))
assert cat == "temp"
assert rel == "GLSLShader_output_00004_.png"
def test_temp_file_in_subfolder(self, fake_dirs):
sub = fake_dirs["temp"] / "sub"
sub.mkdir()
f = sub / "ComfyUI_temp_tczip_00004_.png"
f.touch()
cat, rel = get_asset_category_and_relative_path(str(f))
assert cat == "temp"
assert os.path.normpath(rel) == os.path.normpath("sub/ComfyUI_temp_tczip_00004_.png")
def test_model_file(self, fake_dirs):
f = fake_dirs["models"] / "model.safetensors"
f.touch()
cat, rel = get_asset_category_and_relative_path(str(f))
assert cat == "models"
def test_unknown_path_raises(self, fake_dirs):
with pytest.raises(ValueError, match="not within"):
get_asset_category_and_relative_path("/some/random/path.png")

View File

@@ -0,0 +1,78 @@
from comfy_api.latest._io import Combo, MultiCombo
def test_multicombo_serializes_multi_select_as_object():
multi_combo = MultiCombo.Input(
id="providers",
options=["a", "b", "c"],
default=["a"],
)
serialized = multi_combo.as_dict()
assert serialized["multiselect"] is True
assert "multi_select" in serialized
assert serialized["multi_select"] == {}
def test_multicombo_serializes_multi_select_with_placeholder_and_chip():
multi_combo = MultiCombo.Input(
id="providers",
options=["a", "b", "c"],
default=["a"],
placeholder="Select providers",
chip=True,
)
serialized = multi_combo.as_dict()
assert serialized["multiselect"] is True
assert serialized["multi_select"] == {
"placeholder": "Select providers",
"chip": True,
}
def test_combo_does_not_serialize_multiselect():
"""Regular Combo should not have multiselect in its serialized output."""
combo = Combo.Input(
id="choice",
options=["a", "b", "c"],
)
serialized = combo.as_dict()
# Combo sets multiselect=False, but prune_dict keeps False (not None),
# so it should be present but False
assert serialized.get("multiselect") is False
assert "multi_select" not in serialized
def _validate_combo_values(val, combo_options, is_multiselect):
"""Reproduce the validation logic from execution.py for testing."""
if is_multiselect and isinstance(val, list):
return [v for v in val if v not in combo_options]
else:
return [val] if val not in combo_options else []
def test_multicombo_validation_accepts_valid_list():
options = ["a", "b", "c"]
assert _validate_combo_values(["a", "b"], options, True) == []
def test_multicombo_validation_rejects_invalid_values():
options = ["a", "b", "c"]
assert _validate_combo_values(["a", "x"], options, True) == ["x"]
def test_multicombo_validation_accepts_empty_list():
options = ["a", "b", "c"]
assert _validate_combo_values([], options, True) == []
def test_combo_validation_rejects_list_even_with_valid_items():
"""A regular Combo should not accept a list value."""
options = ["a", "b", "c"]
invalid = _validate_combo_values(["a", "b"], options, False)
assert len(invalid) > 0

View File

@@ -124,9 +124,11 @@ class TestMathExpressionExecute:
with pytest.raises(Exception, match="not defined"):
self._exec("str(a)", a=42)
def test_boolean_result_raises(self):
with pytest.raises(ValueError, match="got bool"):
self._exec("a > b", a=5, b=3)
def test_boolean_result(self):
result = self._exec("a > b", a=5, b=3)
assert result[2] is True
result = self._exec("a > b", a=3, b=5)
assert result[2] is False
def test_empty_expression_raises(self):
with pytest.raises(ValueError, match="Expression cannot be empty"):

View File

@@ -90,6 +90,63 @@ class TestNumberConvertExecute:
assert result[0] == 1000.0
assert result[1] == 1000
# --- Large number precision (string input) ---
def test_string_large_int_above_2_53(self):
"""Text-to-int must not lose precision for integers beyond 2^53."""
big = 2**53 + 1 # 9007199254740993
result = self._exec(str(big))
assert result[1] == big
def test_string_large_negative_int_above_2_53(self):
big = -(2**53 + 1)
result = self._exec(str(big))
assert result[1] == big
def test_string_very_large_int(self):
big = 2**63 + 42
result = self._exec(str(big))
assert result[1] == big
def test_string_large_int_float_output_is_float(self):
"""FLOAT output is still a float (may lose precision, but must be float type)."""
result = self._exec(str(2**53 + 1))
assert isinstance(result[0], float)
# --- Large number precision (int input) ---
def test_int_large_above_2_53(self):
"""Native int input must preserve its value in the INT output."""
big = 2**53 + 1
result = self._exec(big)
assert result[1] == big
def test_int_large_negative_above_2_53(self):
big = -(2**53 + 1)
result = self._exec(big)
assert result[1] == big
def test_int_very_large(self):
big = 2**100
result = self._exec(big)
assert result[1] == big
# --- String decimal / scientific notation fallback ---
def test_string_decimal_still_truncates(self):
"""Strings with decimal points fall back to int(float(...)) truncation."""
result = self._exec("3.7")
assert result[1] == 3
def test_string_negative_decimal_truncates(self):
result = self._exec("-2.9")
assert result[1] == -2
def test_string_scientific_large(self):
result = self._exec("1e18")
assert result[0] == 1e18
assert result[1] == 10**18
# --- STRING error paths ---
def test_empty_string_raises(self):

View File

@@ -1,9 +1,23 @@
from collections import defaultdict
import torch
from comfy.model_detection import detect_unet_config, model_config_from_unet_config
import comfy.supported_models
def _freeze(value):
"""Recursively convert a value to a hashable form so configs can be
compared/used as dict keys or set members."""
if isinstance(value, dict):
return frozenset((k, _freeze(v)) for k, v in value.items())
if isinstance(value, (list, tuple)):
return tuple(_freeze(v) for v in value)
if isinstance(value, set):
return frozenset(_freeze(v) for v in value)
return value
def _make_longcat_comfyui_sd():
"""Minimal ComfyUI-format state dict for pre-converted LongCat-Image weights."""
sd = {}
@@ -110,3 +124,21 @@ class TestModelDetection:
model_config = model_config_from_unet_config(unet_config, sd)
assert model_config is not None
assert type(model_config).__name__ == "FluxSchnell"
def test_unet_config_and_required_keys_combination_is_unique(self):
"""Each model in the registry must have a unique combination of
``unet_config`` and ``required_keys``. If two models share the same
combination, ``BASE.matches`` cannot disambiguate between them and the
first one in the list will always win."""
models = comfy.supported_models.models
groups = defaultdict(list)
for model in models:
key = (_freeze(model.unet_config), _freeze(model.required_keys))
groups[key].append(model.__name__)
duplicates = {k: names for k, names in groups.items() if len(names) > 1}
assert not duplicates, (
"Found models sharing the same (unet_config, required_keys) "
"combination, which makes detection ambiguous: "
+ "; ".join(", ".join(names) for names in duplicates.values())
)

View File

@@ -0,0 +1,109 @@
"""Tests for comfy.deploy_environment."""
import os
import pytest
from comfy import deploy_environment
from comfy.deploy_environment import get_deploy_environment
@pytest.fixture(autouse=True)
def _reset_cache_and_install_dir(tmp_path, monkeypatch):
"""Reset the functools cache and point the ComfyUI install dir at a tmp dir for each test."""
get_deploy_environment.cache_clear()
monkeypatch.setattr(deploy_environment, "_COMFY_INSTALL_DIR", str(tmp_path))
yield
get_deploy_environment.cache_clear()
def _write_env_file(tmp_path, content: str) -> str:
"""Write the env file with exact content (no newline translation).
`newline=""` disables Python's text-mode newline translation so the bytes
on disk match the literal string passed in, regardless of host OS.
Newline-style tests (CRLF, lone CR) rely on this.
"""
path = os.path.join(str(tmp_path), ".comfy_environment")
with open(path, "w", encoding="utf-8", newline="") as f:
f.write(content)
return path
class TestGetDeployEnvironment:
def test_returns_local_git_when_file_missing(self):
assert get_deploy_environment() == "local-git"
def test_reads_value_from_file(self, tmp_path):
_write_env_file(tmp_path, "local-desktop2-standalone\n")
assert get_deploy_environment() == "local-desktop2-standalone"
def test_strips_trailing_whitespace_and_newline(self, tmp_path):
_write_env_file(tmp_path, " local-desktop2-standalone \n")
assert get_deploy_environment() == "local-desktop2-standalone"
def test_only_first_line_is_used(self, tmp_path):
_write_env_file(tmp_path, "first-line\nsecond-line\n")
assert get_deploy_environment() == "first-line"
def test_crlf_line_ending(self, tmp_path):
# Windows editors often save text files with CRLF line endings.
# The CR must not end up in the returned value.
_write_env_file(tmp_path, "local-desktop2-standalone\r\n")
assert get_deploy_environment() == "local-desktop2-standalone"
def test_crlf_multiline_only_first_line_used(self, tmp_path):
_write_env_file(tmp_path, "first-line\r\nsecond-line\r\n")
assert get_deploy_environment() == "first-line"
def test_crlf_with_surrounding_whitespace(self, tmp_path):
_write_env_file(tmp_path, " local-desktop2-standalone \r\n")
assert get_deploy_environment() == "local-desktop2-standalone"
def test_lone_cr_line_ending(self, tmp_path):
# Classic-Mac / some legacy editors use a bare CR.
# Universal-newlines decoding treats it as a line terminator too.
_write_env_file(tmp_path, "local-desktop2-standalone\r")
assert get_deploy_environment() == "local-desktop2-standalone"
def test_empty_file_falls_back_to_default(self, tmp_path):
_write_env_file(tmp_path, "")
assert get_deploy_environment() == "local-git"
def test_empty_after_whitespace_strip_falls_back_to_default(self, tmp_path):
_write_env_file(tmp_path, " \n")
assert get_deploy_environment() == "local-git"
def test_strips_control_chars_within_first_line(self, tmp_path):
# Embedded NUL/control chars in the value should be stripped
# (header-injection / smuggling protection).
_write_env_file(tmp_path, "abc\x00\x07xyz\n")
assert get_deploy_environment() == "abcxyz"
def test_strips_non_ascii_characters(self, tmp_path):
_write_env_file(tmp_path, "café-é\n")
assert get_deploy_environment() == "caf-"
def test_caps_read_at_128_bytes(self, tmp_path):
# A single huge line with no newline must not be fully read into memory.
huge = "x" * 10_000
_write_env_file(tmp_path, huge)
result = get_deploy_environment()
assert result == "x" * 128
def test_result_is_cached_across_calls(self, tmp_path):
path = _write_env_file(tmp_path, "first_value\n")
assert get_deploy_environment() == "first_value"
# Overwrite the file — cached value should still be returned.
with open(path, "w", encoding="utf-8") as f:
f.write("second_value\n")
assert get_deploy_environment() == "first_value"
def test_unreadable_file_falls_back_to_default(self, tmp_path, monkeypatch):
_write_env_file(tmp_path, "should_not_be_used\n")
def _boom(*args, **kwargs):
raise OSError("simulated read failure")
monkeypatch.setattr("builtins.open", _boom)
assert get_deploy_environment() == "local-git"

View File

@@ -1,10 +1,15 @@
"""Tests for feature flags functionality."""
import pytest
from comfy_api.feature_flags import (
get_connection_feature,
supports_feature,
get_server_features,
CLI_FEATURE_FLAG_REGISTRY,
SERVER_FEATURE_FLAGS,
_coerce_flag_value,
_parse_cli_feature_flags,
)
@@ -96,3 +101,83 @@ class TestFeatureFlags:
result = get_connection_feature(sockets_metadata, "sid1", "any_feature")
assert result is False
assert supports_feature(sockets_metadata, "sid1", "any_feature") is False
class TestCoerceFlagValue:
"""Test suite for _coerce_flag_value."""
def test_registered_bool_true(self):
assert _coerce_flag_value("show_signin_button", "true") is True
assert _coerce_flag_value("show_signin_button", "True") is True
def test_registered_bool_false(self):
assert _coerce_flag_value("show_signin_button", "false") is False
assert _coerce_flag_value("show_signin_button", "FALSE") is False
def test_unregistered_key_stays_string(self):
assert _coerce_flag_value("unknown_flag", "true") == "true"
assert _coerce_flag_value("unknown_flag", "42") == "42"
def test_bool_typo_raises(self):
"""Strict bool: typos like 'ture' or 'yes' must raise so the flag can be dropped."""
with pytest.raises(ValueError):
_coerce_flag_value("show_signin_button", "ture")
with pytest.raises(ValueError):
_coerce_flag_value("show_signin_button", "yes")
with pytest.raises(ValueError):
_coerce_flag_value("show_signin_button", "1")
with pytest.raises(ValueError):
_coerce_flag_value("show_signin_button", "")
def test_failed_int_coercion_raises(self, monkeypatch):
"""Malformed values for typed flags must raise; caller decides what to do."""
monkeypatch.setitem(
CLI_FEATURE_FLAG_REGISTRY,
"test_int_flag",
{"type": "int", "default": 0, "description": "test"},
)
with pytest.raises(ValueError):
_coerce_flag_value("test_int_flag", "not_a_number")
class TestParseCliFeatureFlags:
"""Test suite for _parse_cli_feature_flags."""
def test_single_flag(self, monkeypatch):
monkeypatch.setattr("comfy_api.feature_flags.args", type("Args", (), {"feature_flag": ["show_signin_button=true"]})())
result = _parse_cli_feature_flags()
assert result == {"show_signin_button": True}
def test_missing_equals_defaults_to_true(self, monkeypatch):
"""Bare flag without '=' is treated as the string 'true' (and coerced if registered)."""
monkeypatch.setattr("comfy_api.feature_flags.args", type("Args", (), {"feature_flag": ["show_signin_button", "valid=1"]})())
result = _parse_cli_feature_flags()
assert result == {"show_signin_button": True, "valid": "1"}
def test_empty_key_skipped(self, monkeypatch):
monkeypatch.setattr("comfy_api.feature_flags.args", type("Args", (), {"feature_flag": ["=value", "valid=1"]})())
result = _parse_cli_feature_flags()
assert result == {"valid": "1"}
def test_invalid_bool_value_dropped(self, monkeypatch, caplog):
"""A typo'd bool value must be dropped entirely, not silently set to False
and not stored as a raw string. A warning must be logged."""
monkeypatch.setattr(
"comfy_api.feature_flags.args",
type("Args", (), {"feature_flag": ["show_signin_button=ture", "valid=1"]})(),
)
with caplog.at_level("WARNING"):
result = _parse_cli_feature_flags()
assert result == {"valid": "1"}
assert "show_signin_button" not in result
assert any("show_signin_button" in r.message and "drop" in r.message.lower() for r in caplog.records)
class TestCliFeatureFlagRegistry:
"""Test suite for the CLI feature flag registry."""
def test_registry_entries_have_required_fields(self):
for key, info in CLI_FEATURE_FLAG_REGISTRY.items():
assert "type" in info, f"{key} missing 'type'"
assert "default" in info, f"{key} missing 'default'"
assert "description" in info, f"{key} missing 'description'"

View File

@@ -69,7 +69,11 @@ async def test_listuserdata_full_info(aiohttp_client, app, tmp_path):
assert len(result) == 1
assert result[0]["path"] == "file1.txt"
assert "size" in result[0]
assert "modified" in result[0]
assert isinstance(result[0]["modified"], int)
assert isinstance(result[0]["created"], int)
# Verify millisecond magnitude (timestamps after year 2000 in ms are > 946684800000)
assert result[0]["modified"] > 946684800000
assert result[0]["created"] > 946684800000
async def test_listuserdata_split_path(aiohttp_client, app, tmp_path):