feat: ltx sigma string parser with tests
This commit is contained in:
Jason Hoku
+171
@@ -0,0 +1,171 @@
|
||||
"""
|
||||
Root-level pytest configuration for standalone testing outside ComfyUI.
|
||||
|
||||
Installs ComfyUI stubs and a custom import hook so that the package's
|
||||
__init__.py (which pytest's Package.setup() always imports) loads cleanly
|
||||
without a live ComfyUI runtime.
|
||||
"""
|
||||
import sys
|
||||
import types
|
||||
import os
|
||||
import importlib
|
||||
import importlib.abc
|
||||
import importlib.machinery
|
||||
|
||||
|
||||
# Run immediately at conftest import time (before pytest_configure)
|
||||
_NODE_ROOT = os.path.dirname(os.path.abspath(__file__))
|
||||
_PKG_NAME = os.path.basename(_NODE_ROOT) # "ComfyUI-Ultimate-Auto-Sampler-Config-Grid-Testing-Suite"
|
||||
|
||||
|
||||
def _install_stubs():
|
||||
"""Install all stubs needed to import the package __init__.py cleanly."""
|
||||
|
||||
# 1. External deps
|
||||
for name in [
|
||||
"torch", "torchvision", "torchaudio",
|
||||
"folder_paths",
|
||||
"comfy", "comfy.utils", "comfy.sd", "comfy.model_management",
|
||||
"nodes", "aiohttp",
|
||||
]:
|
||||
if name not in sys.modules:
|
||||
sys.modules[name] = types.ModuleType(name)
|
||||
|
||||
# aiohttp.web
|
||||
if "aiohttp.web" not in sys.modules:
|
||||
web = types.ModuleType("aiohttp.web")
|
||||
web.Response = type("Response", (), {"__init__": lambda s, *a, **kw: None})
|
||||
web.json_response = lambda *a, **kw: None
|
||||
web.RouteTableDef = type("RouteTableDef", (), {
|
||||
"get": lambda s, *a, **kw: (lambda f: f),
|
||||
"post": lambda s, *a, **kw: (lambda f: f),
|
||||
"delete": lambda s, *a, **kw: (lambda f: f),
|
||||
"put": lambda s, *a, **kw: (lambda f: f),
|
||||
})
|
||||
sys.modules["aiohttp.web"] = web
|
||||
sys.modules["aiohttp"].web = web # type: ignore[attr-defined]
|
||||
|
||||
# folder_paths.get_output_directory
|
||||
fp = sys.modules["folder_paths"]
|
||||
if not hasattr(fp, "get_output_directory"):
|
||||
fp.get_output_directory = lambda: os.path.join(_NODE_ROOT, "_test_output") # type: ignore
|
||||
|
||||
# server.PromptServer
|
||||
if "server" not in sys.modules:
|
||||
sys.modules["server"] = types.ModuleType("server")
|
||||
srv = sys.modules["server"]
|
||||
if not hasattr(srv, "PromptServer"):
|
||||
_noop_deco = staticmethod(lambda *a, **kw: (lambda f: f))
|
||||
_routes = type("_Routes", (), {
|
||||
"get": _noop_deco, "post": _noop_deco,
|
||||
"delete": _noop_deco, "put": _noop_deco,
|
||||
})()
|
||||
_ps_instance = type("_PS", (), {"routes": _routes})()
|
||||
srv.PromptServer = type("PromptServer", (), {"instance": _ps_instance}) # type: ignore
|
||||
|
||||
# 2. Sub-module stubs (for relative imports in __init__.py)
|
||||
_subs = {
|
||||
"sampler_node": {"SamplerGridTester": type("SamplerGridTester", (), {})},
|
||||
"dashboard_node": {"SamplerConfigDashboardViewer": type("SamplerConfigDashboardViewer", (), {})},
|
||||
"html_generator": {"get_html_template": lambda *a, **kw: ""},
|
||||
"config_builder_node": {"UltimateConfigBuilder": type("UltimateConfigBuilder", (), {})},
|
||||
"json_text_node": {"SmartJSONTextNode": type("SmartJSONTextNode", (), {})},
|
||||
"metadata_packer": {"pack_metadata_into_image": lambda *a, **kw: None},
|
||||
"directory_scanner": {
|
||||
"scan_directory_for_images": lambda *a, **kw: ([], {}),
|
||||
"MEDIA_EXTENSIONS": set(),
|
||||
},
|
||||
"distribution_routes": {},
|
||||
"upscale_runner": {
|
||||
"start_upscale_job": lambda *a, **kw: (None, None),
|
||||
"get_upscale_status": lambda *a, **kw: {},
|
||||
"cancel_upscale_job": lambda *a, **kw: False,
|
||||
},
|
||||
}
|
||||
|
||||
for bare, attrs in _subs.items():
|
||||
fq = f"{_PKG_NAME}.{bare}"
|
||||
if bare not in sys.modules:
|
||||
stub = types.ModuleType(bare)
|
||||
sys.modules[bare] = stub
|
||||
if fq not in sys.modules:
|
||||
sys.modules[fq] = sys.modules[bare]
|
||||
for k, v in attrs.items():
|
||||
setattr(sys.modules[bare], k, v)
|
||||
setattr(sys.modules[fq], k, v)
|
||||
|
||||
# 3. Pre-register the package itself
|
||||
if _PKG_NAME not in sys.modules:
|
||||
pkg_stub = types.ModuleType(_PKG_NAME)
|
||||
pkg_stub.__path__ = [_NODE_ROOT]
|
||||
pkg_stub.__package__ = _PKG_NAME
|
||||
sys.modules[_PKG_NAME] = pkg_stub
|
||||
|
||||
# 4. Ensure node root is on sys.path
|
||||
if _NODE_ROOT not in sys.path:
|
||||
sys.path.insert(0, _NODE_ROOT)
|
||||
|
||||
|
||||
class _PackageInitLoader(importlib.abc.Loader):
|
||||
"""
|
||||
Custom loader for the package __init__.py.
|
||||
Sets __package__ correctly before executing the module so that relative
|
||||
imports (from .sampler_node import X) resolve via sys.modules.
|
||||
"""
|
||||
|
||||
def __init__(self, path):
|
||||
self._path = path
|
||||
|
||||
def create_module(self, spec):
|
||||
return None # use default semantics
|
||||
|
||||
def exec_module(self, module):
|
||||
module.__file__ = self._path
|
||||
module.__package__ = _PKG_NAME
|
||||
module.__name__ = _PKG_NAME
|
||||
module.__path__ = [_NODE_ROOT] # makes relative imports resolve parent pkg
|
||||
# Point sys.modules to this module object so relative imports find it
|
||||
sys.modules[_PKG_NAME] = module
|
||||
with open(self._path, "rb") as fh:
|
||||
src = fh.read()
|
||||
code = compile(src, self._path, "exec")
|
||||
exec(code, module.__dict__) # noqa: S102 – intentional dynamic exec for stub loading
|
||||
|
||||
|
||||
class _PackageInitFinder(importlib.abc.MetaPathFinder):
|
||||
"""
|
||||
Intercepts any attempt to import the package __init__.py and routes it
|
||||
through _PackageInitLoader so relative imports get the correct __package__.
|
||||
"""
|
||||
_INIT_PATH = os.path.join(_NODE_ROOT, "__init__.py")
|
||||
|
||||
def find_spec(self, fullname, path, target=None):
|
||||
# In importlib mode, pytest imports __init__.py as module name "__init__"
|
||||
# (module_name_from_path returns "__init__" for package root __init__.py).
|
||||
# We intercept that import and also the top-level package name import.
|
||||
if fullname in ("__init__", _PKG_NAME):
|
||||
if target is not None:
|
||||
# target is the path pytest wants to load; check it's our __init__.py
|
||||
try:
|
||||
if os.path.normcase(str(target.__spec__.origin)) != os.path.normcase(self._INIT_PATH):
|
||||
return None
|
||||
except Exception:
|
||||
pass
|
||||
spec = importlib.machinery.ModuleSpec(
|
||||
fullname,
|
||||
_PackageInitLoader(self._INIT_PATH),
|
||||
origin=self._INIT_PATH,
|
||||
)
|
||||
spec.submodule_search_locations = [_NODE_ROOT]
|
||||
return spec
|
||||
return None
|
||||
|
||||
|
||||
# Install stubs + hook immediately so they're ready before pytest does anything
|
||||
_install_stubs()
|
||||
sys.meta_path.insert(0, _PackageInitFinder())
|
||||
|
||||
|
||||
def pytest_configure(config):
|
||||
"""Re-ensure stubs are in place after config initialises."""
|
||||
_install_stubs()
|
||||
@@ -0,0 +1,70 @@
|
||||
"""
|
||||
LTX 2.3 Video Generation Module
|
||||
Two-stage SamplerCustomAdvanced pipeline with parallel audio rail.
|
||||
|
||||
Pinned LTX node pack version: TBD — set during first smoke test.
|
||||
Required nodes (looked up via nodes.NODE_CLASS_MAPPINGS):
|
||||
- DiffusionModelLoaderKJ
|
||||
- DualCLIPLoader (built-in)
|
||||
- VAELoaderKJ
|
||||
- LatentUpscaleModelLoader
|
||||
- LTXVPreprocess, LTXVImgToVideoInplace, LTXVConditioning,
|
||||
- LTXVCropGuides, LTXVConcatAVLatent, LTXVSeparateAVLatent,
|
||||
- LTXVEmptyLatentAudio, LTXVAudioVAEDecode,
|
||||
- EmptyLTXVLatentVideo, LTXVLatentUpsampler,
|
||||
- ManualSigmas, KSamplerSelect, RandomNoise, CFGGuider,
|
||||
- SamplerCustomAdvanced, VAEDecodeTiled, CreateVideo, SaveVideo
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import builtins
|
||||
import uuid
|
||||
from typing import List
|
||||
|
||||
|
||||
def safe_print(*args, **kwargs):
|
||||
"""Windows-safe print mirroring config_builder_node.safe_print."""
|
||||
try:
|
||||
builtins.print(*args, **kwargs)
|
||||
except (OSError, ValueError):
|
||||
try:
|
||||
msg = " ".join(str(a) for a in args) + kwargs.get("end", "\n")
|
||||
sys.__stdout__.write(msg)
|
||||
sys.__stdout__.flush()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
print = safe_print
|
||||
|
||||
|
||||
def parse_sigmas(sigma_str):
|
||||
"""Parse a comma-separated sigma string into a list of floats.
|
||||
|
||||
Args:
|
||||
sigma_str: e.g. "0.85, 0.7250, 0.4219, 0.0"
|
||||
|
||||
Returns:
|
||||
List of floats.
|
||||
|
||||
Raises:
|
||||
ValueError: empty string, fewer than 2 values, or any token not a valid float.
|
||||
"""
|
||||
if not sigma_str or not sigma_str.strip():
|
||||
raise ValueError("Sigma string is empty")
|
||||
|
||||
tokens = [t.strip() for t in sigma_str.split(",") if t.strip()]
|
||||
if len(tokens) < 2:
|
||||
raise ValueError(
|
||||
"Sigma string must contain at least 2 comma-separated values, got: " + repr(sigma_str)
|
||||
)
|
||||
|
||||
result = []
|
||||
for tok in tokens:
|
||||
try:
|
||||
result.append(float(tok))
|
||||
except ValueError:
|
||||
raise ValueError("Sigma token " + repr(tok) + " is not a valid float (in " + repr(sigma_str) + ")")
|
||||
return result
|
||||
@@ -0,0 +1,6 @@
|
||||
[pytest]
|
||||
testpaths = tests
|
||||
addopts = --import-mode=importlib
|
||||
pythonpath = .
|
||||
filterwarnings =
|
||||
ignore::DeprecationWarning
|
||||
@@ -0,0 +1,46 @@
|
||||
"""
|
||||
Pytest configuration for standalone testing outside ComfyUI.
|
||||
|
||||
The custom node's __init__.py requires ComfyUI internals (server, folder_paths,
|
||||
torch, etc.) that are not available in a plain Python environment.
|
||||
We stub those out here so pytest can collect and run unit tests without ComfyUI.
|
||||
"""
|
||||
import sys
|
||||
import types
|
||||
import os
|
||||
|
||||
# ── Stub out ComfyUI-only modules before any test or __init__.py import ──────
|
||||
_COMFY_STUBS = [
|
||||
"server",
|
||||
"folder_paths",
|
||||
"nodes",
|
||||
"torch",
|
||||
"aiohttp",
|
||||
"aiohttp.web",
|
||||
"comfy",
|
||||
"comfy.utils",
|
||||
"comfy.sd",
|
||||
"comfy.model_management",
|
||||
]
|
||||
|
||||
for _mod_name in _COMFY_STUBS:
|
||||
if _mod_name not in sys.modules:
|
||||
_stub = types.ModuleType(_mod_name)
|
||||
# aiohttp.web needs a minimal RouteTableDef stub used by distribution_routes
|
||||
if _mod_name == "aiohttp.web":
|
||||
class _RouteTableDef: # noqa: N801
|
||||
def get(self, *a, **kw):
|
||||
return lambda fn: fn
|
||||
def post(self, *a, **kw):
|
||||
return lambda fn: fn
|
||||
def delete(self, *a, **kw):
|
||||
return lambda fn: fn
|
||||
def put(self, *a, **kw):
|
||||
return lambda fn: fn
|
||||
_stub.RouteTableDef = _RouteTableDef
|
||||
sys.modules[_mod_name] = _stub
|
||||
|
||||
# Ensure the custom node root is on sys.path so `ltx_video_generation` is importable
|
||||
_NODE_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
if _NODE_ROOT not in sys.path:
|
||||
sys.path.insert(0, _NODE_ROOT)
|
||||
@@ -0,0 +1,32 @@
|
||||
"""Tests for LTX sigma string parsing."""
|
||||
import pytest
|
||||
import sys
|
||||
import os
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from ltx_video_generation import parse_sigmas
|
||||
|
||||
|
||||
def test_parse_valid_sigmas():
|
||||
result = parse_sigmas("1.0, 0.5, 0.0")
|
||||
assert result == [1.0, 0.5, 0.0]
|
||||
|
||||
|
||||
def test_parse_strips_whitespace():
|
||||
result = parse_sigmas(" 0.85, 0.7250,0.4219, 0.0 ")
|
||||
assert result == [0.85, 0.725, 0.4219, 0.0]
|
||||
|
||||
|
||||
def test_parse_single_value_raises():
|
||||
with pytest.raises(ValueError, match="at least 2"):
|
||||
parse_sigmas("1.0")
|
||||
|
||||
|
||||
def test_parse_empty_raises():
|
||||
with pytest.raises(ValueError, match="empty"):
|
||||
parse_sigmas("")
|
||||
|
||||
|
||||
def test_parse_malformed_raises():
|
||||
with pytest.raises(ValueError, match="not a valid float"):
|
||||
parse_sigmas("1.0, abc, 0.0")
|
||||
Reference in New Issue
Block a user