diff --git a/conftest.py b/conftest.py new file mode 100644 index 0000000..2d75b9d --- /dev/null +++ b/conftest.py @@ -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() diff --git a/ltx_video_generation.py b/ltx_video_generation.py new file mode 100644 index 0000000..351e159 --- /dev/null +++ b/ltx_video_generation.py @@ -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 diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..9b33eb7 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,6 @@ +[pytest] +testpaths = tests +addopts = --import-mode=importlib +pythonpath = . +filterwarnings = + ignore::DeprecationWarning diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..040373d --- /dev/null +++ b/tests/conftest.py @@ -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) diff --git a/tests/test_ltx_sigma_parser.py b/tests/test_ltx_sigma_parser.py new file mode 100644 index 0000000..935e824 --- /dev/null +++ b/tests/test_ltx_sigma_parser.py @@ -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")