"""Standalone verification for merge_video_track_with_ffmpeg failure_reason. Runs without ComfyUI/torch: stubs the heavy imports, then exercises every ``return None`` path of merge_video_track_with_ffmpeg and the compose_h3_project_video error propagation, asserting each produces a specific, human-readable reason. Usage: python scripts/test_merge_failure_reason.py """ from __future__ import annotations import sys import types from pathlib import Path PACKAGE_ROOT = Path(__file__).resolve().parent.parent def _install_stubs() -> None: # Minimal torch stub: only attributes evaluated at IMPORT time are needed # (annotations like `torch.Tensor`, `torch.device`, `torch.dtype` in # image.py, and `torch.nn.functional` imported by image/audio/multitrack). # No torch function actually runs on the paths under test. if "torch" not in sys.modules: torch = types.ModuleType("torch") torch.__path__ = [] # mark as package so torch.nn.functional imports resolve nn = types.ModuleType("torch.nn") functional = types.ModuleType("torch.nn.functional") functional.pad = lambda *a, **k: None nn.functional = functional torch.nn = nn torch.Tensor = object torch.device = object torch.dtype = object torch.float32 = object() torch.float64 = object() sys.modules["torch"] = torch sys.modules["torch.nn"] = nn sys.modules["torch.nn.functional"] = functional if "folder_paths" not in sys.modules: fp = types.ModuleType("folder_paths") temp_dir = PACKAGE_ROOT / ".tmp-temp" temp_dir.mkdir(exist_ok=True) fp.get_output_directory = lambda: str(PACKAGE_ROOT / ".tmp-output") fp.get_temp_directory = lambda: str(PACKAGE_ROOT / ".tmp-temp") fp.get_input_directory = lambda: "" fp.get_filename_list = lambda *a, **k: [] fp.get_full_path = lambda *a, **k: None sys.modules["folder_paths"] = fp if "server" not in sys.modules: sys.modules["server"] = types.ModuleType("server") _install_stubs() sys.path.insert(0, str(PACKAGE_ROOT)) # Import the REAL utils package: with the stubs above, the eager `from .image # import *` in utils/__init__.py resolves, so multitrack's `from . import # FFMPEG_RESIZE_METHODS, ...` also works and h3_project loads untouched. from utils.video import merge_video_track_with_ffmpeg # noqa: E402 import utils.video as video_mod # noqa: E402 import utils.h3_project as h3_project # noqa: E402 CHECKS: list[tuple[str, bool, str]] = [] def check(name: str, condition: bool, detail: str = "") -> None: CHECKS.append((name, bool(condition), detail)) # --------------------------------------------------------------------------- # 1. FFmpeg not found: point PATH lookup at an empty search space. # --------------------------------------------------------------------------- import shutil # noqa: E402 _real_which = shutil.which shutil.which = lambda name, **k: None try: reasons: list[str] = [] out = merge_video_track_with_ffmpeg( [{"source": "whatever.mp4", "start_frame": 0, "end_frame": 10}], 10, 24.0, 64, 64, failure_reason=reasons, ) check("ffmpeg-missing returns None", out is None) check( "ffmpeg-missing reason mentions FFmpeg executable", len(reasons) == 1 and "FFmpeg executable not found" in reasons[0], str(reasons), ) finally: shutil.which = _real_which # --------------------------------------------------------------------------- # 2. Without failure_reason the function still returns None silently and no # caller-visible behavior changes (backwards compatibility). # --------------------------------------------------------------------------- shutil.which = lambda name, **k: None try: out = merge_video_track_with_ffmpeg( [{"source": "whatever.mp4", "start_frame": 0, "end_frame": 10}], 10, 24.0, 64, 64, ) check("no-collector path still returns None", out is None) finally: shutil.which = _real_which # --------------------------------------------------------------------------- # 3. ffmpeg "found": redirect get_ffmpeg_path via monkeypatched shutil.which to # a real executable so we reach the precondition checks. # --------------------------------------------------------------------------- import os # noqa: E402 fake_ffmpeg = Path(PACKAGE_ROOT / ".tmp-fake-ffmpeg") fake_ffmpeg.write_text("") os.chmod(fake_ffmpeg, 0o755) shutil.which = lambda name, **k: str(fake_ffmpeg) if name == "ffmpeg" else None try: # Empty timeline reasons = [] out = merge_video_track_with_ffmpeg([], 0, 24.0, 64, 64, failure_reason=reasons) check("empty timeline returns None", out is None) check( "empty timeline reason", len(reasons) == 1 and "timeline is empty" in reasons[0], str(reasons), ) # Invalid frame rate reasons = [] out = merge_video_track_with_ffmpeg( [{"source": "x.mp4", "start_frame": 0, "end_frame": 1}], 1, 0.0, 64, 64, failure_reason=reasons, ) check("invalid fps returns None", out is None) check("invalid fps reason", len(reasons) == 1 and "frame rate" in reasons[0], str(reasons)) # Odd dimensions reasons = [] out = merge_video_track_with_ffmpeg( [{"source": "x.mp4", "start_frame": 0, "end_frame": 1}], 1, 24.0, 63, 64, failure_reason=reasons, ) check("odd width returns None", out is None) check( "odd width reason names size", len(reasons) == 1 and "63x64" in reasons[0] and "odd" in reasons[0], str(reasons), ) # Missing segment sources reasons = [] out = merge_video_track_with_ffmpeg( [ {"source": "/definitely/not/here/a.mp4", "start_frame": 0, "end_frame": 4}, {"source": "/definitely/not/here/b.mp4", "start_frame": 4, "end_frame": 8}, {"source": "/definitely/not/here/c.mp4", "start_frame": 8, "end_frame": 12}, ], 12, 24.0, 64, 64, failure_reason=reasons, ) check("missing sources returns None", out is None) check( "missing sources reason lists each file", len(reasons) == 1 and "a.mp4" in reasons[0] and "b.mp4" in reasons[0] and "c.mp4" in reasons[0] and "missing" in reasons[0], str(reasons), ) # Missing locked audio reasons = [] out = merge_video_track_with_ffmpeg( [ { "source": str(fake_ffmpeg), "audio_source": "/definitely/not/here/locked.wav", "start_frame": 0, "end_frame": 4, } ], 4, 24.0, 64, 64, failure_reason=reasons, ) check("missing locked audio returns None", out is None) check( "locked audio reason names file", len(reasons) == 1 and "locked.wav" in reasons[0], str(reasons), ) # FFmpeg command failure: run the real (invalid) fake binary -> OSError or # non-zero return. The fake file is not executable code; on POSIX exec # raises OSError (ENOEXEC / format error), on the sandbox it surfaces as # either the OSError path or a non-zero returncode path. Both must record. reasons = [] out = merge_video_track_with_ffmpeg( [{"source": str(fake_ffmpeg), "start_frame": 0, "end_frame": 1}], 1, 24.0, 64, 64, failure_reason=reasons, ) check("command failure returns None", out is None) check( "command failure records a reason", len(reasons) == 1 and ("failed to start" in reasons[0] or "exited with code" in reasons[0]), str(reasons), ) finally: shutil.which = _real_which fake_ffmpeg.unlink(missing_ok=True) # --------------------------------------------------------------------------- # 4. compose_h3_project_video propagates the reason into the RuntimeError. # --------------------------------------------------------------------------- def _fake_merge(segments, total_length, frame_rate, w, h, **kwargs): reasons = kwargs.get("failure_reason") if reasons is not None: reasons.append("FFmpeg exited with code 1: Unknown encoder 'libx264'") return None # compose_h3_project_video imports merge_video_track_with_ffmpeg from the # `utils.video` module at call time (function-local import), so patch the # attribute on the already-loaded video module, not on h3_project. h3_project.load_h3_project_data = lambda *a, **k: { "width": 848, "height": 480, "frame_rate": 24.0, "clips": [ { "index": 0, "enabled": True, "file_path": "easy_media/projects/p/out/seg_000.mp4", "updated_at": 1, "source_frame_count": 16, "source_start_frame": 0, "source_end_frame": 16, "video_files": [], } ], } video_mod.merge_video_track_with_ffmpeg = _fake_merge try: h3_project.compose_h3_project_video("proj", None) check("compose raises on None", False, "no exception raised") except RuntimeError as error: message = str(error) check( "compose error carries the real reason", "could not compose the H3 project video" in message and "Unknown encoder 'libx264'" in message, message, ) except Exception as error: # noqa: BLE001 check("compose raises RuntimeError only", False, repr(error)) # --------------------------------------------------------------------------- passed = sum(1 for _, ok, _ in CHECKS if ok) failed = len(CHECKS) - passed for name, ok, detail in CHECKS: print(f"{'PASS' if ok else 'FAIL'}: {name}" + (f" [{detail}]" if detail and not ok else "")) print(f"\n{passed}/{len(CHECKS)} checks passed") sys.exit(1 if failed else 0)