scripts/test_merge_failure_reason.py

"""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)