import importlib.util from io import BytesIO import json import sys import types from fractions import Fraction from pathlib import Path import pytest import torch import torch.nn.functional as F class _Port: def __init__(self, name=None, **kwargs): self.name = name self.kwargs = kwargs class _PortType: @staticmethod def Input(name, **kwargs): return _Port(name, **kwargs) @staticmethod def Output(name=None, **kwargs): return _Port(name, **kwargs) class _DynamicCombo(_PortType): @staticmethod def Option(name, inputs): return name, inputs class _Autogrow: Type = dict class TemplatePrefix: def __init__(self, input, prefix, min=1, max=10): self.input = input self.prefix = prefix self.min = min self.max = max @staticmethod def Input(name, **kwargs): return _Port(name, **kwargs) class _NodeOutput: def __init__(self, *values, expand=None, **kwargs): self.values = values self.expand = expand self.kwargs = kwargs class _Schema: def __init__(self, **kwargs): self.__dict__.update(kwargs) class _NestedTensor: is_nested = True def __init__(self, values): self.values = tuple(values) self.tensors = self.values def unbind(self): return self.values def _h3_context_latent(value: int | float = 0, video_steps: int = 7): video = torch.full((1, 24, video_steps, 2, 2), float(value)) pixel_frames = sum((1, 4, 4, 4, 4)[index % 5] for index in range(video_steps)) audio_steps = round(pixel_frames * 5 / 3) audio = torch.full((1, 32, 2, audio_steps), float(value) + 100) return {"samples": _NestedTensor((video, audio))} def _empty_h3_context_latent(video_steps: int = 7): video = torch.zeros(1, 24, video_steps, 2, 2) pixel_frames = sum((1, 4, 4, 4, 4)[index % 5] for index in range(video_steps)) audio_steps = round(pixel_frames * 5 / 3) audio = torch.zeros(1, 32, 2, audio_steps) return {"samples": _NestedTensor((video, audio))} def _h3_conditioning_cache_dir(tmp_path): return tmp_path / "easy_media" / "h3_conditioning_cache" def _h3_video_anchor_latent(value: int | float = 0): return {"samples": torch.full((1, 24, 2, 2, 2), float(value))} class _ProgressBar: instances = [] def __init__(self, total): self.total = total self.updates = [] self.instances.append(self) def update_absolute(self, value, total=None, preview=None): self.updates.append((value, total)) class _Clip: def __init__(self): self.tokenize_calls = [] def tokenize(self, prompt, **kwargs): tokens = {"prompt": prompt, "kwargs": kwargs} self.tokenize_calls.append(tokens) return tokens def encode_from_tokens_scheduled(self, tokens): return [(torch.tensor([1.0]), {"tokens": tokens})] class _Vae: def __init__(self): self.encoded = [] def encode(self, value): self.encoded.append(value) temporal = 2 if value.shape[0] > 1 else 1 return torch.zeros(1, 24, temporal, 2, 2) class _AudioVae: audio_sample_rate = 32000 def __init__(self): self.encoded = [] def encode(self, value): self.encoded.append(value) return torch.zeros(1, 32, 2, 4) class _ImageResizeKJWithNvidia: @classmethod def INPUT_TYPES(cls): return { "required": { "image": ("IMAGE",), "width": ("INT", {"default": 512}), "height": ("INT", {"default": 512}), "upscale_method": (["nvidia_rtx_vsr", "lanczos"],), "keep_proportion": (["stretch", "resize"], {"default": "stretch"}), "pad_color": ("STRING", {"default": "0, 0, 0"}), "crop_position": (["center"], {"default": "center"}), "divisible_by": ("INT", {"default": 2}), } } class _MiniMaxLatentUpscaler: pass class _MiniMaxMotionContextTrim: pass class MiniMaxH3: unet_config = {"image_model": "minimax_h3"} class _MiniMaxH3Model: def __init__(self): self.model = types.SimpleNamespace(model_config=MiniMaxH3()) def _load_minimax_node(monkeypatch): io = types.SimpleNamespace( Audio=_PortType, AnyType=_PortType, Autogrow=_Autogrow, Boolean=_PortType, Clip=_PortType, Combo=_PortType, ComfyNode=object, Conditioning=_PortType, ControlAfterGenerate=types.SimpleNamespace(fixed="fixed"), DynamicCombo=_DynamicCombo, Hidden=types.SimpleNamespace(prompt="PROMPT", unique_id="UNIQUE_ID"), Custom=lambda **kwargs: _PortType, Float=_PortType, Guider=_PortType, Image=_PortType, Int=_PortType, Latent=_PortType, Model=_PortType, NodeOutput=_NodeOutput, Noise=_PortType, Sampler=_PortType, Schema=_Schema, Sigmas=_PortType, String=_PortType, Vae=_PortType, Video=_PortType, ) comfy_api = types.ModuleType("comfy_api") comfy_api_latest = types.ModuleType("comfy_api.latest") comfy_api_latest.io = io comfy_api_latest.InputImpl = types.SimpleNamespace( VideoFromFile=lambda path: types.SimpleNamespace(path=path) ) comfy_api_latest.Types = types.SimpleNamespace() comfy_api.latest = comfy_api_latest core_nodes = types.ModuleType("nodes") core_nodes.MAX_RESOLUTION = 16384 core_nodes.NODE_CLASS_MAPPINGS = {} comfy = types.ModuleType("comfy") comfy.__path__ = [] model_management = types.ModuleType("comfy.model_management") model_management.intermediate_device = lambda: torch.device("cpu") model_management.get_torch_device = lambda: torch.device("cpu") nested_tensor = types.ModuleType("comfy.nested_tensor") nested_tensor.NestedTensor = _NestedTensor comfy_utils = types.ModuleType("comfy.utils") comfy_utils.common_upscale = ( lambda samples, width, height, method, crop: F.interpolate( samples, size=(height, width), mode="nearest" ) ) _ProgressBar.instances.clear() comfy_utils.ProgressBar = _ProgressBar def save_torch_file(state, path, metadata=None): torch.save({"state": state, "metadata": metadata}, path) def load_torch_file( path, safe_load=False, device=None, return_metadata=False, ): del safe_load stored = torch.load( BytesIO(Path(path).read_bytes()), map_location=device or torch.device("cpu"), weights_only=True, ) if isinstance(stored, dict) and set(stored) == {"state", "metadata"}: state = stored["state"] metadata = stored["metadata"] else: state = stored metadata = None return (state, metadata) if return_metadata else state comfy_utils.save_torch_file = save_torch_file # The fixture writes torch archives; avoid suffix-based safetensors detection. comfy_utils.load_torch_file = load_torch_file comfy.model_management = model_management comfy.nested_tensor = nested_tensor comfy.utils = comfy_utils node_helpers = types.ModuleType("node_helpers") def conditioning_set_values(conditioning, values): return [(tensor, {**metadata, **values}) for tensor, metadata in conditioning] node_helpers.conditioning_set_values = conditioning_set_values torchaudio = types.ModuleType("torchaudio") torchaudio.functional = types.SimpleNamespace( resample=lambda waveform, source, target: waveform ) folder_paths = types.ModuleType("folder_paths") folder_paths.get_filename_list = lambda category: [] folder_paths.get_output_directory = lambda: "/tmp" folder_paths.get_temp_directory = lambda: "/tmp" server = types.ModuleType("server") server.PromptServer = types.SimpleNamespace( instance=types.SimpleNamespace(send_sync=lambda _event, _payload: None) ) package = types.ModuleType("easy_media") package.__path__ = [] nodes_package = types.ModuleType("easy_media.nodes") nodes_package.__path__ = [] basic_module = types.ModuleType("easy_media.nodes.basic") def prepare_multitrack_project_media(info): has_locked_audio = any( isinstance(track, dict) and track.get("audio_locked") is True for track in info.get("tracks", []) ) locked_audio = {"prepared_locked_audio": True} if has_locked_audio else None return info, [], [], [], locked_audio basic_module.prepare_multitrack_project_media = prepare_multitrack_project_media basic_module.crop_multitrack_project_media = ( lambda shared_audio, shared_video, locked_audio, *_args: ( shared_audio, shared_video, locked_audio, ) ) project_modules_package = types.ModuleType("easy_media.modules") project_modules_package.__path__ = [] motion_context_package = types.ModuleType("easy_media.modules.motion_context") motion_context_package.__path__ = [] utils_package = types.ModuleType("easy_media.utils") utils_package.__path__ = [str(Path(__file__).parents[1] / "utils")] utils_package.log_node_info = lambda *_args, **_kwargs: None log_spec = importlib.util.spec_from_file_location( "stage_log_under_test", Path(__file__).parents[1] / "utils" / "log.py" ) log_module = importlib.util.module_from_spec(log_spec) log_spec.loader.exec_module(log_module) utils_package.log_stage_time = log_module.log_stage_time utils_package.instrument_node_timing = log_module.instrument_node_timing utils_package.synchronize_execution_device = log_module.synchronize_execution_device utils_package.save_audio_to_temp_wav = lambda _audio: None utils_package.FFMPEG_RESIZE_METHODS = frozenset() for name in ( "audio_db_to_gain", "audio_volume_db", "equirectangular_to_perspective", "load_audio_waveform", "load_image_tensor", "multitrack_is_shared_reference", "multitrack_is_muted_image", "multitrack_media_identity", "multitrack_segments_in_window", "multitrack_slot_name", "resize_image", "resize_video_with_ffmpeg", "resolve_video_path", ): setattr(utils_package, name, lambda *_args, **_kwargs: None) utils_package.audio_is_muted = ( lambda settings: isinstance(settings, dict) and settings.get("muted") is True ) models_module = types.ModuleType("easy_media.utils.models") models_module.detect_turbo_model = lambda model: types.SimpleNamespace( is_turbo=False, as_dict=lambda: { "status": "unknown", "is_turbo": False, "source": "fallback", "evidence": "test detector", "patch_count": 0, } ) models_module.detect_turbo_lora_from_prompt = lambda prompt, node_id: None root = Path(__file__).parents[1] h3_presets_spec = importlib.util.spec_from_file_location( "easy_media.utils.h3_presets", root / "utils" / "h3_presets.py" ) assert h3_presets_spec is not None and h3_presets_spec.loader is not None h3_presets_module = importlib.util.module_from_spec(h3_presets_spec) h3_presets_spec.loader.exec_module(h3_presets_module) monkeypatch.setitem(sys.modules, "comfy_api", comfy_api) monkeypatch.setitem(sys.modules, "comfy_api.latest", comfy_api_latest) monkeypatch.setitem(sys.modules, "folder_paths", folder_paths) monkeypatch.setitem(sys.modules, "easy_media.utils", utils_package) audio_gain_module = types.ModuleType("easy_media.utils.audio_gain") audio_gain_module.audio_db_to_gain = utils_package.audio_db_to_gain audio_gain_module.audio_is_muted = utils_package.audio_is_muted audio_gain_module.audio_volume_db = utils_package.audio_volume_db panorama_module = types.ModuleType("easy_media.utils.panorama") panorama_module.equirectangular_to_perspective = ( utils_package.equirectangular_to_perspective ) monkeypatch.setitem( sys.modules, "easy_media.utils.audio_gain", audio_gain_module, ) monkeypatch.setitem( sys.modules, "easy_media.utils.panorama", panorama_module, ) multitrack_spec = importlib.util.spec_from_file_location( "easy_media.utils.multitrack", root / "utils" / "multitrack.py" ) assert multitrack_spec is not None and multitrack_spec.loader is not None multitrack_module = importlib.util.module_from_spec(multitrack_spec) monkeypatch.setitem( sys.modules, "easy_media.utils.multitrack", multitrack_module, ) multitrack_spec.loader.exec_module(multitrack_module) h3_project_spec = importlib.util.spec_from_file_location( "easy_media.utils.h3_project", root / "utils" / "h3_project.py" ) assert h3_project_spec is not None and h3_project_spec.loader is not None h3_project_module = importlib.util.module_from_spec(h3_project_spec) monkeypatch.setitem( sys.modules, "easy_media.utils.h3_project", h3_project_module, ) h3_project_spec.loader.exec_module(h3_project_module) h3_project_module.prepare_multitrack_project_media = ( lambda info: basic_module.prepare_multitrack_project_media(info) ) h3_project_module.crop_multitrack_project_media = ( lambda shared_audio, shared_video, locked_audio, *args: ( basic_module.crop_multitrack_project_media( shared_audio, shared_video, locked_audio, *args, ) ) ) utils_spec = importlib.util.spec_from_file_location( "easy_media.utils.minimax", root / "utils" / "minimax.py" ) assert utils_spec is not None and utils_spec.loader is not None utils_module = importlib.util.module_from_spec(utils_spec) utils_spec.loader.exec_module(utils_module) motion_context_spec = importlib.util.spec_from_file_location( "easy_media.modules.motion_context.core", root / "modules" / "motion_context" / "core.py", ) assert motion_context_spec is not None and motion_context_spec.loader is not None motion_context_module = importlib.util.module_from_spec(motion_context_spec) motion_context_spec.loader.exec_module(motion_context_module) drift_control_module = types.ModuleType( "easy_media.modules.motion_context.drift_control_av" ) drift_control_module.apply_context_swap_drift_control = ( lambda model, target_latent, context_latent, **kwargs: ( model, target_latent, int(kwargs.get("context_length", 22)), ) ) modules = { "comfy_api": comfy_api, "comfy_api.latest": comfy_api_latest, "nodes": core_nodes, "comfy": comfy, "comfy.model_management": model_management, "comfy.nested_tensor": nested_tensor, "comfy.utils": comfy_utils, "node_helpers": node_helpers, "torchaudio": torchaudio, "folder_paths": folder_paths, "server": server, "easy_media": package, "easy_media.nodes": nodes_package, "easy_media.nodes.basic": basic_module, "easy_media.modules": project_modules_package, "easy_media.modules.motion_context": motion_context_package, "easy_media.modules.motion_context.core": motion_context_module, "easy_media.modules.motion_context.drift_control_av": drift_control_module, "easy_media.utils": utils_package, "easy_media.utils.multitrack": multitrack_module, "easy_media.utils.h3_presets": h3_presets_module, "easy_media.utils.h3_project": h3_project_module, "easy_media.utils.models": models_module, "easy_media.utils.minimax": utils_module, } for name, module in modules.items(): monkeypatch.setitem(sys.modules, name, module) monkeypatch.delitem(sys.modules, "comfy_extras.nodes_minimax_h3", raising=False) # Other test modules install a reduced graph_utils stub globally. Reload the # real execution package so this module always sees GraphBuilder and # ExecutionBlocker regardless of test order. monkeypatch.delitem(sys.modules, "comfy_execution.graph_utils", raising=False) monkeypatch.delitem(sys.modules, "comfy_execution", raising=False) path = root / "nodes" / "minimax.py" try: spec = importlib.util.spec_from_file_location("easy_media.nodes.minimax", path) if spec is None or spec.loader is None: return None module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) project_spec = importlib.util.spec_from_file_location( "easy_media.nodes.project", root / "nodes" / "project.py", ) if project_spec is None or project_spec.loader is None: return None project_module = importlib.util.module_from_spec(project_spec) sys.modules[project_spec.name] = project_module project_spec.loader.exec_module(project_module) monkeypatch.setattr(project_module, "install_project_memory_cleanup", lambda: None) module.EasyMultiTrackProject = project_module.EasyMultiTrackProject module.EasyMultiTrackProjectVideoCombine = ( project_module.EasyMultiTrackProjectVideoCombine ) module._project_module = project_module return module except (FileNotFoundError, ImportError): raise def _load_moxie_combine(monkeypatch): """Load moxie_combine.py into the stub environment set up above.""" root = Path(__file__).parents[1] spec = importlib.util.spec_from_file_location( "easy_media.moxie_combine", root / "moxie_combine.py" ) assert spec is not None and spec.loader is not None module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module def _image_values(*values): return torch.tensor(values, dtype=torch.float32).reshape(len(values), 1, 1, 1) def _base_inputs(**overrides): inputs = { "clip": [_Clip()], "vae": [_Vae()], "audio_vae": [], "images": [], "videos": [], "audios": [], "prompt": ["prompt"], "mode": ["multi_frames"], "width": [32], "height": [32], "length": [5], "ref_image_size": ["match"], } inputs.update(overrides) return inputs def _h3_project_inputs(**overrides): inputs = { "model_loader": [{ "model": _MiniMaxH3Model(), "clip": object(), "vae": object(), "audio_vae": object(), }], "tracks_info": [{ "width": 1344, "height": 768, "frame_rate": 24, "format": "MiniMax", "tracks": [{ "type": "task", "segments": [{ "start_frame": 0, "end_frame": 120, "content": { "task_mode": "default", "continuity_mode": "shot", "images": [], "user_prompt": "a cinematic scene", }, }], }], }], } inputs.update(overrides) return inputs def _h3_sampling_mode(mode, **children): return [{"sampling_mode": [mode], **children}] def _graph_node(output, class_type): return next( node for node in output.expand.values() if node["class_type"] == class_type ) def test_module_loads_without_native_minimax_nodes(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None def test_h3_conditioning_cache_round_trips_and_skips_lazy_inputs( monkeypatch, tmp_path, ): module = _load_minimax_node(monkeypatch) schema_inputs = { item.name: item for item in module.EasyH3ConditioningCache.define_schema().inputs } assert schema_inputs["conditioning"].kwargs["lazy"] is True assert schema_inputs["latent"].kwargs["lazy"] is True monkeypatch.setattr( module.folder_paths, "get_temp_directory", lambda: str(tmp_path), ) cache_logs = [] monkeypatch.setattr( module, "log_node_info", lambda name, message: cache_logs.append((name, message)), ) clip = _Clip() vae = _Vae() audio_vae = _AudioVae() model = object() conditioning = [( torch.arange(6, dtype=torch.float32).reshape(1, 2, 3), { "pooled_output": torch.ones(1, 3), "minimax_token_tags": torch.tensor([0, 1], dtype=torch.int64), "minimax_refs": [{ "kind": "video_audio", "latent_t": 2, "latent": torch.full((1, 24, 2, 2, 2), 3.0), "audio_latent": torch.full((1, 32, 2, 4), 4.0), }], }, )] latent = _empty_h3_context_latent(video_steps=2) first_status = { "_easy_media_cache_status": { "project_media": "first-load", "segment_media": "first-load", "task_output": "first-load", } } assert module.EasyH3ConditioningCache.check_lazy_status( project_name="demo", segment_index=0, tracks_info=first_status, task_output_ready="prompt", model=model, clip=clip, vae=vae, audio_vae=audio_vae, ) == ["conditioning", "latent"] first = module.EasyH3ConditioningCache.execute( project_name="demo", segment_index=0, tracks_info=first_status, task_output_ready="prompt", model=model, clip=clip, vae=vae, audio_vae=audio_vae, conditioning=conditioning, latent=latent, ) cache_path = _h3_conditioning_cache_dir(tmp_path) / "conditioning_0.safetensors" assert cache_path.is_file() assert first.values == (conditioning, latent) stored = torch.load( BytesIO(cache_path.read_bytes()), map_location=torch.device("cpu"), weights_only=True, ) assert not any(name.startswith("latent.") for name in stored["state"]) cache_metadata = json.loads( stored["metadata"]["easy_media_h3_conditioning_cache"] ) assert cache_metadata["schema_version"] == "3" assert cache_metadata["scope_token"] assert "file(" in cache_logs[-1][1] assert "=conditioning(" in cache_logs[-1][1] assert "+reference-video(" in cache_logs[-1][1] assert "+reference-audio(" in cache_logs[-1][1] assert "initial latent" not in cache_logs[-1][1] hit_status = { "_easy_media_cache_status": { "project_media": "cache-hit", "segment_media": "cache-hit", "task_output": "cache-hit", } } original_load = module.load_h3_conditioning_cache load_calls = [] def tracked_load(*args, **kwargs): load_calls.append(args) return original_load(*args, **kwargs) monkeypatch.setattr(module, "load_h3_conditioning_cache", tracked_load) required = module.EasyH3ConditioningCache.check_lazy_status( project_name="demo", segment_index=0, tracks_info=hit_status, task_output_ready="prompt", model=model, clip=clip, vae=vae, audio_vae=audio_vae, ) assert required == [] restored = module.EasyH3ConditioningCache.execute( project_name="demo", segment_index=0, tracks_info=hit_status, task_output_ready="prompt", model=model, clip=clip, vae=vae, audio_vae=audio_vae, ) assert len(load_calls) == 1 restored_conditioning, restored_latent = restored.values assert torch.equal(restored_conditioning[0][0], conditioning[0][0]) assert torch.equal( restored_conditioning[0][1]["minimax_refs"][0]["audio_latent"], conditioning[0][1]["minimax_refs"][0]["audio_latent"], ) assert restored_latent["samples"].is_nested is True assert all( torch.equal(actual, expected) for actual, expected in zip( restored_latent["samples"].unbind(), latent["samples"].unbind(), ) ) def test_h3_conditioning_cache_reencodes_when_model_changes( monkeypatch, tmp_path, ): module = _load_minimax_node(monkeypatch) monkeypatch.setattr( module.folder_paths, "get_temp_directory", lambda: str(tmp_path), ) clip = _Clip() vae = _Vae() audio_vae = _AudioVae() model = object() hit_status = { "_easy_media_cache_status": { "project_media": "cache-hit", "segment_media": "cache-hit", "task_output": "cache-hit", } } module.EasyH3ConditioningCache.execute( project_name="demo", segment_index=1, tracks_info=hit_status, task_output_ready="prompt", model=model, clip=clip, vae=vae, audio_vae=audio_vae, conditioning=[(torch.ones(1), {})], latent=_empty_h3_context_latent(), ) replacement_model = object() required = module.EasyH3ConditioningCache.check_lazy_status( project_name="demo", segment_index=1, tracks_info=hit_status, task_output_ready="prompt", model=replacement_model, clip=clip, vae=vae, audio_vae=audio_vae, ) assert required == ["conditioning", "latent"] module.EasyH3ConditioningCache.execute( project_name="demo", segment_index=1, tracks_info=hit_status, task_output_ready="prompt", model=replacement_model, clip=clip, vae=vae, audio_vae=audio_vae, conditioning=[(torch.full((1,), 2.0), {})], latent=_empty_h3_context_latent(), ) cache_dir = _h3_conditioning_cache_dir(tmp_path) assert [path.name for path in cache_dir.glob("conditioning_1*.safetensors")] == [ "conditioning_1.safetensors" ] assert module.EasyH3ConditioningCache.check_lazy_status( project_name="demo", segment_index=1, tracks_info=hit_status, task_output_ready="prompt", model=replacement_model, clip=clip, vae=vae, audio_vae=audio_vae, ) == [] def test_h3_conditioning_cache_refuses_nonzero_initial_latent( monkeypatch, tmp_path, ): module = _load_minimax_node(monkeypatch) cache_path = tmp_path / "conditioning_0.safetensors" with pytest.raises(ValueError, match="must be zero-filled"): module.save_h3_conditioning_cache( [(torch.ones(1), {})], _h3_context_latent(1), cache_path, "encoder", "scope", ) assert not cache_path.exists() def test_h3_conditioning_cache_invalidates_temp_pool_once_per_queue( monkeypatch, tmp_path, ): module = _load_minimax_node(monkeypatch) monkeypatch.setattr( module.folder_paths, "get_temp_directory", lambda: str(tmp_path), ) clip = _Clip() vae = _Vae() audio_vae = _AudioVae() model = object() miss_status = { "_easy_media_cache_status": { "project_media": "first-load", "segment_media": "first-load", "task_output": "first-load", } } for segment_index in (0, 1): module.EasyH3ConditioningCache.execute( project_name="demo", segment_index=segment_index, tracks_info=miss_status, task_output_ready="prompt", model=model, clip=clip, vae=vae, audio_vae=audio_vae, conditioning=[(torch.tensor([segment_index]), {})], latent=_empty_h3_context_latent(), ) assert sorted( path.name for path in _h3_conditioning_cache_dir(tmp_path).glob("*.safetensors") ) == ["conditioning_0.safetensors", "conditioning_1.safetensors"] def test_h3_conditioning_cache_switching_project_resets_temp_pool( monkeypatch, tmp_path, ): module = _load_minimax_node(monkeypatch) monkeypatch.setattr( module.folder_paths, "get_temp_directory", lambda: str(tmp_path), ) clip = _Clip() vae = _Vae() audio_vae = _AudioVae() model = object() hit_status = { "_easy_media_cache_status": { "project_media": "cache-hit", "segment_media": "cache-hit", "task_output": "cache-hit", } } for project_name, segment_index in (("first", 0), ("second", 1)): module.EasyH3ConditioningCache.execute( project_name=project_name, segment_index=segment_index, tracks_info=hit_status, task_output_ready="prompt", model=model, clip=clip, vae=vae, audio_vae=audio_vae, conditioning=[(torch.tensor([segment_index]), {})], latent=_empty_h3_context_latent(), ) assert [ path.name for path in _h3_conditioning_cache_dir(tmp_path).glob("*.safetensors") ] == ["conditioning_1.safetensors"] def test_h3_conditioning_cache_keeps_only_five_recent_segments( monkeypatch, tmp_path, ): module = _load_minimax_node(monkeypatch) monkeypatch.setattr( module.folder_paths, "get_temp_directory", lambda: str(tmp_path), ) clip = _Clip() vae = _Vae() audio_vae = _AudioVae() model = object() hit_status = { "_easy_media_cache_status": { "project_media": "cache-hit", "segment_media": "cache-hit", "task_output": "cache-hit", } } for segment_index in range(6): module.EasyH3ConditioningCache.execute( project_name="demo", segment_index=segment_index, tracks_info=hit_status, task_output_ready="prompt", model=model, clip=clip, vae=vae, audio_vae=audio_vae, conditioning=[(torch.tensor([segment_index]), {})], latent=_empty_h3_context_latent(), ) cache_files = sorted( path.name for path in _h3_conditioning_cache_dir(tmp_path).glob("*.safetensors") ) assert len(cache_files) == 5 assert "conditioning_0.safetensors" not in cache_files def test_h3_conditioning_cache_unavailable_pool_falls_back_to_encoding( monkeypatch, tmp_path, ): module = _load_minimax_node(monkeypatch) monkeypatch.setattr( module.folder_paths, "get_temp_directory", lambda: str(tmp_path), ) monkeypatch.setattr( module, "prepare_h3_conditioning_cache_pool", lambda *_args, **_kwargs: None, ) clip = _Clip() vae = _Vae() audio_vae = _AudioVae() model = object() hit_status = { "_easy_media_cache_status": { "project_media": "cache-hit", "segment_media": "cache-hit", "task_output": "cache-hit", } } required = module.EasyH3ConditioningCache.check_lazy_status( project_name="demo", segment_index=0, tracks_info=hit_status, task_output_ready="prompt", model=model, clip=clip, vae=vae, audio_vae=audio_vae, ) assert required == ["conditioning", "latent"] conditioning = [(torch.ones(1), {})] latent = _empty_h3_context_latent() result = module.EasyH3ConditioningCache.execute( project_name="demo", segment_index=0, tracks_info=hit_status, task_output_ready="prompt", model=model, clip=clip, vae=vae, audio_vae=audio_vae, conditioning=conditioning, latent=latent, ) assert result.values == (conditioning, latent) assert not _h3_conditioning_cache_dir(tmp_path).exists() def test_h3_conditioning_cache_scope_rejects_stale_file_when_clear_fails( monkeypatch, tmp_path, ): module = _load_minimax_node(monkeypatch) monkeypatch.setattr( module.folder_paths, "get_temp_directory", lambda: str(tmp_path), ) clip = _Clip() vae = _Vae() audio_vae = _AudioVae() model = object() hit_status = { "_easy_media_cache_status": { "project_media": "cache-hit", "segment_media": "cache-hit", "task_output": "cache-hit", } } module.EasyH3ConditioningCache.execute( project_name="first", segment_index=0, tracks_info=hit_status, task_output_ready="prompt", model=model, clip=clip, vae=vae, audio_vae=audio_vae, conditioning=[(torch.ones(1), {})], latent=_empty_h3_context_latent(), ) cache_path = _h3_conditioning_cache_dir(tmp_path) / "conditioning_0.safetensors" assert cache_path.is_file() cache_module = sys.modules["easy_media.utils.h3_conditioning_cache"] monkeypatch.setattr( cache_module, "_clear_h3_conditioning_cache_files", lambda _cache_dir: None, ) required = module.EasyH3ConditioningCache.check_lazy_status( project_name="second", segment_index=0, tracks_info=hit_status, task_output_ready="prompt", model=model, clip=clip, vae=vae, audio_vae=audio_vae, ) assert cache_path.is_file() assert required == ["conditioning", "latent"] def test_h3_conditioning_cache_corrupt_body_falls_back_in_same_queue( monkeypatch, tmp_path, ): module = _load_minimax_node(monkeypatch) monkeypatch.setattr( module.folder_paths, "get_temp_directory", lambda: str(tmp_path), ) clip = _Clip() vae = _Vae() audio_vae = _AudioVae() model = object() hit_status = { "_easy_media_cache_status": { "project_media": "cache-hit", "segment_media": "cache-hit", "task_output": "cache-hit", } } module.EasyH3ConditioningCache.execute( project_name="demo", segment_index=0, tracks_info=hit_status, task_output_ready="prompt", model=model, clip=clip, vae=vae, audio_vae=audio_vae, conditioning=[(torch.ones(1), {})], latent=_empty_h3_context_latent(), ) cache_path = _h3_conditioning_cache_dir(tmp_path) / "conditioning_0.safetensors" monkeypatch.setattr( module, "load_h3_conditioning_cache", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("corrupt body")), ) required = module.EasyH3ConditioningCache.check_lazy_status( project_name="demo", segment_index=0, tracks_info=hit_status, task_output_ready="prompt", model=model, clip=clip, vae=vae, audio_vae=audio_vae, ) assert required == ["conditioning", "latent"] assert not cache_path.exists() def test_fallback_conditioning_nodes_use_native_node_ids(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None assert ( module.MiniMaxH3ImageToVideoFallback.define_schema().node_id == "MiniMaxH3ImageToVideo" ) assert ( module.MiniMaxH3ReferenceToVideoFallback.define_schema().node_id == "MiniMaxH3ReferenceToVideo" ) def test_missing_native_conditioning_nodes_are_selected_for_registration(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None assert module.get_minimax_h3_fallback_nodes() == [ module.MiniMaxH3ImageToVideoFallback, module.MiniMaxH3ReferenceToVideoFallback, ] def test_reference_mode_routes_to_native_node_when_available(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None module.comfy_nodes.NODE_CLASS_MAPPINGS.update( { "MiniMaxH3ImageToVideo": object, "MiniMaxH3ReferenceToVideo": object, } ) output = module.EasyMiniMaxH3ToVideo.execute( **_base_inputs(mode=["reference"], images=[_image_values(1)]) ) assert module.get_minimax_h3_fallback_nodes() == [] assert _graph_node(output, module.REFERENCE_BRIDGE_NODE_ID) def test_missing_native_reference_registers_fallback_and_routes_through_bridge( monkeypatch, ): module = _load_minimax_node(monkeypatch) assert module is not None module.comfy_nodes.NODE_CLASS_MAPPINGS["MiniMaxH3ImageToVideo"] = object output = module.EasyMiniMaxH3ToVideo.execute( **_base_inputs(mode=["reference"], images=[_image_values(1)]) ) assert module.get_minimax_h3_fallback_nodes() == [ module.MiniMaxH3ReferenceToVideoFallback, ] assert _graph_node(output, module.REFERENCE_BRIDGE_NODE_ID) def test_schema_exposes_list_media_inputs_without_image_position(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None schema = module.EasyMiniMaxH3ToVideo.define_schema() inputs = {port.name: port for port in schema.inputs} assert schema.node_id == "easy minimaxH3ToVideo" assert schema.is_input_list is True assert schema.enable_expand is True assert list(inputs) == [ "clip", "vae", "audio_vae", "images", "audios", "videos", "prompt", "mode", "width", "height", "length", "locked_video_timing_frames", "ref_image_size", ] assert inputs["audio_vae"].kwargs["optional"] is True assert inputs["mode"].kwargs["options"] == [ "reference", "multi_frames", "last_frame", ] assert [output.name for output in schema.outputs] == ["positive", "latent"] def test_multitrack_h3_project_schema_exposes_pipeline_configuration(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None schema = module.EasyMultiTrackProject.define_schema() inputs = {port.name: port for port in schema.inputs} assert schema.node_id == "easy multitrackProject" assert schema.is_input_list is True assert schema.enable_expand is True assert not getattr(schema, "is_output_node", False) assert schema.not_idempotent is True assert schema.hidden == ["PROMPT", "UNIQUE_ID"] assert list(inputs) == [ "tracks_info", "model_loader", "model_loader_2nd", "sampler", "sampler_2nd", "sigmas", "sigmas_2nd", "project_name", "project_save", "segment_start_number", "segment_count", "seed", "sampling_plan", "sampling_mode", "1st_pass_only", "disable_2nd_noise", "upscale_by", "upscale_model", "enabled_tiling", ] for name in ( "sampler", "sigmas", ): assert inputs[name].kwargs["optional"] is True assert inputs[name].kwargs["raw_link"] is True assert inputs[name].kwargs["lazy"] is True assert inputs["model_loader"].kwargs["raw_link"] is True assert inputs["model_loader"].kwargs["lazy"] is True assert inputs["project_name"].kwargs["default"] == "" assert inputs["project_save"].kwargs["options"] == ["new", "override"] assert inputs["project_save"].kwargs["default"] == "override" sampling_plan_options = inputs["sampling_plan"].kwargs["options"] assert sampling_plan_options == [ "custom", "ultra_light", "light", "medium", "high", ] assert json.loads(json.dumps(sampling_plan_options)) == sampling_plan_options assert inputs["sampling_plan"].kwargs["default"] == "light" sampling_mode_options = inputs["sampling_mode"].kwargs["options"] assert [name for name, _ in sampling_mode_options] == [ "single", "dual", "selflift", "passthrough", ] assert sampling_mode_options[0][1] == [] assert sampling_mode_options[1][1] == [] assert sampling_mode_options[3][1] == [] selflift_inputs = {port.name: port for port in sampling_mode_options[2][1]} assert list(selflift_inputs) == [ "transition_ratio", "lowres_scale", ] assert selflift_inputs["transition_ratio"].kwargs["default"] == 0.6 assert selflift_inputs["lowres_scale"].kwargs["default"] == 0.6 for name in ("sampler_2nd", "sigmas_2nd", "model_loader_2nd"): assert inputs[name].kwargs["optional"] is True assert inputs[name].kwargs["raw_link"] is True assert inputs[name].kwargs["lazy"] is True assert inputs["1st_pass_only"].kwargs["default"] is False assert inputs["disable_2nd_noise"].kwargs["default"] is False assert inputs["upscale_by"].kwargs["default"] == 1.250 assert inputs["upscale_by"].kwargs["step"] == 0.001 assert inputs["upscale_by"].kwargs["round"] == 0.001 assert inputs["upscale_by"].kwargs["extra_dict"] == {"precision": 3} assert inputs["segment_start_number"].kwargs["default"] == 1 assert inputs["segment_start_number"].kwargs["min"] == 1 assert inputs["segment_count"].kwargs["default"] == -1 assert [output.name for output in schema.outputs] == [ "PROJECT_NAME", "LOCKED_AUDIO", ] def test_h3_project_passthrough_skips_conditioning_and_keeps_context(monkeypatch): module = _load_minimax_node(monkeypatch) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs(sampling_mode=_h3_sampling_mode("passthrough")) ) nodes = list(result.expand.values()) types = {node["class_type"] for node in nodes} assert "easy h3PassthroughVideo" in types assert "easy minimaxH3ToVideo" not in types assert "easy h3ConditioningCache" not in types assert "SamplerCustomAdvanced" not in types assert "easy h3SegmentSamplingStart" not in types passthrough = _graph_node(result, "easy h3PassthroughVideo") assert passthrough["inputs"]["frame_count"] == 120 assert passthrough["inputs"]["fps"] == 24.0 artifact = _graph_node(result, "easy h3ProjectArtifact") context = result.expand[artifact["inputs"]["context_latent"][0]] assert context["class_type"] == "easy h3MotionContextLatentTrim" assert context["inputs"]["context_length"] == "22" def test_h3_project_static_prepare_exposes_media_cache_boundary(monkeypatch): module = _load_minimax_node(monkeypatch) static_schema = module._project_module.EasyH3ProjectStaticPrepare.define_schema() assert static_schema.node_id == "easy h3ProjectStaticPrepare" assert static_schema.enable_expand is True def test_multitrack_h3_project_keeps_model_and_media_as_prepare_links(monkeypatch): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs(model_loader=[["loader", 0]]) module.EasyMultiTrackProject.hidden = types.SimpleNamespace( prompt={ "project-node": { "inputs": { "tracks_info": ["multitrack-info", 0], "model_loader": ["loader", 0], }, }, }, unique_id="project-node", ) result = module.EasyMultiTrackProject.execute(**inputs) media_id, media_node = next( (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy h3ProjectStaticPrepare" and node_id.endswith("project_media_prepare") ) model_id, model_node = next( (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy h3ProjectStaticPrepare" and node_id.endswith("project_model_prepare") ) segment_id, segment_node = next( (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy h3ProjectStaticPrepare" and node_id.endswith("segment_static_prepare_0") ) task_id, task = next( (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy multiTrackTaskOutput" ) conditioning = _graph_node(result, "easy minimaxH3ToVideo") assert model_node["inputs"]["model_loader"] == ["loader", 0] assert "tracks_info" not in model_node["inputs"] assert media_node["inputs"]["tracks_info"] == ["multitrack-info", 0] assert "model_loader" not in media_node["inputs"] assert segment_node["inputs"]["project_static"] == [media_id, 0] assert task["inputs"]["tracks_info"] == [segment_id, 1] assert conditioning["inputs"]["clip"] == [model_id, 4] assert conditioning["inputs"]["images"] == [task_id, 4] assert "seed" not in model_node["inputs"] assert "seed" not in media_node["inputs"] assert "seed" not in segment_node["inputs"] assert "seed" not in task["inputs"] assert "seed" not in conditioning["inputs"] def test_linked_selflift_keeps_segment_order_after_cached_media(monkeypatch): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs( model_loader=[["loader", 0]], sampling_mode=_h3_sampling_mode("selflift"), ) first = inputs["tracks_info"][0]["tracks"][0]["segments"][0] inputs["tracks_info"][0]["tracks"][0]["segments"].append({ **first, "start_frame": 120, "end_frame": 240, "content": {**first["content"]}, }) result = module.EasyMultiTrackProject.execute(**inputs) artifact_id = next( node_id for node_id, node in result.expand.items() if node["class_type"] == "easy h3ProjectArtifact" and node["inputs"]["segment_index"] == 0 ) second_selflift = next( node for node_id, node in result.expand.items() if node["class_type"] == "easy minimaxH3SelfLiftSampler" and node_id.endswith("selflift_sample_1") ) second_segment_prepare = next( node for node_id, node in result.expand.items() if node["class_type"] == "easy h3ProjectStaticPrepare" and node_id.endswith("segment_static_prepare_1") ) assert second_selflift["inputs"]["previous"] == [artifact_id, 0] assert second_segment_prepare["inputs"]["previous"] == [artifact_id, 0] def test_h3_project_static_prepare_materializes_once_then_crops(monkeypatch): module = _load_minimax_node(monkeypatch) project_module = module._project_module loader = { "model": _MiniMaxH3Model(), "clip": "clip", "vae": "vae", "audio_vae": "audio-vae", } prepared = [] shared_image = _image_values(1) shared_audio = {"waveform": torch.ones(1, 1, 1), "sample_rate": 1} shared_video = object() locked_audio = {"waveform": torch.ones(1, 1, 1), "sample_rate": 1} def prepare(info): prepared.append(info) return info, [shared_image], [shared_audio], [shared_video], locked_audio monkeypatch.setattr(project_module, "prepare_multitrack_project_media", prepare) monkeypatch.setattr( project_module, "crop_multitrack_project_media", lambda audio, video, locked, *_args: (audio, video, locked), ) tracks_info = _h3_project_inputs()["tracks_info"] global_result = project_module.EasyH3ProjectStaticPrepare.execute( model_loader=loader, tracks_info=tracks_info, sampling_plan="custom", sampler="sampler", sigmas="sigmas", ) segment_result = project_module.EasyH3ProjectStaticPrepare.execute( project_static=global_result.values[0], task_start_frame=0, task_duration_frames=120, generation_mode="reference", ) restored_global = project_module.EasyH3ProjectStaticPrepare.execute( tracks_info=tracks_info, ) restored_segment = project_module.EasyH3ProjectStaticPrepare.execute( project_static=restored_global.values[0], task_start_frame=0, task_duration_frames=120, generation_mode="reference", ) def assert_no_container_cycle(value, ancestors=None): if not isinstance(value, (dict, list, tuple)): return ancestors = set() if ancestors is None else ancestors assert id(value) not in ancestors nested_ancestors = ancestors | {id(value)} children = value.values() if isinstance(value, dict) else value for child in children: assert_no_container_cycle(child, nested_ancestors) assert len(prepared) == 1 assert global_result.values[2:7] == ( loader["model"], loader["model"], "clip", "vae", "audio-vae", ) assert segment_result.values[1]["_preloaded_media"] == { "images": [shared_image], "audio": [shared_audio], "video": [shared_video], } assert segment_result.values[8] is locked_audio assert restored_global.values[0]["shared_images"] is global_result.values[0]["shared_images"] assert restored_global.values[0]["shared_audio"] is global_result.values[0]["shared_audio"] assert restored_global.values[0]["shared_video"] is global_result.values[0]["shared_video"] assert restored_segment.values[1] is segment_result.values[1] assert restored_segment.values[8] is locked_audio assert_no_container_cycle(tracks_info) def test_multitrack_h3_project_loads_segment_media_from_tracks_info(monkeypatch): module = _load_minimax_node(monkeypatch) result = module.EasyMultiTrackProject.execute(**_h3_project_inputs()) task_node = next( node for node in result.expand.values() if node["class_type"] == "easy multiTrackTaskOutput" ) assert set(task_node["inputs"]) == { "tracks_info", "task_index", "prompt_format", } @pytest.mark.parametrize( "sampling_mode,first_only", [ ("single", False), ("dual", False), ("dual", True), ("selflift", False), ], ) def test_project_memory_boundaries_follow_artifact_saves(monkeypatch, sampling_mode, first_only): module = _load_minimax_node(monkeypatch) installed = [] monkeypatch.setattr(module._project_module, "install_project_memory_cleanup", lambda: installed.append(True)) result = module.EasyMultiTrackProject.execute(**_h3_project_inputs( sampling_mode=_h3_sampling_mode(sampling_mode, **{"1st_pass_only": [first_only]}), upscale_by=[1.0], )) assert installed == [True] tagged = [node for node in result.expand.values() if "easy_media_segment" in node.get("_meta", {})] assert tagged assert all(node["_meta"]["easy_media_segment"] == 0 for node in tagged) boundaries = [node for node in tagged if node["_meta"].get("easy_media_segment_saved")] assert len(boundaries) == 1 assert boundaries[0]["class_type"] == "easy h3ProjectArtifact" assert "video_path" in boundaries[0]["inputs"] assert "context_latent" in boundaries[0]["inputs"] assert all( "easy_media_segment" not in node.get("_meta", {}) for node in result.expand.values() if node["class_type"] in { "KSamplerSelect", "ManualSigmas", "easy h3ProjectStaticPrepare", "easy multiTrackTaskOutput", } ) conditioning = _graph_node(result, "easy minimaxH3ToVideo") assert conditioning["_meta"]["easy_media_segment"] == 0 def test_multitrack_h3_project_prepends_shared_media_before_h3_conditioning( monkeypatch, ): module = _load_minimax_node(monkeypatch) shared_image = _image_values(1) shared_audio = {"waveform": torch.ones(1, 1, 1), "sample_rate": 1} shared_video = object() inputs = _h3_project_inputs() inputs["tracks_info"][0]["tracks"][0]["segments"][0]["content"][ "task_mode" ] = "ref" monkeypatch.setattr( module._project_module, "prepare_multitrack_project_media", lambda info: ( info, [shared_image], [shared_audio], [shared_video], None, ), ) monkeypatch.setattr( module._project_module, "crop_multitrack_project_media", lambda shared_audios, shared_videos, locked_audio, *_args: ( shared_audios, shared_videos, locked_audio, ), ) result = module.EasyMultiTrackProject.execute(**inputs) conditioning = _graph_node(result, "easy minimaxH3ToVideo") task_node_id, task_node = next( (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy multiTrackTaskOutput" ) preloaded_media = task_node["inputs"]["tracks_info"]["_preloaded_media"] assert preloaded_media["images"][0] is shared_image assert preloaded_media["audio"][0] is shared_audio assert preloaded_media["video"][0] is shared_video assert conditioning["inputs"]["images"] == [task_node_id, 4] assert conditioning["inputs"]["audios"] == [task_node_id, 5] assert conditioning["inputs"]["videos"] == [task_node_id, 6] assert not any(name.startswith("shared_") for name in conditioning["inputs"]) assert not any( node["class_type"] == "easy h3ProjectMediaPreprocessor" for node in result.expand.values() ) def test_multitrack_h3_minus_one_defers_each_task_media_until_its_loop(monkeypatch): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs(segment_count=[-1]) inputs["tracks_info"][0]["tracks"][0]["segments"].append({ "start_frame": 120, "end_frame": 240, "content": {"user_prompt": "second", "task_mode": "default"}, }) result = module.EasyMultiTrackProject.execute(**inputs) task_nodes = sorted( ( (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy multiTrackTaskOutput" ), key=lambda item: item[1]["inputs"]["task_index"], ) first_artifact_id = next( node_id for node_id, node in result.expand.items() if node["class_type"] == "easy h3ProjectArtifact" and node["inputs"]["segment_index"] == 0 ) assert "previous" not in task_nodes[0][1]["inputs"] assert task_nodes[1][1]["inputs"]["previous"] == [first_artifact_id, 0] def test_multitrack_h3_project_outputs_locked_audio_used_by_generation(monkeypatch): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs() inputs["tracks_info"][0]["tracks"].append({ "type": "audio", "audio_locked": True, "segments": [{ "start_frame": 0, "end_frame": 120, "content": {"media_type": "audio"}, }], }) result = module.EasyMultiTrackProject.execute(**inputs) audio_lock = _graph_node(result, "easy minimaxH3AudioLock") align = _graph_node(result, "easy h3LockedAudioDurationAlign") saved_video = _graph_node(result, "easy saveVideo") assert result.values[1] == {"prepared_locked_audio": True} assert audio_lock["inputs"]["audio"] == {"prepared_locked_audio": True} artifact = _graph_node(result, "easy h3ProjectArtifact") assert "locked_audio" not in artifact["inputs"] assert align["inputs"]["fps"] == 24.0 selector = _graph_node(result, "easy h3LockedAudioSelect") assert selector["inputs"]["locked_audio"] == {"prepared_locked_audio": True} assert saved_video["inputs"]["input_mode.audio"][0].endswith( "locked_audio_select_0" ) assert not any( node["class_type"] == "easy multiTrackTaskOutput" and node["inputs"]["task_index"] == -1 for node in result.expand.values() ) def test_multitrack_h3_project_outputs_none_without_locked_audio(monkeypatch): module = _load_minimax_node(monkeypatch) result = module.EasyMultiTrackProject.execute(**_h3_project_inputs()) assert result.values[1] is None assert not any( node["class_type"] == "easy h3LockedAudioDurationAlign" for node in result.expand.values() ) saved_video = _graph_node(result, "easy saveVideo") audio_link = saved_video["inputs"]["input_mode.audio"] assert result.expand[audio_link[0]]["class_type"] == "VAEDecodeAudio" def test_muted_locked_video_preserves_timing_without_audio_lock(monkeypatch): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs() info = inputs["tracks_info"][0] info["tracks"].append({ "type": "video", "audio_locked": True, "muted": True, "segments": [{ "start_frame": 0, "end_frame": 120, "content": {"media_type": "video"}, }], }) result = module.EasyMultiTrackProject.execute(**inputs) assert not any( node["class_type"] == "easy minimaxH3AudioLock" for node in result.expand.values() ) trim = _graph_node(result, "easy h3ContextMediaTrim") assert trim["inputs"]["output_frames"] == 120 assert trim["inputs"]["fit_video_duration"] is True assert trim["inputs"]["pad_audio"] is True saved_video = _graph_node(result, "easy saveVideo") audio_link = saved_video["inputs"]["input_mode.audio"] assert result.expand[audio_link[0]]["class_type"] == "easy h3ContextMediaTrim" def test_locked_audio_select_falls_back_when_video_has_no_audio(monkeypatch): module = _load_minimax_node(monkeypatch) generated = {"waveform": torch.ones(1, 1, 4), "sample_rate": 24} missing = module.EasyH3LockedAudioSelect.execute(generated, None) malformed = module.EasyH3LockedAudioSelect.execute(generated, {}) assert missing.values[0] is generated assert malformed.values[0] is generated def test_multitrack_h3_project_does_not_log_execution_events_while_expanding( monkeypatch, ): module = _load_minimax_node(monkeypatch) messages = [] monkeypatch.setattr( module._project_module, "log_node_info", lambda node_name, message: messages.append((node_name, message)), ) module.EasyMultiTrackProject.execute(**_h3_project_inputs()) assert messages == [ ("MultiTrack Project", "Found 1 segments; processing 1"), ] def test_h3_segment_execution_markers_log_the_current_segment(monkeypatch): module = _load_minimax_node(monkeypatch) messages = [] notifications = [] monkeypatch.setattr( module, "log_node_info", lambda node_name, message: messages.append((node_name, message)), ) server = types.ModuleType("server") server.PromptServer = types.SimpleNamespace( instance=types.SimpleNamespace( send_sync=lambda event, payload: notifications.append((event, payload)) ) ) monkeypatch.setitem(sys.modules, "server", server) latent = {"samples": torch.zeros(1)} sampling_output = module.EasyH3SegmentSamplingStart.execute( object(), object(), object(), object(), latent, "demo", 3, "first" ) save_end_output = module.EasyH3SegmentSaveEnd.execute("video.mp4", "demo", 3) assert sampling_output.values[-1] is latent assert save_end_output.values == ("video.mp4",) assert notifications == [ ( "easy_multitrack_project_refresh", { "project_name": "demo", "phase": "before", "segment_index": 3, "sampling_pass": "first", }, ), ] assert len(messages) == 1 assert messages[0][0] == "MultiTrack Project" assert messages[0][1].startswith("Sampling segment 3 (first):") assert "sampler_name=None" in messages[0][1] def test_h3_context_media_trim_removes_head_and_grid_tail_together(monkeypatch): module = _load_minimax_node(monkeypatch) images = torch.arange(40, dtype=torch.float32).reshape(40, 1, 1, 1) audio = { "waveform": torch.arange(80, dtype=torch.float32).reshape(1, 1, 80), "sample_rate": 48, } result = module.EasyH3ContextMediaTrim.execute( images, audio, trim_frames=5, output_frames=22, fps=24.0, ) output_images, output_audio = result.values assert output_images[:, 0, 0, 0].tolist() == list(range(5, 27)) assert output_audio["waveform"].flatten().tolist() == list(range(10, 54)) assert output_audio["sample_rate"] == 48 def test_h3_context_media_trim_can_leave_locked_audio_short_for_alignment( monkeypatch, ): module = _load_minimax_node(monkeypatch) images = torch.zeros(5, 1, 1, 3) audio = { "waveform": torch.arange(8, dtype=torch.float32).reshape(1, 1, 8), "sample_rate": 2, } result = module.EasyH3ContextMediaTrim.execute( images, audio, trim_frames=0, output_frames=5, pad_audio=False, fps=1.0, ) assert result.values[0].shape[0] == 5 assert result.values[1]["waveform"].shape[-1] == 8 @pytest.mark.parametrize( ("total_frames", "expected_start", "expected_padding"), [ (22, 0, 0), (120, 102, 4), (124, 102, 0), (153, 136, 5), (168, 153, 7), (240, 221, 3), ], ) def test_h3_context_media_trim_preserves_vae_chunk_phase_for_encoding( monkeypatch, total_frames, expected_start, expected_padding, ): module = _load_minimax_node(monkeypatch) images = torch.arange(total_frames, dtype=torch.float32).reshape( total_frames, 1, 1, 1 ) audio = { "waveform": torch.zeros(1, 1, total_frames * 2), "sample_rate": 48, } result = module.EasyH3ContextMediaTrim.execute( images, audio, trim_frames=0, output_frames=22, phase_align_video_encode=True, ) output_images, output_audio = result.values expected = list(range(expected_start, total_frames)) expected.extend([total_frames - 1] * expected_padding) assert output_images[:, 0, 0, 0].tolist() == expected assert len(output_images) == 22 assert output_images[-1].item() == images[-1].item() assert output_audio is audio def test_h3_context_media_trim_rejects_empty_phase_aligned_video(monkeypatch): module = _load_minimax_node(monkeypatch) audio = {"waveform": torch.zeros(1, 1, 2), "sample_rate": 48} with pytest.raises(ValueError, match="at least one frame"): module.EasyH3ContextMediaTrim.execute( torch.empty(0, 1, 1, 1), audio, output_frames=22, phase_align_video_encode=True, ) def test_h3_context_trim_embeds_phase_aligned_five_frame_anchor(monkeypatch): module = _load_minimax_node(monkeypatch) images = torch.arange(9 * 2 * 2 * 3, dtype=torch.float32).reshape(9, 2, 2, 3) vae = _Vae() trimmed = {"samples": "trimmed"} monkeypatch.setattr( module, "trim_motion_context_latent", lambda _latent, context_length: trimmed.copy(), ) output = module.EasyH3MotionContextLatentTrim.execute( {"samples": "full"}, "22", anchor_images=images, vae=vae, ).values[0] assert len(vae.encoded) == 1 expected = images[-1:].expand(5, *images.shape[1:]) assert torch.equal(vae.encoded[0], expected) assert output["samples"] == "trimmed" assert output["anchor_samples"].shape[2] == 2 def test_h3_locked_audio_duration_align_matches_decoded_video_without_padding( monkeypatch, ): module = _load_minimax_node(monkeypatch) images = torch.zeros(209, 1, 1, 3) waveform = torch.linspace(-1.0, 1.0, 417_600).reshape(1, 1, -1) result = module.EasyH3LockedAudioDurationAlign.execute( images, {"waveform": waveform, "sample_rate": 48_000}, fps=24.0, ) aligned = result.values[0] assert aligned["sample_rate"] == 48_000 assert aligned["waveform"].shape == (1, 1, 418_000) assert aligned["waveform"][..., 0].item() == pytest.approx(-1.0, abs=1e-4) assert aligned["waveform"][..., -1].item() == pytest.approx(1.0, abs=1e-4) def test_h3_locked_audio_duration_align_rejects_more_than_one_audio_latent_hop( monkeypatch, ): module = _load_minimax_node(monkeypatch) images = torch.zeros(209, 1, 1, 3) waveform = torch.zeros(1, 1, 416_000) with pytest.raises(ValueError, match="mismatch is too large"): module.EasyH3LockedAudioDurationAlign.execute( images, {"waveform": waveform, "sample_rate": 48_000}, fps=24.0, ) def test_multitrack_h3_project_waits_for_previous_artifact_before_sampling( monkeypatch, ): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs() first_segment = inputs["tracks_info"][0]["tracks"][0]["segments"][0] inputs["tracks_info"][0]["tracks"][0]["segments"].append({ **first_segment, "start_frame": 121, "end_frame": 241, "content": {**first_segment["content"]}, }) result = module.EasyMultiTrackProject.execute(**inputs) sampling_starts = sorted( ( node for node in result.expand.values() if node["class_type"] == "easy h3SegmentSamplingStart" ), key=lambda node: node["inputs"]["segment_index"], ) artifacts = sorted( ( node for node in result.expand.values() if node["class_type"] == "easy h3ProjectArtifact" ), key=lambda node: node["inputs"]["segment_index"], ) assert "previous" not in sampling_starts[0]["inputs"] assert sampling_starts[1]["inputs"]["previous"] == artifacts[1]["inputs"][ "previous" ] def test_h3_project_artifact_schema(monkeypatch): module = _load_minimax_node(monkeypatch) artifact_schema = module.EasyH3ProjectArtifact.define_schema() artifact_inputs = {port.name: port for port in artifact_schema.inputs} assert list(artifact_inputs) == [ "project_name", "project_save", "segment_index", "context_latent", "context_latent_low", "video_path", "tracks_info", "continuity_mode", "sampling_pass", "seed", "previous", "audio", ] assert artifact_inputs["context_latent_low"].kwargs["optional"] is True assert artifact_inputs["seed"].kwargs["optional"] is True assert artifact_inputs["project_save"].kwargs["options"] == [ "new", "override", ] assert artifact_inputs["project_save"].kwargs["default"] == "new" def test_multitrack_h3_project_render_schema_uses_project_data_widget(monkeypatch): module = _load_minimax_node(monkeypatch) schema = module.EasyMultiTrackProjectVideoCombine.define_schema() inputs = {port.name: port for port in schema.inputs} assert schema.node_id == "easy multitrackProjectVideoCombine" assert not getattr(schema, "is_output_node", False) assert inputs["project_name"].kwargs["force_input"] is True assert "project_data" in inputs assert schema.hidden == ["UNIQUE_ID"] assert [output.name for output in schema.outputs] == [ "VIDEO", "FILENAME_PREFIX", ] def test_multitrack_h3_project_render_returns_video_and_filename_prefix(monkeypatch): module = _load_minimax_node(monkeypatch) compose_calls = [] notifications = [] monkeypatch.setattr( module._project_module, "compose_h3_project_video", lambda project_name, data: compose_calls.append((project_name, data)) or Path("/temp/demo.mp4"), ) module.EasyMultiTrackProjectVideoCombine.hidden = types.SimpleNamespace( unique_id="17" ) server = types.ModuleType("server") server.PromptServer = types.SimpleNamespace( instance=types.SimpleNamespace( send_sync=lambda event, payload: notifications.append((event, payload)) ) ) monkeypatch.setitem(sys.modules, "server", server) output = module.EasyMultiTrackProjectVideoCombine.execute( "demo", json.dumps({"project_name": "demo", "clips": []}), ) assert output.values[0].path == "/temp/demo.mp4" assert output.values[1] == "easy_media/projects/demo/out/demo" assert compose_calls == [("demo", {"project_name": "demo", "clips": []})] assert notifications == [( "easy-media.project.selected", {"node_id": "17", "project_name": "demo"}, )] def test_project_video_combine_blocks_both_outputs_when_auto_combine_is_disabled(monkeypatch): module = _load_minimax_node(monkeypatch) compose = monkeypatch.setattr( module._project_module, "compose_h3_project_video", lambda *_args: pytest.fail("disabled auto combine must not compose video"), ) del compose output = module.EasyMultiTrackProjectVideoCombine.execute( "demo", {"project_name": "demo", "clips": [], "auto_combine": False}, ) assert type(output.values[0]).__name__ == "ExecutionBlocker" assert output.values[1] is output.values[0] def test_moxie_combine_schema_exposes_components_only(monkeypatch): _load_minimax_node(monkeypatch) combine = _load_moxie_combine(monkeypatch) schema = combine.MoxieCombine.define_schema() inputs = {port.name: port for port in schema.inputs} assert schema.node_id == "MoxieCombine" assert schema.display_name == "Moxie Project Video Combine" assert inputs["project_name"].kwargs["force_input"] is True assert "project_data" in inputs assert [output.name for output in schema.outputs] == ["images", "audio", "fps"] def test_moxie_combine_returns_components_without_prefix(monkeypatch): module = _load_minimax_node(monkeypatch) combine = _load_moxie_combine(monkeypatch) components = types.SimpleNamespace(images="images", audio="audio", frame_rate=24) video = types.SimpleNamespace(get_components=lambda: components) calls = [] monkeypatch.setattr( module.EasyMultiTrackProjectVideoCombine, "execute", classmethod( lambda cls, *args, **kwargs: calls.append(kwargs) or (video, "easy_media/projects/demo/out/demo") ), ) output = combine.MoxieCombine.execute("demo", {"project_name": "demo"}) assert output.values == ("images", "audio", 24.0) assert calls == [{ "project_name": "demo", "project_data": {"project_name": "demo"}, }] def test_moxie_combine_blocks_all_outputs_when_auto_combine_is_disabled(monkeypatch): from comfy_execution.graph_utils import ExecutionBlocker module = _load_minimax_node(monkeypatch) combine = _load_moxie_combine(monkeypatch) blocker = ExecutionBlocker(None) monkeypatch.setattr( module.EasyMultiTrackProjectVideoCombine, "execute", classmethod(lambda cls, *args, **kwargs: (blocker, blocker)), ) output = combine.MoxieCombine.execute( "demo", {"project_name": "demo", "auto_combine": False} ) assert output.values == (blocker, blocker, blocker) def test_easy_h3_hard_and_hires_context_schemas_and_wrappers(monkeypatch): module = _load_minimax_node(monkeypatch) assert not hasattr(module, "EasyMiniMaxH3MotionContext") hard_schema = module.EasyMiniMaxH3MotionContextHard.define_schema() hard_inputs = {port.name: port for port in hard_schema.inputs} assert hard_schema.node_id == "easy MiniMaxH3MotionContextHard" assert hard_inputs["context_length"].kwargs["default"] == "22" assert "audio_context_length" not in hard_inputs assert [output.name for output in hard_schema.outputs] == [ "conditioning", "trim_frames", "latent", ] motion_context_calls = [] monkeypatch.setattr( module, "apply_reencoded_anchor_motion_context", lambda **kwargs: motion_context_calls.append(kwargs) or ("conditioned", 22, "hard-latent"), ) hard_output = module.EasyMiniMaxH3MotionContextHard.execute( "conditioning", "vae", {"samples": "target"}, {"samples": "context"}, ) assert hard_output.values == ("conditioned", 22, "hard-latent") assert motion_context_calls == [ { "conditioning": "conditioning", "vae": "vae", "latent": {"samples": "target"}, "context_length": "22", "context_latent": {"samples": "context"}, "anchor_length": "5", } ] hires_schema = module.EasyMiniMaxH3HiResContinuity.define_schema() assert hires_schema.node_id == "easy MiniMaxH3HiResContinuity" monkeypatch.setattr( module, "apply_hires_anchor_continuity", lambda **_kwargs: ("hires-latent", 22), ) hires_output = module.EasyMiniMaxH3HiResContinuity.execute( {"samples": "current"}, {"samples": "previous"}, ) assert hires_output.values == ("hires-latent", 22) trim_schema = module.EasyH3MotionContextLatentTrim.define_schema() assert trim_schema.node_id == "easy h3MotionContextLatentTrim" monkeypatch.setattr( module, "trim_motion_context_latent", lambda latent, context_length: { "source": latent, "context_length": context_length, }, ) trim_output = module.EasyH3MotionContextLatentTrim.execute( {"samples": "full"}, "22", ) assert trim_output.values == ( {"source": {"samples": "full"}, "context_length": "22"}, ) def test_easy_h3_hard_and_hires_context_have_chinese_localization(): locale_path = Path(__file__).parents[1] / "locales" / "zh" / "nodeDefs.json" if not locale_path.exists(): pytest.skip("locales/zh/nodeDefs.json is not bundled in this repository") node_defs = json.loads( (Path(__file__).parents[1] / "locales" / "zh" / "nodeDefs.json").read_text() ) hard_translation = node_defs["easy MiniMaxH3MotionContextHard"] assert set(hard_translation["inputs"]) == { "conditioning", "vae", "latent", "context_latent", "context_length", "video_transition_steps", "audio_transition_steps", "video_anchor_only", } assert "easy MiniMaxH3HiResContinuity" in node_defs assert "easy MiniMaxH3MotionContextAnchor" not in node_defs assert "easy MiniMaxH3HiResAnchorContinuity" not in node_defs def test_multitrack_h3_project_has_matching_chinese_localization(): locale_path = Path(__file__).parents[1] / "locales" / "zh" / "nodeDefs.json" if not locale_path.exists(): pytest.skip("locales/zh/nodeDefs.json is not bundled in this repository") node_defs = json.loads( (Path(__file__).parents[1] / "locales" / "zh" / "nodeDefs.json").read_text() ) translation = node_defs["easy multitrackProject"] assert translation["display_name"] == "多轨项目" assert set(translation["inputs"]) == { "model_loader", "tracks_info", "sampler", "sigmas", "project_name", "project_save", "sampling_plan", "sampling_mode", "transition_ratio", "lowres_scale", "enabled_tiling", "tile_count", "model_loader_2nd", "1st_pass_only", "disable_2nd_noise", "upscale_by", "upscale_model", "segment_start_number", "segment_count", "seed", "sampler_2nd", "sigmas_2nd", } assert translation["outputs"] == { "0": {"name": "项目名称"}, "1": {"name": "锁定音频"}, } def test_multitrack_h3_project_expands_single_task_sampling_pipeline(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None log_messages = [] monkeypatch.setattr( module._project_module, "log_node_info", lambda node_name, message=None: log_messages.append((node_name, message)), ) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( project_name="demo", sampling_mode=_h3_sampling_mode("single"), ) ) nodes_by_type = {node["class_type"]: node for node in result.expand.values()} assert result.values[0][1] == 0 assert nodes_by_type["KSamplerSelect"]["inputs"]["sampler_name"] == "er_sde" assert "ManualSigmas" in nodes_by_type conditioning = nodes_by_type["easy minimaxH3ToVideo"]["inputs"] assert conditioning["mode"] == "multi_frames" assert conditioning["ref_image_size"] == "match" assert (conditioning["width"], conditioning["height"]) == (1344, 768) assert "SamplerCustomAdvanced" in nodes_by_type assert "easy h3SegmentSamplingStart" in nodes_by_type assert ( nodes_by_type["easy h3SegmentSamplingStart"]["inputs"]["project_name"] == "demo" ) assert "easy h3SegmentSaveEnd" in nodes_by_type assert "VAEDecode" in nodes_by_type assert "VAEDecodeAudio" in nodes_by_type assert "easy h3SegmentEncodingStart" not in nodes_by_type assert "easy saveVideo" in nodes_by_type save_inputs = nodes_by_type["easy saveVideo"]["inputs"] assert save_inputs["input_mode"] == "images+audio" assert save_inputs["output_mode"] == "hide&save" assert "input_mode.images" in save_inputs assert "input_mode.audio" in save_inputs assert save_inputs["input_mode.fps"] == 24.0 assert "easy h3ProjectArtifact" in nodes_by_type save_video_id = next( node_id for node_id, node in result.expand.items() if node["class_type"] == "easy saveVideo" ) save_end_id = next( node_id for node_id, node in result.expand.items() if node["class_type"] == "easy h3SegmentSaveEnd" ) assert nodes_by_type["easy h3SegmentSaveEnd"]["inputs"]["video_path"] == [ save_video_id, 1, ] assert nodes_by_type["easy h3SegmentSaveEnd"]["inputs"]["project_name"] == "demo" assert nodes_by_type["easy h3ProjectArtifact"]["inputs"]["video_path"] == [ save_end_id, 0, ] assert nodes_by_type["easy h3ProjectArtifact"]["inputs"]["project_save"] == "new" assert nodes_by_type["easy h3ProjectArtifact"]["inputs"]["seed"] == 42 assert log_messages == [ ("MultiTrack Project", "Found 1 segments; processing 1"), ] def test_multitrack_project_patches_sampler_when_loader_has_preview_vae(monkeypatch): module = _load_minimax_node(monkeypatch) module.comfy_nodes.NODE_CLASS_MAPPINGS["ImageResizeKJv2"] = _ImageResizeKJWithNvidia module.EasyMultiTrackProject.hidden = types.SimpleNamespace(unique_id="42") inputs = _h3_project_inputs(sampling_mode=_h3_sampling_mode("dual")) preview_vae = object() inputs["model_loader"][0]["preview_vae"] = preview_vae result = module.EasyMultiTrackProject.execute(**inputs) preview_samplers = [ node for node in result.expand.values() if node["class_type"] == "easy h3SamplingPreviewSampler" ] assert len(preview_samplers) == 2 assert not any( node["class_type"] == "SamplerCustomAdvanced" for node in result.expand.values() ) assert {node["inputs"]["sampling_pass"] for node in preview_samplers} == { "first", "second", } assert all(node["inputs"]["preview_vae"] is preview_vae for node in preview_samplers) assert all(node["inputs"]["preview_node_id"] == "42" for node in preview_samplers) assert all(node["inputs"]["preview_fps"] == 24.0 for node in preview_samplers) def test_h3_sampling_preview_sampler_exposes_optional_tiling(monkeypatch): module = _load_minimax_node(monkeypatch) schema = module._project_module.EasyH3SamplingPreviewSampler.define_schema() inputs = {port.name: port for port in schema.inputs} assert inputs["enabled_tiling"].kwargs["default"] is False assert inputs["enabled_tiling"].kwargs["optional"] is True assert inputs["tile_count"].kwargs["default"] == 2 assert inputs["tile_count"].kwargs["min"] == 2 assert inputs["tile_count"].kwargs["max"] == 8 assert inputs["preview_vae"].kwargs["optional"] is True def test_multitrack_project_tiles_inside_second_pass_sampler(monkeypatch): module = _load_minimax_node(monkeypatch) module.comfy_nodes.NODE_CLASS_MAPPINGS["ImageResizeKJv2"] = _ImageResizeKJWithNvidia inputs = _h3_project_inputs( sampling_mode=_h3_sampling_mode("dual"), enabled_tiling=[{"enabled_tiling": ["true"], "tile_count": [4]}], ) result = module.EasyMultiTrackProject.execute(**inputs) nodes = list(result.expand.values()) tiled_sampler = next( node for node in nodes if node["class_type"] == "easy h3SamplingPreviewSampler" ) assert tiled_sampler["inputs"]["enabled_tiling"] is True assert tiled_sampler["inputs"]["tile_count"] == 4 assert not any(node["class_type"] == "easy h3TiledModel" for node in nodes) def test_multitrack_project_adds_preview_inputs_to_selflift_sampler(monkeypatch): module = _load_minimax_node(monkeypatch) module.EasyMultiTrackProject.hidden = types.SimpleNamespace(unique_id="84") inputs = _h3_project_inputs(sampling_mode=_h3_sampling_mode("selflift")) preview_vae = object() inputs["model_loader"][0]["preview_vae"] = preview_vae result = module.EasyMultiTrackProject.execute(**inputs) sampler = _graph_node(result, "easy minimaxH3SelfLiftSampler") assert sampler["inputs"]["preview_vae"] is preview_vae assert sampler["inputs"]["preview_node_id"] == "84" assert sampler["inputs"]["sampling_pass"] == "selflift" def test_multitrack_h3_project_reads_ref_image_size_from_each_segment(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None inputs = _h3_project_inputs(sampling_mode=_h3_sampling_mode("single")) inputs["tracks_info"][0]["tracks"][0]["segments"][0]["content"].update({ "task_mode": "ref", "ref_image_size": "max", }) result = module.EasyMultiTrackProject.execute(**inputs) conditioning = _graph_node(result, "easy minimaxH3ToVideo")["inputs"] assert conditioning["mode"] == "reference" assert conditioning["ref_image_size"] == "max" def test_multitrack_h3_project_logs_steps_and_reports_progress(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None log_messages = [] monkeypatch.setattr( module._project_module, "log_node_info", lambda node_name, message=None: log_messages.append((node_name, message)), ) module.EasyMultiTrackProject.execute( **_h3_project_inputs(sampling_mode=_h3_sampling_mode("single")) ) assert log_messages == [ ("MultiTrack Project", "Found 1 segments; processing 1"), ] progress = _ProgressBar.instances[-1] assert progress.total == 100 assert progress.updates[0] == (0, 100) assert progress.updates[-1] == (100, 100) assert all( current[0] <= following[0] for current, following in zip(progress.updates, progress.updates[1:]) ) def test_multitrack_h3_project_locks_task_audio_before_sampling(monkeypatch): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs(sampling_mode=_h3_sampling_mode("single")) task = inputs["tracks_info"][0]["tracks"][0]["segments"][0] inputs["tracks_info"][0]["tracks"].append({ "id": "locked-audio-track", "type": "audio", "audio_locked": True, "segments": [{ "id": "locked-audio", "start_frame": task["start_frame"], "end_frame": task["end_frame"], "content": {"media_type": "audio"}, }], }) result = module.EasyMultiTrackProject.execute(**inputs) nodes = result.expand lock_id, audio_lock = next( (node_id, node) for node_id, node in nodes.items() if node["class_type"] == "easy minimaxH3AudioLock" ) conditioning_cache_id = next( node_id for node_id, node in nodes.items() if node["class_type"] == "easy h3ConditioningCache" ) sampling_start = next( node for node in nodes.values() if node["class_type"] == "easy h3SegmentSamplingStart" ) assert audio_lock["inputs"]["latent"] == [conditioning_cache_id, 1] assert audio_lock["inputs"]["audio"] == {"prepared_locked_audio": True} assert audio_lock["inputs"]["remix_strength"] == 1.0 assert audio_lock["inputs"]["prepend_frames"] == 0 assert audio_lock["inputs"]["frame_rate"] == 24.0 assert sampling_start["inputs"]["latent_image"] == [lock_id, 0] def test_multitrack_h3_fast_dual_non_turbo_uses_preset_sigmas_and_pixel_upscale( monkeypatch, ): module = _load_minimax_node(monkeypatch) assert module is not None module.comfy_nodes.NODE_CLASS_MAPPINGS["ImageResizeKJv2"] = ( _ImageResizeKJWithNvidia ) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( sampling_mode=_h3_sampling_mode("dual", disable_2nd_noise=[True]), sampling_plan=["light"], ) ) nodes = list(result.expand.values()) conditioning = next( node for node in nodes if node["class_type"] == "easy minimaxH3ToVideo" ) resize = next(node for node in nodes if node["class_type"] == "ImageResizeKJv2") separate = next( node for node in nodes if node["class_type"] == "LTXVSeparateAVLatent" ) encode = next(node for node in nodes if node["class_type"] == "VAEEncode") concat = next( node for node in nodes if node["class_type"] == "LTXVConcatAVLatent" ) samples = [ node for node in nodes if node["class_type"] == "SamplerCustomAdvanced" ] assert not any(node["class_type"] == "SplitSigmas" for node in nodes) sigma_schedules = [ node["inputs"]["sigmas"] for node in nodes if node["class_type"] == "ManualSigmas" ] assert len(sigma_schedules) == 2 assert sigma_schedules[0].startswith("1.0000, 0.9901") assert sigma_schedules[1].startswith("0.6316, 0.4877") assert (conditioning["inputs"]["width"], conditioning["inputs"]["height"]) == ( 1344, 768, ) assert resize["inputs"]["upscale_method"] == "nvidia_rtx_vsr" assert (resize["inputs"]["width"], resize["inputs"]["height"]) == (1664, 960) separate_id = next( node_id for node_id, node in result.expand.items() if node is separate ) encode_id = next( node_id for node_id, node in result.expand.items() if node is encode ) assert concat["inputs"]["video_latent"] == [encode_id, 0] assert concat["inputs"]["audio_latent"] == [separate_id, 1] assert len(samples) == 2 sampler_names = [ node["inputs"]["sampler_name"] for node in nodes if node["class_type"] == "KSamplerSelect" ] assert sampler_names == ["euler", "sa_solver"] assert any(node["class_type"] == "DisableNoise" for node in nodes) def test_multitrack_h3_selflift_uses_target_size_and_one_progressive_sample( monkeypatch, ): module = _load_minimax_node(monkeypatch) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( sampling_mode=_h3_sampling_mode("selflift"), sampling_plan=["medium"], upscale_by=[1.25], upscale_model=["None"], ) ) nodes = list(result.expand.values()) selflift = next( node for node in nodes if node["class_type"] == "easy minimaxH3SelfLiftSampler" ) artifact = next( node for node in nodes if node["class_type"] == "easy h3ProjectArtifact" ) conditioning = next( node for node in nodes if node["class_type"] == "easy minimaxH3ToVideo" ) sigma_nodes = [node for node in nodes if node["class_type"] == "ManualSigmas"] assert len(sigma_nodes) == 1 assert sigma_nodes[0]["inputs"]["sigmas"].startswith("1.0000, 0.9956") assert not any(node["class_type"] == "KSamplerSelect" for node in nodes) assert not any(node["class_type"] == "SamplerCustomAdvanced" for node in nodes) assert not any(node["class_type"] == "VAEEncode" for node in nodes) assert not any(node["class_type"] == "ImageResizeKJv2" for node in nodes) assert (conditioning["inputs"]["width"], conditioning["inputs"]["height"]) == ( 1344, 768, ) assert selflift["inputs"]["transition_ratio"] == 0.6 assert selflift["inputs"]["lowres_scale"] == 0.6 assert selflift["inputs"]["rho"] == 0.0 assert selflift["inputs"]["w_min"] == 0.25 assert selflift["inputs"]["w_max"] == 0.7 assert selflift["inputs"]["enabled_tiling"] is False assert "project_name" not in selflift["inputs"] assert "segment_index" not in selflift["inputs"] assert "upscale_by" not in selflift["inputs"] assert selflift["inputs"]["upscaler_model"] == "None" assert selflift["inputs"]["highres_model"] is selflift["inputs"]["model"] assert artifact["inputs"]["sampling_pass"] == "single" assert artifact["inputs"]["tracks_info"]["width"] == 1344 assert artifact["inputs"]["tracks_info"]["height"] == 768 def test_multitrack_h3_selflift_reads_nested_upscale_model(monkeypatch): module = _load_minimax_node(monkeypatch) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( sampling_mode=_h3_sampling_mode( "selflift", upscale_by=[2.5], upscale_model=["h3_upscale.safetensors"], ), ) ) selflift = next( node for node in result.expand.values() if node["class_type"] == "easy minimaxH3SelfLiftSampler" ) assert selflift["inputs"]["lowres_scale"] == 0.6 assert "upscale_by" not in selflift["inputs"] assert selflift["inputs"]["upscaler_model"] == "h3_upscale.safetensors" def test_multitrack_h3_selflift_passes_dynamic_sampling_values(monkeypatch): module = _load_minimax_node(monkeypatch) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( sampling_mode=_h3_sampling_mode( "selflift", transition_ratio=[0.75], lowres_scale=[0.5], ), enabled_tiling=[ {"enabled_tiling": ["true"], "tile_count": [4]} ], ) ) selflift = next( node for node in result.expand.values() if node["class_type"] == "easy minimaxH3SelfLiftSampler" ) assert selflift["inputs"]["transition_ratio"] == 0.75 assert selflift["inputs"]["lowres_scale"] == 0.5 assert selflift["inputs"]["enabled_tiling"] is True assert selflift["inputs"]["tile_count"] == 4 @pytest.mark.parametrize("continuity_mode", ["context", "context_swap"]) def test_multitrack_h3_selflift_supports_context_and_locked_audio( monkeypatch, continuity_mode, ): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs(sampling_mode=_h3_sampling_mode("selflift")) info = inputs["tracks_info"][0] info["tracks"][0]["segments"].append( { "start_frame": 120, "end_frame": 240, "content": { "task_mode": "default", "continuity_mode": continuity_mode, "images": [], "user_prompt": "continue seamlessly", }, } ) info["tracks"].append( { "id": "locked-audio-track", "type": "audio", "audio_locked": True, "segments": [ { "id": "locked-audio", "start_frame": 0, "end_frame": 240, "content": {"media_type": "audio"}, } ], } ) result = module.EasyMultiTrackProject.execute(**inputs) selflift_nodes = [ (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy minimaxH3SelfLiftSampler" ] assert len(selflift_nodes) == 2 context_selflift = selflift_nodes[1][1] audio_lock_id = next( node_id for node_id, node in result.expand.items() if node["class_type"] == "easy minimaxH3AudioLock" and node["inputs"]["prepend_frames"] == 22 ) context_type = ( "easy MiniMaxH3MotionContextHard" if continuity_mode == "context" else "easy MiniMaxH3ContextSwap" ) context_id, context_node = next( (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == context_type ) assert context_node["inputs"]["latent"] == [audio_lock_id, 0] latent_output = 2 if continuity_mode == "context" else 1 assert context_selflift["inputs"]["latent_image"] == [context_id, latent_output] low_context_link = context_selflift["inputs"]["low_context_latent"] low_context_node = result.expand[low_context_link[0]] assert low_context_node["class_type"] == "easy h3MotionContextLatentTrim" if continuity_mode == "context_swap": assert context_selflift["inputs"]["model"] == [context_id, 0] def test_selflift_sampler_schema_forces_euler_by_not_exposing_a_sampler(monkeypatch): module = _load_minimax_node(monkeypatch) schema = module.EasyMiniMaxH3SelfLiftSampler.define_schema() inputs = {port.name: port for port in schema.inputs} assert schema.node_id == "easy minimaxH3SelfLiftSampler" assert schema.is_dev_only is True assert "sampler" not in inputs assert list(inputs)[-2:] == ["previous", "highres_model"] assert inputs["highres_model"].kwargs["optional"] is True assert inputs["transition_ratio"].kwargs["default"] == 0.6 assert inputs["lowres_scale"].kwargs["default"] == 0.6 assert inputs["rho"].kwargs["default"] == pytest.approx(0.1) assert inputs["w_max"].kwargs["default"] == pytest.approx(0.7) assert inputs["w_min"].kwargs["default"] == pytest.approx(0.25) assert inputs["enabled_tiling"].kwargs["default"] is False assert inputs["enabled_tiling"].kwargs["optional"] is True assert inputs["highres_tiling"].kwargs["default"] is False assert inputs["highres_tiling"].kwargs["optional"] is True assert inputs["highres_tiling"].kwargs["advanced"] is True assert inputs["low_context_latent"].kwargs["optional"] is True assert "project_name" not in inputs assert inputs["segment_index"].kwargs["optional"] is True assert len(schema.outputs) == 2 assert "upscale_by" not in inputs def test_bundled_h3_latent_upscaler_schema_exposes_three_resize_modes(monkeypatch): module = _load_minimax_node(monkeypatch) schema = module.EasyMiniMaxH3LatentUpscaler.define_schema() inputs = {port.name: port for port in schema.inputs} modes = inputs["mode"].kwargs["options"] assert schema.node_id == "easy minimaxH3LatentUpscaler" assert list(inputs) == [ "latent", "model_name", "mode", "align", "enable_temporal_chunking", "force_unload", ] assert [name for name, _ in modes] == [ "scale by multiplier", "target dimensions", "megapixels", ] assert inputs["align"].kwargs["default"] == 32 assert inputs["enable_temporal_chunking"].kwargs["default"] is True assert inputs["force_unload"].kwargs["default"] is True locale_path = Path(__file__).parents[1] / "locales" / "zh" / "nodeDefs.json" if not locale_path.exists(): pytest.skip("locales/zh/nodeDefs.json is not bundled in this repository") node_defs = json.loads( (Path(__file__).parents[1] / "locales" / "zh" / "nodeDefs.json").read_text() ) assert set(node_defs["easy minimaxH3LatentUpscaler"]["inputs"]) == { "latent", "model_name", "mode", "scale", "width", "height", "megapixels", "align", "enable_temporal_chunking", "force_unload", } def test_bundled_h3_latent_upscaler_preserves_audio_and_resizes_mask(monkeypatch): module = _load_minimax_node(monkeypatch) calls = [] upscaler_module = types.ModuleType( "easy_media.modules.selflift.h3_latent_upscale" ) def learned_latent_lift(video, target_size, model_name, **kwargs): calls.append((tuple(video.shape), target_size, model_name, kwargs)) return torch.full( (*video.shape[:-2], *target_size), 7.0, dtype=video.dtype, ) upscaler_module.learned_latent_lift = learned_latent_lift monkeypatch.setitem( sys.modules, "easy_media.modules.selflift.h3_latent_upscale", upscaler_module, ) video = torch.zeros(1, 24, 3, 4, 6) audio = torch.randn(1, 32, 2, 9) video_mask = torch.ones(1, 1, 3, 4, 6) audio_mask = torch.zeros(1, 1, 2, 9) latent = { "samples": _NestedTensor((video, audio)), "noise_mask": _NestedTensor((video_mask, audio_mask)), "batch_index": [0], } output = module.EasyMiniMaxH3LatentUpscaler.execute( latent=latent, model_name="h3_upscale.safetensors", mode={"mode": "target dimensions", "width": 192, "height": 128}, align=32, enable_temporal_chunking=False, force_unload=False, ).values[0] upscaled_video, output_audio = output["samples"].unbind() upscaled_mask, output_audio_mask = output["noise_mask"].unbind() assert calls == [ ( (1, 24, 3, 4, 6), (8, 12), "h3_upscale.safetensors", {"enable_temporal_chunking": False, "force_unload": False}, ) ] assert upscaled_video.shape == (1, 24, 3, 8, 12) assert torch.equal(output_audio, audio) assert upscaled_mask.shape == (1, 1, 3, 8, 12) assert torch.equal(output_audio_mask, audio_mask) assert output["batch_index"] == [0] def test_selflift_sampler_uses_lowres_scale(monkeypatch): module = _load_minimax_node(monkeypatch) captured = {} sampled = {"samples": object()} low_context = {"samples": object()} sampling_module = types.ModuleType("easy_media.modules.selflift.sampling") def progressive_sample_h3(**kwargs): captured.update(kwargs) return sampled, low_context sampling_module.progressive_sample_h3 = progressive_sample_h3 monkeypatch.setitem( sys.modules, "easy_media.modules.selflift.sampling", sampling_module, ) result = module.EasyMiniMaxH3SelfLiftSampler.execute( model=object(), positive=[], vae=object(), latent_image={"samples": torch.zeros(1, 24, 1, 60, 104)}, sigmas=torch.tensor([1.0, 0.0]), seed=42, transition_ratio=0.6, lowres_scale=0.6, ) assert captured["lowres_scale"] == pytest.approx(0.6) assert captured["rho"] == pytest.approx(0.1) assert captured["w_min"] == 0.25 assert captured["w_max"] == 0.7 assert captured["highres_tiling"] is False assert captured["low_context_latent"] is None assert result.values == (sampled, low_context) def test_selflift_sampler_forwards_high_resolution_model(monkeypatch): module = _load_minimax_node(monkeypatch) captured = {} sampling_module = types.ModuleType("easy_media.modules.selflift.sampling") def progressive_sample_h3(**kwargs): captured.update(kwargs) return {"samples": object()}, {"samples": object()} sampling_module.progressive_sample_h3 = progressive_sample_h3 monkeypatch.setitem(sys.modules, "easy_media.modules.selflift.sampling", sampling_module) first_model = object() second_model = object() module.EasyMiniMaxH3SelfLiftSampler.execute( model=first_model, highres_model=second_model, positive=[], vae=object(), latent_image={"samples": torch.zeros(1, 24, 1, 60, 104)}, sigmas=torch.tensor([1.0, 0.0]), seed=42, ) assert captured["model"] is first_model assert captured["highres_model"] is second_model def test_selflift_sampler_forwards_artifact_correction_controls(monkeypatch): module = _load_minimax_node(monkeypatch) captured = {} sampling_module = types.ModuleType("easy_media.modules.selflift.sampling") def progressive_sample_h3(**kwargs): captured.update(kwargs) return {"samples": object()}, {"samples": object()} sampling_module.progressive_sample_h3 = progressive_sample_h3 monkeypatch.setitem( sys.modules, "easy_media.modules.selflift.sampling", sampling_module, ) module.EasyMiniMaxH3SelfLiftSampler.execute( model=object(), positive=[], vae=object(), latent_image={"samples": torch.zeros(1, 24, 1, 60, 104)}, sigmas=torch.tensor([1.0, 0.0]), seed=42, rho=0.2, w_max=0.8, w_min=0.3, ) assert captured["rho"] == pytest.approx(0.2) assert captured["w_max"] == pytest.approx(0.8) assert captured["w_min"] == pytest.approx(0.3) def test_selflift_sampler_uses_saved_low_context_for_masked_video(monkeypatch): module = _load_minimax_node(monkeypatch) captured = {} sampled = {"samples": object()} low_result = {"samples": object()} sampling_module = types.ModuleType("easy_media.modules.selflift.sampling") def progressive_sample_h3(**kwargs): captured.update(kwargs) return sampled, low_result sampling_module.progressive_sample_h3 = progressive_sample_h3 monkeypatch.setitem( sys.modules, "easy_media.modules.selflift.sampling", sampling_module, ) video = torch.zeros(1, 24, 7, 4, 6) audio = torch.zeros(1, 32, 2, 20) video_mask = torch.ones(1, 1, 7, 4, 6) video_mask[:, :, :2] = 0.0 audio_mask = torch.ones(1, 1, 2, 20) saved_low_context = { "samples": _NestedTensor( (torch.ones(1, 24, 2, 3, 4), torch.ones_like(audio)) ) } result = module.EasyMiniMaxH3SelfLiftSampler.execute( model=object(), positive=[], vae=object(), latent_image={ "samples": _NestedTensor((video, audio)), "noise_mask": _NestedTensor((video_mask, audio_mask)), }, sigmas=torch.tensor([1.0, 0.5, 0.0]), seed=42, enabled_tiling=True, tile_count=4, low_context_latent=saved_low_context, ) assert captured["highres_tiling"] is True assert captured["tile_count"] == 4 assert captured["low_context_latent"] is saved_low_context assert captured["rho"] == 0.1 assert captured["w_min"] == 0.25 assert captured["w_max"] == 0.7 assert result.values == (sampled, low_result) def test_selflift_sampler_keeps_audio_only_mask_on_selflift(monkeypatch): module = _load_minimax_node(monkeypatch) captured = {} sampled = {"samples": object()} low_context = {"samples": object()} sampling_module = types.ModuleType("easy_media.modules.selflift.sampling") def progressive_sample_h3(**kwargs): captured.update(kwargs) return sampled, low_context sampling_module.progressive_sample_h3 = progressive_sample_h3 monkeypatch.setitem( sys.modules, "easy_media.modules.selflift.sampling", sampling_module, ) video = torch.zeros(1, 24, 7, 4, 6) audio = torch.zeros(1, 32, 2, 20) video_mask = torch.ones(1, 1, 7, 4, 6) audio_mask = torch.zeros(1, 1, 2, 20) result = module.EasyMiniMaxH3SelfLiftSampler.execute( model=object(), positive=[], vae=object(), latent_image={ "samples": _NestedTensor((video, audio)), "noise_mask": _NestedTensor((video_mask, audio_mask)), }, sigmas=torch.tensor([1.0, 0.5, 0.0]), seed=42, ) assert captured["rho"] == pytest.approx(0.1) assert captured["lowres_scale"] == pytest.approx(0.6) assert result.values == (sampled, low_context) @pytest.mark.parametrize("upscale_model", ["None", "h3_upscale.safetensors"]) @pytest.mark.parametrize( ("upscale_by", "expected_size"), [(None, (1664, 960)), ([1.0], (1344, 768)), ([1.234], (1664, 960)), (2.0, (2688, 1536))], ) def test_multitrack_project_uses_flat_upscale_by( monkeypatch, tmp_path, upscale_model, upscale_by, expected_size ): module = _load_minimax_node(monkeypatch) module.comfy_nodes.NODE_CLASS_MAPPINGS["ImageResizeKJv2"] = _ImageResizeKJWithNvidia monkeypatch.setattr(module.folder_paths, "get_output_directory", lambda: str(tmp_path)) inputs = _h3_project_inputs(sampling_mode=["dual"], upscale_model=[upscale_model]) if upscale_by is not None: inputs["upscale_by"] = upscale_by result = module.EasyMultiTrackProject.execute(**inputs) conditioning = _graph_node(result, "easy minimaxH3ToVideo")["inputs"] assert (conditioning["width"], conditioning["height"]) == (1344, 768) artifact_info = _graph_node(result, "easy h3ProjectArtifact")["inputs"]["tracks_info"] assert (artifact_info["width"], artifact_info["height"]) == expected_size manifest = json.loads((tmp_path / "easy_media/projects/default/project.json").read_text()) assert (manifest["width"], manifest["height"]) == expected_size assert (inputs["tracks_info"][0]["width"], inputs["tracks_info"][0]["height"]) == (1344, 768) upscale_type = "ImageResizeKJv2" if upscale_model == "None" else "easy minimaxH3LatentUpscaler" if expected_size == (1344, 768): assert not any(node["class_type"] == upscale_type for node in result.expand.values()) else: upscale = _graph_node(result, upscale_type)["inputs"] prefix = "" if upscale_model == "None" else "mode." assert (upscale[f"{prefix}width"], upscale[f"{prefix}height"]) == expected_size conditionings = [node["inputs"] for node in result.expand.values() if node["class_type"] == "easy minimaxH3ToVideo"] assert (conditionings[1]["width"], conditionings[1]["height"]) == expected_size def test_multitrack_reference_dual_reuses_conditioning_for_second_pass(monkeypatch): module = _load_minimax_node(monkeypatch) module.comfy_nodes.NODE_CLASS_MAPPINGS["ImageResizeKJv2"] = _ImageResizeKJWithNvidia inputs = _h3_project_inputs(sampling_mode=_h3_sampling_mode("dual")) inputs["tracks_info"][0]["tracks"][0]["segments"][0]["content"]["task_mode"] = "ref" result = module.EasyMultiTrackProject.execute(**inputs) graph = result.expand conditioning_nodes = [ node for node in graph.values() if node["class_type"] == "easy minimaxH3ToVideo" ] assert len(conditioning_nodes) == 1 assert conditioning_nodes[0]["inputs"]["mode"] == "reference" cache_id = next( node_id for node_id, node in graph.items() if node["class_type"] == "easy h3ConditioningCache" ) guiders = [ node for node in graph.values() if node["class_type"] == "BasicGuider" ] assert len(guiders) == 2 assert all(node["inputs"]["conditioning"] == [cache_id, 0] for node in guiders) def test_multitrack_h3_project_forwards_override_save_mode(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None result = module.EasyMultiTrackProject.execute( **_h3_project_inputs(project_save=["override"]) ) artifact = _graph_node(result, "easy h3ProjectArtifact") assert artifact["inputs"]["project_save"] == "override" def test_multitrack_project_clears_remaining_old_segments_only_for_unlimited_override( monkeypatch, ): module = _load_minimax_node(monkeypatch) clear_calls = [] monkeypatch.setattr( module._project_module, "clear_h3_project_segments_from", lambda project_name, start_index, output_directory: clear_calls.append( (project_name, start_index, output_directory) ) or [], ) module.EasyMultiTrackProject.execute( **_h3_project_inputs( project_save=["new"], segment_start_number=[1], segment_count=[-1], ) ) assert clear_calls == [] module.EasyMultiTrackProject.execute( **_h3_project_inputs( project_save=["override"], segment_start_number=[1], segment_count=[-1], ) ) assert clear_calls == [("default", 0, "/tmp")] clear_calls.clear() module.EasyMultiTrackProject.execute( **_h3_project_inputs( project_save=["override"], segment_start_number=[1], segment_count=[1], ) ) assert clear_calls == [] def test_multitrack_project_rejects_zero_start_number(monkeypatch): module = _load_minimax_node(monkeypatch) with pytest.raises(ValueError, match="segment_start_number must be at least 1"): module.EasyMultiTrackProject.execute( **_h3_project_inputs(segment_start_number=[0]) ) def test_multitrack_h3_medium_dual_uses_selected_latent_upscale_model( monkeypatch, ): module = _load_minimax_node(monkeypatch) assert module is not None result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( sampling_mode=_h3_sampling_mode("dual"), upscale_model=["h3_upscale.safetensors"], ) ) nodes = list(result.expand.values()) sampler_names = [ node["inputs"]["sampler_name"] for node in nodes if node["class_type"] == "KSamplerSelect" ] upscale = next( node for node in nodes if node["class_type"] == "easy minimaxH3LatentUpscaler" ) separate = next( node for node in nodes if node["class_type"] == "LTXVSeparateAVLatent" ) concat = next( node for node in nodes if node["class_type"] == "LTXVConcatAVLatent" ) assert sampler_names == ["er_sde", "sa_solver"] assert sum(node["class_type"] == "ManualSigmas" for node in nodes) == 2 assert upscale["inputs"]["model_name"] == "h3_upscale.safetensors" assert upscale["inputs"]["mode"] == "target dimensions" assert upscale["inputs"]["align"] == 32 assert upscale["inputs"]["enable_temporal_chunking"] is True assert upscale["inputs"]["force_unload"] is True assert ( upscale["inputs"]["mode.width"], upscale["inputs"]["mode.height"], ) == ( 1664, 960, ) separate_id = next( node_id for node_id, node in result.expand.items() if node is separate ) upscale_id = next( node_id for node_id, node in result.expand.items() if node is upscale ) assert upscale["inputs"]["latent"] == [separate_id, 0] assert concat["inputs"]["video_latent"] == [upscale_id, 0] assert concat["inputs"]["audio_latent"] == [separate_id, 1] assert not any( node["class_type"] == "LatentUpscaleModelLoader" for node in nodes ) assert not any(node["class_type"] == "ImageResizeKJv2" for node in nodes) def test_multitrack_h3_selected_upscale_model_uses_bundled_node(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( sampling_mode=_h3_sampling_mode("dual"), upscale_model=["h3_upscale.safetensors"], ) ) assert _graph_node(result, "easy minimaxH3LatentUpscaler") def test_multitrack_h3_second_pass_at_one_x_reuses_first_pass_latent(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( sampling_mode=_h3_sampling_mode("dual", upscale_by=[1.0]), ) ) nodes = list(result.expand.values()) assert sum(node["class_type"] == "SamplerCustomAdvanced" for node in nodes) == 2 assert sum( node["class_type"] == "easy h3SegmentSamplingStart" for node in nodes ) == 2 assert sum(node["class_type"] == "easy h3SegmentSaveEnd" for node in nodes) == 1 assert not any(node["class_type"] == "ImageResizeKJv2" for node in nodes) assert not any( node["class_type"] == "easy minimaxH3LatentUpscaler" for node in nodes ) def test_multitrack_h3_connected_sampler_and_sigmas_force_custom(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None custom_sampler = object() custom_sigmas = torch.tensor([1.0, 0.0]) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( sampling_plan=["medium"], sampler=[custom_sampler], sigmas=[custom_sigmas], ) ) nodes = list(result.expand.values()) sampling_start = next( node for node in nodes if node["class_type"] == "easy h3SegmentSamplingStart" ) assert sampling_start["inputs"]["sampler"] is custom_sampler assert sampling_start["inputs"]["sigmas"] is custom_sigmas assert not any(node["class_type"] == "KSamplerSelect" for node in nodes) assert not any(node["class_type"] == "ManualSigmas" for node in nodes) def test_multitrack_h3_dynamic_dual_sampler_and_sigmas_override_each_preset( monkeypatch, ): module = _load_minimax_node(monkeypatch) assert module is not None first_pass_sampler = object() first_pass_sigmas = torch.tensor([1.0, 0.0]) second_pass_sampler = object() second_pass_sigmas = torch.tensor([0.8, 0.0]) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( sampler=[first_pass_sampler], sigmas=[first_pass_sigmas], seed=[12], sampling_mode=_h3_sampling_mode( "dual", sampler_2nd=[second_pass_sampler], sigmas_2nd=[second_pass_sigmas], upscale_by=[1.0], ), ) ) samples = [ node for node in result.expand.values() if node["class_type"] == "SamplerCustomAdvanced" ] noise_seeds = [ node["inputs"]["noise_seed"] for node in result.expand.values() if node["class_type"] == "RandomNoise" ] sampling_starts = [ node for node in result.expand.values() if node["class_type"] == "easy h3SegmentSamplingStart" ] assert sampling_starts[0]["inputs"]["sampler"] is first_pass_sampler assert sampling_starts[0]["inputs"]["sigmas"] is first_pass_sigmas assert sampling_starts[1]["inputs"]["sampler"] is second_pass_sampler assert sampling_starts[1]["inputs"]["sigmas"] is second_pass_sigmas assert samples[1]["inputs"]["sampler"] == [ next( node_id for node_id, node in result.expand.items() if node is sampling_starts[1] ), 2, ] assert noise_seeds == [12, 12] assert not any(node["class_type"] == "KSamplerSelect" for node in result.expand.values()) assert not any(node["class_type"] == "ManualSigmas" for node in result.expand.values()) def test_multitrack_h3_dynamic_dual_model_loader_uses_second_model_for_second_pass( monkeypatch, ): module = _load_minimax_node(monkeypatch) assert module is not None first_model = _MiniMaxH3Model() second_model = _MiniMaxH3Model() first_vae = object() first_audio_vae = object() second_vae = object() second_audio_vae = object() result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( model_loader=[ { "model": first_model, "clip": object(), "vae": first_vae, "audio_vae": first_audio_vae, } ], sampling_mode=_h3_sampling_mode( "dual", model_loader_2nd=[ { "model": second_model, "clip": object(), "vae": second_vae, "audio_vae": second_audio_vae, } ], upscale_by=[1.0], ), ) ) guider_models = [ node["inputs"]["model"] for node in result.expand.values() if node["class_type"] == "BasicGuider" ] assert guider_models == [first_model, second_model] assert _graph_node(result, "VAEDecode")["inputs"]["vae"] is first_vae assert _graph_node(result, "VAEDecodeAudio")["inputs"]["vae"] is first_audio_vae def test_multitrack_h3_selflift_uses_second_model_only_for_high_resolution(monkeypatch): module = _load_minimax_node(monkeypatch) first_model = _MiniMaxH3Model() second_model = _MiniMaxH3Model() result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( model_loader=[{ "model": first_model, "clip": object(), "vae": object(), "audio_vae": object(), }], sampling_mode=_h3_sampling_mode( "selflift", model_loader_2nd=[{"model": second_model}], ), ) ) selflift = _graph_node(result, "easy minimaxH3SelfLiftSampler") assert selflift["inputs"]["model"] is first_model assert selflift["inputs"]["highres_model"] is second_model def test_linked_selflift_forwards_second_model_loader(monkeypatch): module = _load_minimax_node(monkeypatch) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( model_loader=[["first-loader", 0]], sampling_mode=_h3_sampling_mode( "selflift", model_loader_2nd=[["second-loader", 0]] ), ) ) model_id, model_prepare = next( (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy h3ProjectStaticPrepare" and node_id.endswith("project_model_prepare") ) selflift = _graph_node(result, "easy minimaxH3SelfLiftSampler") assert model_prepare["inputs"]["model_loader_2nd"] == ["second-loader", 0] assert selflift["inputs"]["model"] == [model_id, 2] assert selflift["inputs"]["highres_model"] == [model_id, 3] def test_multitrack_h3_rejects_incomplete_custom_sampling(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None with pytest.raises(ValueError, match="both sampler and sigmas"): module.EasyMultiTrackProject.execute( **_h3_project_inputs(sampler=[object()]) ) def test_multitrack_h3_first_pass_preview_skips_second_model_loader(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( sampling_mode=_h3_sampling_mode( "dual", **{"1st_pass_only": [True]}, model_loader_2nd=[{}], ) ) ) assert sum( node["class_type"] == "SamplerCustomAdvanced" for node in result.expand.values() ) == 1 def test_multitrack_h3_first_pass_preview_only_builds_first_selected_task(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None info = _h3_project_inputs()["tracks_info"][0] second = { "start_frame": 120, "end_frame": 240, "content": { "task_mode": "l2v", "continuity_mode": "context", "images": [{"media_index": 0}], "user_prompt": "continue", }, } info["tracks"][0]["segments"].append(second) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( tracks_info=[info], sampling_mode=_h3_sampling_mode("dual", **{"1st_pass_only": [True]}), sampling_plan=["light"], ) ) nodes = list(result.expand.values()) assert sum(node["class_type"] == "easy multiTrackTaskOutput" for node in nodes) == 1 assert sum(node["class_type"] == "SamplerCustomAdvanced" for node in nodes) == 1 assert not any(node["class_type"] == "ImageResizeKJv2" for node in nodes) conditioning = next( node for node in nodes if node["class_type"] == "easy minimaxH3ToVideo" ) assert (conditioning["inputs"]["width"], conditioning["inputs"]["height"]) == ( 1344, 768, ) artifact = next( node for node in nodes if node["class_type"] == "easy h3ProjectArtifact" ) assert artifact["inputs"]["sampling_pass"] == "first" first_pass_sample_id = next( node_id for node_id, node in result.expand.items() if node["class_type"] == "SamplerCustomAdvanced" ) assert artifact["inputs"]["context_latent"] == [first_pass_sample_id, 1] assert artifact["inputs"]["context_latent_low"] == [first_pass_sample_id, 1] def test_multitrack_h3_dual_resumes_saved_first_pass_checkpoint( monkeypatch, tmp_path ): module = _load_minimax_node(monkeypatch) assert module is not None monkeypatch.setattr( module.folder_paths, "get_output_directory", lambda: str(tmp_path), ) project_dir = tmp_path / "easy_media" / "projects" / "demo" project_dir.mkdir(parents=True) (project_dir / "context_latent_0_1.safetensors").write_bytes(b"checkpoint") (project_dir / "project.json").write_text( json.dumps({ "segments": { "0": { "active_generation": 1, "generations": { "1": { "context_latent": "context_latent_0_1.safetensors", "sampling_pass": "first", } }, } } }), encoding="utf-8", ) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( project_name="demo", project_save=["override"], sampling_mode=_h3_sampling_mode("dual", upscale_by=[1.0]), ) ) nodes = list(result.expand.values()) sampling_starts = [ node for node in nodes if node["class_type"] == "easy h3SegmentSamplingStart" ] assert sum( node["class_type"] == "SamplerCustomAdvanced" for node in nodes ) == 1 assert [node["inputs"]["sampling_pass"] for node in sampling_starts] == [ "second" ] checkpoint_load = next( node for node in nodes if node["class_type"] == "easy h3ProjectContextLatentLoad" ) assert checkpoint_load["inputs"] == { "project_name": "demo", "segment_index": 0, } artifact = next( node for node in nodes if node["class_type"] == "easy h3ProjectArtifact" ) assert artifact["inputs"]["sampling_pass"] == "second" retained_manifest = json.loads( (project_dir / "project.json").read_text(encoding="utf-8") ) assert "0" in retained_manifest["segments"] def test_multitrack_h3_context_chain_uses_previous_segment_latent(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None module.comfy_nodes.NODE_CLASS_MAPPINGS["MiniMaxH3MotionContextTrim"] = ( _MiniMaxMotionContextTrim ) info = _h3_project_inputs()["tracks_info"][0] info["tracks"][0]["segments"].append( { "start_frame": 120, "end_frame": 240, "content": { "task_mode": "l2v", "continuity_mode": "context", "images": [{"media_index": 0}], "user_prompt": "continue", }, } ) info["tracks"].append( { "id": "locked-audio-track", "type": "audio", "audio_locked": True, "segments": [ { "id": "locked-audio", "start_frame": 0, "end_frame": 240, "content": {"media_type": "audio"}, } ], } ) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs(tracks_info=[info]) ) nodes = list(result.expand.values()) motion_id, motion = next( (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy MiniMaxH3MotionContextHard" ) first_context_trim_id = motion["inputs"]["context_latent"][0] first_context_trim = result.expand[first_context_trim_id] assert first_context_trim["class_type"] == "easy h3MotionContextLatentTrim" assert first_context_trim["inputs"]["context_length"] == "22" first_context_source = result.expand[first_context_trim["inputs"]["latent"][0]] assert first_context_source["class_type"] == "LTXVConcatAVLatent" assert motion["inputs"]["context_latent"] == [first_context_trim_id, 0] assert "audio_context_length" not in motion["inputs"] assert motion["inputs"]["context_length"] == "22" audio_locks = [ (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy minimaxH3AudioLock" ] assert [node["inputs"]["prepend_frames"] for _, node in audio_locks] == [0, 22] context_audio_lock_id = next( node_id for node_id, node in audio_locks if node["inputs"]["prepend_frames"] == 22 ) assert motion["inputs"]["latent"] == [context_audio_lock_id, 0] math_id, math_node = next( (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "ComfyMathExpression" ) context_conditioning = next( node for node in nodes if node["class_type"] == "easy minimaxH3ToVideo" and node["inputs"]["length"] == [math_id, 1] ) assert math_node["inputs"]["expression"] == "a + 34" task_length_link = math_node["inputs"]["values.a"] assert task_length_link == 124 assert "a" not in math_node["inputs"] assert context_conditioning["inputs"]["length"] == [math_id, 1] trim = next( node for node in nodes if node["class_type"] == "easy h3ContextMediaTrim" and node["inputs"]["trim_frames"] == [motion_id, 1] ) assert trim["inputs"]["trim_frames"] == [motion_id, 1] assert trim["inputs"]["output_frames"] == 120 assert trim["inputs"]["pad_audio"] is False assert "fit_video_duration" not in trim["inputs"] trim_id = next( node_id for node_id, node in result.expand.items() if node is trim ) context_align = next( node for node in nodes if node["class_type"] == "easy h3LockedAudioDurationAlign" and node["inputs"]["audio"] == [trim_id, 1] ) assert context_align["inputs"]["images"] == [trim_id, 0] context_video = next( node for node in nodes if node["class_type"] == "easy saveVideo" and node["inputs"]["input_mode.images"] == [trim_id, 0] ) # The task audio already excludes the context prefix; do not trim it again. context_selector = result.expand[ context_video["inputs"]["input_mode.audio"][0] ] assert context_selector["class_type"] == "easy h3LockedAudioSelect" assert context_selector["inputs"]["locked_audio"] == { "prepared_locked_audio": True, } artifacts = [ (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy h3ProjectArtifact" ] assert len(artifacts) == 2 assert artifacts[1][1]["inputs"]["previous"] == [artifacts[0][0], 0] def test_multitrack_h3_context_uses_embedded_five_frame_anchor(monkeypatch): module = _load_minimax_node(monkeypatch) info = _h3_project_inputs()["tracks_info"][0] info["tracks"][0]["segments"].append( { "start_frame": 120, "end_frame": 240, "content": { "task_mode": "default", "continuity_mode": "context", "images": [], "user_prompt": "test direct anchor", }, } ) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs(tracks_info=[info]) ) motion = next( node for node in result.expand.values() if node["class_type"] == "easy MiniMaxH3MotionContextHard" ) context_id, context_output = motion["inputs"]["context_latent"] assert context_output == 0 context_trim = result.expand[context_id] assert context_trim["class_type"] == "easy h3MotionContextLatentTrim" assert set(context_trim["inputs"]) == { "latent", "context_length", "anchor_images", "vae", } first_artifact = next( node for node_id, node in result.expand.items() if "artifact_0" in node_id and node["class_type"] == "easy h3ProjectArtifact" ) assert first_artifact["inputs"]["context_latent"] == [context_id, 0] assert "anchor_latent" not in first_artifact["inputs"] assert not any( node["class_type"] in { "easy h3VideoAnchorEncode", "easy MiniMaxH3MotionContextAnchor", } for node in result.expand.values() ) def test_multitrack_h3_partial_shot_run_embeds_anchor_in_context_latent( monkeypatch, ): module = _load_minimax_node(monkeypatch) info = _h3_project_inputs()["tracks_info"][0] info["tracks"][0]["segments"].append( { "start_frame": 120, "end_frame": 240, "content": { "task_mode": "default", "continuity_mode": "context", "images": [], "user_prompt": "generate later", }, } ) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( tracks_info=[info], segment_start_number=[1], segment_count=[1], ) ) context_id, context_trim = next( (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy h3MotionContextLatentTrim" ) artifact = next( node for node in result.expand.values() if node["class_type"] == "easy h3ProjectArtifact" ) assert artifact["inputs"]["context_latent"] == [context_id, 0] assert "anchor_images" in context_trim["inputs"] assert "vae" in context_trim["inputs"] assert "anchor_latent" not in artifact["inputs"] def test_multitrack_h3_first_context_task_does_not_add_an_empty_prefix(monkeypatch): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs() info = inputs["tracks_info"][0] info["tracks"][0]["segments"][0]["content"]["continuity_mode"] = "context" info["tracks"].append( { "type": "audio", "audio_locked": True, "segments": [ { "start_frame": 0, "end_frame": 120, "content": {"media_type": "audio"}, } ], } ) result = module.EasyMultiTrackProject.execute(**inputs) nodes = list(result.expand.values()) conditioning = next( node for node in nodes if node["class_type"] == "easy minimaxH3ToVideo" ) audio_lock = next( node for node in nodes if node["class_type"] == "easy minimaxH3AudioLock" ) assert conditioning["inputs"]["length"] == 124 assert audio_lock["inputs"]["prepend_frames"] == 0 delivery_trim = next( node for node in nodes if node["class_type"] == "easy h3ContextMediaTrim" and not node["inputs"].get("phase_align_video_encode") ) assert delivery_trim["inputs"]["output_frames"] == 120 assert "fit_video_duration" not in delivery_trim["inputs"] assert not any( node["class_type"] in { "ComfyMathExpression", "easy MiniMaxH3MotionContextHard", } for node in nodes ) def test_multitrack_h3_dual_context_uses_separate_low_and_hires_latents(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None module.comfy_nodes.NODE_CLASS_MAPPINGS.update( { "ImageResizeKJv2": _ImageResizeKJWithNvidia, "MiniMaxH3MotionContextTrim": _MiniMaxMotionContextTrim, } ) info = _h3_project_inputs()["tracks_info"][0] info["tracks"][0]["segments"].append( { "start_frame": 120, "end_frame": 240, "content": { "task_mode": "l2v", "continuity_mode": "context", "images": [{"media_index": 0}], "user_prompt": "continue", }, } ) info["tracks"][0]["segments"].append( { "start_frame": 240, "end_frame": 360, "content": { "task_mode": "default", "continuity_mode": "shot", "images": [], "user_prompt": "cut to another shot", }, } ) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( tracks_info=[info], sampling_mode=["dual"], sampling_plan=["light"], ) ) samples = [ (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "SamplerCustomAdvanced" ] motion = next( node for node in result.expand.values() if node["class_type"] == "easy MiniMaxH3MotionContextHard" ) low_trim_id, low_trim = next( (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy h3MotionContextLatentTrim" and node["inputs"]["latent"] == [samples[0][0], 1] ) assert motion["inputs"]["context_latent"] == [low_trim_id, 0] assert low_trim["inputs"]["context_length"] == "22" hires = next( node for node in result.expand.values() if node["class_type"] == "easy MiniMaxH3HiResContinuity" ) high_trim_id, high_trim = next( (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy h3MotionContextLatentTrim" and node["inputs"]["latent"] == [samples[1][0], 1] ) assert hires["inputs"]["previous_hires_latent"] == [high_trim_id, 0] assert high_trim["inputs"]["context_length"] == "22" motion_id = next( node_id for node_id, node in result.expand.items() if node is motion ) low_context_conditioning_id = motion["inputs"]["conditioning"][0] low_context_conditioning = result.expand[low_context_conditioning_id] encoded_low_conditioning = result.expand[ low_context_conditioning["inputs"]["conditioning"][0] ] hires_context_conditioning_id = next( node_id for node_id, node in result.expand.items() if node["class_type"] == "easy minimaxH3ToVideo" and node["inputs"]["prompt"] == encoded_low_conditioning["inputs"]["prompt"] and node["inputs"]["width"] == 1664 and node["inputs"]["height"] == 960 ) context_guiders = [ (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "BasicGuider" and tuple(node["inputs"]["conditioning"]) in { (motion_id, 0), (hires_context_conditioning_id, 0), } ] assert len(context_guiders) == 2 first_context_guider = next( (node_id, node) for node_id, node in context_guiders if node["inputs"]["conditioning"] == [motion_id, 0] ) second_context_guider = next( (node_id, node) for node_id, node in context_guiders if node["inputs"]["conditioning"] == [hires_context_conditioning_id, 0] ) second_sampling_start = [ node for node in result.expand.values() if node["class_type"] == "easy h3SegmentSamplingStart" and node["inputs"]["sampling_pass"] == "second" and node["inputs"]["segment_index"] == 1 ][0] first_second_pass_start = [ node for node in result.expand.values() if node["class_type"] == "easy h3SegmentSamplingStart" and node["inputs"]["sampling_pass"] == "second" and node["inputs"]["segment_index"] == 0 ][0] final_second_pass_start = [ node for node in result.expand.values() if node["class_type"] == "easy h3SegmentSamplingStart" and node["inputs"]["sampling_pass"] == "second" and node["inputs"]["segment_index"] == 2 ][0] normal_sigma_node = result.expand[first_second_pass_start["inputs"]["sigmas"][0]] context_sigma_node = result.expand[second_sampling_start["inputs"]["sigmas"][0]] assert normal_sigma_node["class_type"] == "ManualSigmas" assert normal_sigma_node["inputs"]["sigmas"].startswith("0.6316, 0.4877") assert context_sigma_node["class_type"] == "ManualSigmas" assert context_sigma_node["inputs"]["sigmas"] == ( "0.50, 0.30, 0.14, 0.06, 0.0" ) assert final_second_pass_start["inputs"]["sigmas"] == ( first_second_pass_start["inputs"]["sigmas"] ) assert first_context_guider[1]["inputs"]["conditioning"] == [motion_id, 0] assert second_sampling_start["inputs"]["guider"] == [ second_context_guider[0], 0, ] second_segment_artifact = [ node for node in result.expand.values() if node["class_type"] == "easy h3ProjectArtifact" ][1] hires_context_link = second_segment_artifact["inputs"]["context_latent"] low_context_link = second_segment_artifact["inputs"]["context_latent_low"] hires_trim = result.expand[hires_context_link[0]] assert hires_trim["class_type"] == "easy h3MotionContextLatentTrim" assert result.expand[hires_trim["inputs"]["latent"][0]]["class_type"] == ( "LTXVConcatAVLatent" ) low_context_trim = result.expand[low_context_link[0]] assert low_context_trim["class_type"] == "easy h3MotionContextLatentTrim" assert "anchor_images" not in low_context_trim["inputs"] assert hires_context_link != low_context_link context_concat_links = [ hires_trim["inputs"]["latent"], low_context_trim["inputs"]["latent"], ] for concat_link in context_concat_links: concat = result.expand[concat_link[0]] video_encode = result.expand[concat["inputs"]["video_latent"][0]] video_trim = result.expand[video_encode["inputs"]["pixels"][0]] audio_encode = result.expand[concat["inputs"]["audio_latent"][0]] assert video_encode["class_type"] == "VAEEncode" assert video_trim["class_type"] == "easy h3ContextMediaTrim" assert video_trim["inputs"]["output_frames"] == 22 assert video_trim["inputs"]["phase_align_video_encode"] is True assert audio_encode["class_type"] == "VAEEncodeAudio" assert audio_encode["inputs"]["audio"] == [ video_encode["inputs"]["pixels"][0], 1, ] def test_multitrack_h3_context_swap_uses_drift_control_only_in_first_pass( monkeypatch, ): module = _load_minimax_node(monkeypatch) module.comfy_nodes.NODE_CLASS_MAPPINGS.update( { "ImageResizeKJv2": _ImageResizeKJWithNvidia, "MiniMaxH3MotionContextTrim": _MiniMaxMotionContextTrim, } ) info = _h3_project_inputs()["tracks_info"][0] info["tracks"][0]["segments"].append( { "start_frame": 120, "end_frame": 240, "content": { "task_mode": "ref", "continuity_mode": "context_swap", "images": [{"media_index": 0}], "user_prompt": "replace the character and preserve motion", }, } ) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( tracks_info=[info], sampling_mode=["dual"], sampling_plan=["light"], ) ) noise_nodes = [ (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy MiniMaxH3ContextSwap" ] assert len(noise_nodes) == 1 first_noise_id, first_noise = noise_nodes[0] assert "first_pass_context_swap_noise" in first_noise_id hires = _graph_node(result, "easy MiniMaxH3HiResContinuity") assert first_noise["inputs"]["context_length"] == "22" assert not any( node["class_type"] == "easy MiniMaxH3MotionContextHard" for node in result.expand.values() ) artifacts = [ node for node in result.expand.values() if node["class_type"] == "easy h3ProjectArtifact" ] low_context_id = first_noise["inputs"]["context_latent"][0] low_context = result.expand[low_context_id] assert low_context["class_type"] == "easy h3MotionContextLatentTrim" assert artifacts[0]["inputs"]["context_latent_low"] == [low_context_id, 0] assert result.expand[low_context["inputs"]["latent"][0]]["class_type"] == ( "SamplerCustomAdvanced" ) assert hires["inputs"]["previous_hires_latent"] == artifacts[0]["inputs"][ "context_latent" ] assert artifacts[1]["inputs"]["continuity_mode"] == "context_swap" assert all( artifact["inputs"]["context_latent"] != [first_noise_id, 0] for artifact in artifacts ) def test_multitrack_h3_connected_second_pass_sampling_overrides_context_preset( monkeypatch, ): module = _load_minimax_node(monkeypatch) info = _h3_project_inputs()["tracks_info"][0] info["tracks"][0]["segments"].append( { "start_frame": 120, "end_frame": 240, "content": { "task_mode": "l2v", "continuity_mode": "context", "images": [{"media_index": 0}], "user_prompt": "continue", }, } ) second_pass_sampler = object() second_pass_sigmas = torch.tensor([0.4, 0.2, 0.0]) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( tracks_info=[info], sampling_plan=["medium"], sampling_mode=_h3_sampling_mode( "dual", sampler_2nd=[second_pass_sampler], sigmas_2nd=[second_pass_sigmas], upscale_by=[1.0], ), ) ) second_pass_starts = [ node for node in result.expand.values() if node["class_type"] == "easy h3SegmentSamplingStart" and node["inputs"]["sampling_pass"] == "second" ] assert len(second_pass_starts) == 2 assert all( node["inputs"]["sampler"] is second_pass_sampler and node["inputs"]["sigmas"] is second_pass_sigmas for node in second_pass_starts ) assert not any( node["class_type"] == "ManualSigmas" and node["inputs"]["sigmas"] == "0.50, 0.30, 0.14, 0.06, 0.0" for node in result.expand.values() ) def test_multitrack_h3_consecutive_context_uses_previous_trimmed_latent(monkeypatch): module = _load_minimax_node(monkeypatch) info = _h3_project_inputs()["tracks_info"][0] task_track = info["tracks"][0] task_track["segments"].extend( [ { "start_frame": 120, "end_frame": 240, "content": { "task_mode": "l2v", "continuity_mode": "context", "images": [], "user_prompt": "continue once", }, }, { "start_frame": 240, "end_frame": 360, "content": { "task_mode": "l2v", "continuity_mode": "context", "images": [], "user_prompt": "continue twice", }, }, ] ) info["tracks"].append( { "type": "audio", "audio_locked": True, "segments": [ { "start_frame": 0, "end_frame": 360, "content": {"media_type": "audio"}, } ], } ) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs(tracks_info=[info]) ) motions = [ node for node in result.expand.values() if node["class_type"] == "easy MiniMaxH3MotionContextHard" ] artifacts = [ node for node in result.expand.values() if node["class_type"] == "easy h3ProjectArtifact" ] assert len(motions) == 2 assert len(artifacts) == 3 first_context_link = artifacts[0]["inputs"]["context_latent"] first_context_trim = result.expand[first_context_link[0]] assert first_context_trim["class_type"] == ( "easy h3MotionContextLatentTrim" ) first_context_source = result.expand[first_context_trim["inputs"]["latent"][0]] assert first_context_source["class_type"] == "LTXVConcatAVLatent" assert motions[0]["inputs"]["context_latent"] == first_context_link second_context_link = artifacts[1]["inputs"]["context_latent"] assert result.expand[second_context_link[0]]["class_type"] == ( "easy h3MotionContextLatentTrim" ) assert motions[1]["inputs"]["context_latent"] == second_context_link second_context_trim = result.expand[second_context_link[0]] second_context_source = result.expand[second_context_trim["inputs"]["latent"][0]] assert second_context_source["class_type"] == "LTXVConcatAVLatent" assert [ node["inputs"]["prepend_frames"] for node in result.expand.values() if node["class_type"] == "easy minimaxH3AudioLock" ] == [0, 22, 22] def test_multitrack_h3_loop_start_loads_previous_project_context(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None project_module = sys.modules["easy_media.nodes.project"] monkeypatch.setattr( project_module, "has_h3_context_latent", lambda *args, **kwargs: True ) module.comfy_nodes.NODE_CLASS_MAPPINGS.update( { "MiniMaxH3MotionContextTrim": _MiniMaxMotionContextTrim, } ) info = _h3_project_inputs()["tracks_info"][0] info["tracks"][0]["segments"].append( { "start_frame": 120, "end_frame": 240, "content": { "task_mode": "l2v", "continuity_mode": "context", "images": [{"media_index": 0}], "user_prompt": "continue", }, } ) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( tracks_info=[info], segment_start_number=[2], segment_count=[1], ) ) nodes = list(result.expand.values()) load = next( node for node in nodes if node["class_type"] == "easy h3ProjectContextLatentLoad" ) assert load["inputs"]["segment_index"] == 0 assert load["inputs"]["resolution"] == "high" assert any( node["class_type"] == "easy MiniMaxH3MotionContextHard" for node in nodes ) def test_multitrack_h3_context_loop_start_loads_single_saved_context(monkeypatch): module = _load_minimax_node(monkeypatch) project_module = sys.modules["easy_media.nodes.project"] monkeypatch.setattr( project_module, "has_h3_context_latent", lambda *args, **kwargs: True ) info = _h3_project_inputs()["tracks_info"][0] info["tracks"][0]["segments"].append( { "start_frame": 120, "end_frame": 240, "content": { "task_mode": "default", "continuity_mode": "context", "images": [], "user_prompt": "resume direct anchor", }, } ) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( tracks_info=[info], segment_start_number=[2], segment_count=[1], ) ) loads = { node["inputs"]["resolution"]: (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy h3ProjectContextLatentLoad" } assert set(loads) == {"high"} motion = next( node for node in result.expand.values() if node["class_type"] == "easy MiniMaxH3MotionContextHard" ) assert motion["inputs"]["context_latent"] == [loads["high"][0], 0] def test_multitrack_h3_context_swap_loop_start_uses_saved_high_context(monkeypatch): module = _load_minimax_node(monkeypatch) project_module = sys.modules["easy_media.nodes.project"] monkeypatch.setattr( project_module, "has_h3_context_latent", lambda *args, **kwargs: True ) info = _h3_project_inputs()["tracks_info"][0] info["tracks"][0]["segments"].append( { "start_frame": 120, "end_frame": 240, "content": { "task_mode": "default", "continuity_mode": "context_swap", "images": [], "user_prompt": "resume swap", }, } ) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( tracks_info=[info], segment_start_number=[2], segment_count=[1], ) ) loads = [ (node_id, node) for node_id, node in result.expand.items() if node["class_type"] == "easy h3ProjectContextLatentLoad" ] assert len(loads) == 1 assert loads[0][1]["inputs"]["resolution"] == "high" swap = next( node for node in result.expand.values() if node["class_type"] == "easy MiniMaxH3ContextSwap" ) assert swap["inputs"]["context_latent"] == [loads[0][0], 0] def test_multitrack_h3_context_start_rejects_missing_previous_latent(monkeypatch): module = _load_minimax_node(monkeypatch) project_module = sys.modules["easy_media.nodes.project"] monkeypatch.setattr( project_module, "has_h3_context_latent", lambda *args, **kwargs: False ) info = _h3_project_inputs()["tracks_info"][0] info["tracks"][0]["segments"].append( { "start_frame": 120, "end_frame": 240, "content": { "task_mode": "l2v", "continuity_mode": "context", "images": [{"media_index": 0}], }, } ) with pytest.raises(ValueError, match="segment 1 has no active context latent"): module.EasyMultiTrackProject.execute( **_h3_project_inputs( tracks_info=[info], segment_start_number=[2], segment_count=[1], ) ) def test_multitrack_h3_selflift_context_start_requires_exact_low_latent(monkeypatch): module = _load_minimax_node(monkeypatch) project_module = sys.modules["easy_media.nodes.project"] checked_resolutions = [] def has_context(*args, **kwargs): checked_resolutions.append( (kwargs.get("resolution"), kwargs.get("allow_low_fallback", True)) ) return kwargs.get("resolution") == "high" monkeypatch.setattr(project_module, "has_h3_context_latent", has_context) info = _h3_project_inputs()["tracks_info"][0] info["tracks"][0]["segments"].append( { "start_frame": 120, "end_frame": 240, "content": { "task_mode": "l2v", "continuity_mode": "context_swap", "images": [{"media_index": 0}], }, } ) with pytest.raises(ValueError, match="no active low-resolution context latent"): module.EasyMultiTrackProject.execute( **_h3_project_inputs( tracks_info=[info], sampling_mode=_h3_sampling_mode("selflift"), segment_start_number=[2], segment_count=[1], ) ) assert checked_resolutions == [("high", True), ("low", False)] def test_multitrack_h3_selflift_context_swap_start_uses_both_saved_resolutions( monkeypatch, ): module = _load_minimax_node(monkeypatch) project_module = sys.modules["easy_media.nodes.project"] monkeypatch.setattr( project_module, "has_h3_context_latent", lambda *args, **kwargs: True ) info = _h3_project_inputs()["tracks_info"][0] info["tracks"][0]["segments"].append( { "start_frame": 120, "end_frame": 240, "content": { "task_mode": "default", "continuity_mode": "context_swap", "images": [], }, } ) result = module.EasyMultiTrackProject.execute( **_h3_project_inputs( tracks_info=[info], sampling_mode=_h3_sampling_mode("selflift"), segment_start_number=[2], segment_count=[1], ) ) loads = { node["inputs"]["resolution"]: node_id for node_id, node in result.expand.items() if node["class_type"] == "easy h3ProjectContextLatentLoad" } assert set(loads) == {"high", "low"} swap = _graph_node(result, "easy MiniMaxH3ContextSwap") selflift = _graph_node(result, "easy minimaxH3SelfLiftSampler") assert swap["inputs"]["context_latent"] == [loads["high"], 0] assert selflift["inputs"]["low_context_latent"] == [loads["low"], 0] def test_multitrack_h3_project_uses_prompt_graph_as_last_turbo_fallback( monkeypatch, capsys ): module = _load_minimax_node(monkeypatch) assert module is not None prompt = { "3": { "class_type": "easy multitrackProject", "inputs": {"model_loader": ["2", 0]}, }, "2": { "class_type": "easy modelLoaderPack", "inputs": {"model": ["1", 0]}, }, "1": { "class_type": "LoraLoaderModelOnly", "inputs": {"lora_name": "minimax_h3_turbo_4step.safetensors"}, }, } graph_result = types.SimpleNamespace( is_turbo=True, as_dict=lambda: { "status": "turbo", "is_turbo": True, "source": "graph_prompt", "evidence": "node 1", "patch_count": 0, }, ) graph_calls = [] monkeypatch.setattr( module._project_module, "detect_turbo_lora_from_prompt", lambda received_prompt, node_id: graph_calls.append( (received_prompt, node_id) ) or graph_result, ) module.EasyMultiTrackProject.hidden = types.SimpleNamespace( prompt=prompt, unique_id="3", ) module.EasyMultiTrackProject.execute(**_h3_project_inputs()) assert graph_calls == [(prompt, "3")] assert graph_calls == [(prompt, "3")] def test_multitrack_h3_project_skips_prompt_graph_after_model_turbo_match( monkeypatch, ): module = _load_minimax_node(monkeypatch) assert module is not None model_result = types.SimpleNamespace( is_turbo=True, as_dict=lambda: { "status": "turbo", "is_turbo": True, "source": "model_patches", "evidence": "patch match", "patch_count": 1, }, ) monkeypatch.setattr( module._project_module, "detect_turbo_model", lambda model: model_result, ) def fail_graph_fallback(prompt, node_id): raise AssertionError("graph fallback must not run after a model match") monkeypatch.setattr( module._project_module, "detect_turbo_lora_from_prompt", fail_graph_fallback ) module.EasyMultiTrackProject.execute(**_h3_project_inputs()) def test_multitrack_h3_project_requires_a_model_loader_dictionary(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None with pytest.raises(TypeError, match="FAST_MODEL_LOADER dictionary"): module.EasyMultiTrackProject.execute( model_loader=[], tracks_info=[{}], ) def test_multitrack_h3_project_requires_core_model_components(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None with pytest.raises(ValueError, match="clip, vae"): module.EasyMultiTrackProject.execute( model_loader=[{"model": object()}], tracks_info=[{}], ) def test_multitrack_project_rejects_non_minimax_h3_model(monkeypatch): module = _load_minimax_node(monkeypatch) with pytest.raises(ValueError, match="supports only MiniMaxH3"): module.EasyMultiTrackProject.execute( **_h3_project_inputs(model_loader=[{ "model": object(), "clip": object(), "vae": object(), "audio_vae": object(), }]) ) def test_h3_project_artifact_writes_manifest_and_rotates_ten_generations( monkeypatch, tmp_path ): module = _load_minimax_node(monkeypatch) assert module is not None monkeypatch.setattr( module.folder_paths, "get_output_directory", lambda: str(tmp_path), ) info = _h3_project_inputs()["tracks_info"][0] project_dir = tmp_path / "easy_media" / "projects" / "demo" project_dir.mkdir(parents=True) for run in range(12): staged = project_dir / f".staged_{run}.mp4" staged.write_bytes(f"video-{run}".encode()) high_context = _h3_context_latent(run, video_steps=12) high_context["anchor_samples"] = _h3_video_anchor_latent(run)["samples"] low_context = _h3_context_latent(-run, video_steps=12) low_context["anchor_samples"] = _h3_video_anchor_latent(-run)["samples"] output = module.EasyH3ProjectArtifact.execute( project_name="demo", project_save="new", segment_index=0, context_latent=high_context, context_latent_low=low_context, video_path=f"output/{staged.relative_to(tmp_path)}", tracks_info=info, seed=0xFFFFFFFFFFFFFFFF, ) assert output.values == ("demo",) assert len(list(project_dir.glob("video_0_*.mp4"))) == 10 assert not list(project_dir.glob("latent_0_*")) assert len(list(project_dir.glob("context_latent_0_*.safetensors"))) == 10 assert len(list(project_dir.glob("context_latent_low_0_*.safetensors"))) == 10 assert not list(project_dir.glob("anchor_latent_0_*.safetensors")) assert not list(project_dir.glob("anchor_latent_low_0_*.safetensors")) manifest = json.loads((project_dir / "project.json").read_text()) assert manifest["version"] == 2 assert manifest["project_name"] == "demo" assert (manifest["width"], manifest["height"], manifest["fps"]) == ( 1344, 768, 24.0, ) assert "tracks_info" not in manifest assert manifest["task_segments"] == [ { "index": 0, "continuity_mode": "shot", "task_mode": "default", "audio_locked": False, }, ] assert manifest["segments"]["0"]["task_mode"] == "default" assert len(manifest["segments"]["0"]["generations"]) == 10 loaded = module.EasyH3ProjectContextLatentLoad.execute("demo", 0) loaded_streams = loaded.values[0]["samples"].unbind() assert len(loaded_streams) == 2 assert all(isinstance(stream, torch.Tensor) for stream in loaded_streams) assert loaded_streams[0].shape[2] == 7 assert loaded_streams[1].shape[-1] == 37 assert loaded.values[0]["anchor_samples"].shape == (1, 24, 2, 2, 2) loaded_low = module.EasyH3ProjectContextLatentLoad.execute( "demo", 0, resolution="low" ) loaded_low_streams = loaded_low.values[0]["samples"].unbind() assert loaded_low_streams[0].shape[2] == 12 assert loaded_low_streams[1].shape[-1] == 65 assert loaded_low.values[0]["anchor_samples"].shape == (1, 24, 2, 2, 2) active_generation = str(manifest["segments"]["0"]["active_generation"]) active_files = manifest["segments"]["0"]["generations"][active_generation] assert active_files["context_latent_low"].startswith("context_latent_low_0_") assert "anchor_latent" not in active_files assert "anchor_latent_low" not in active_files assert active_files["continuity_mode"] == "shot" assert active_files["seed"] == 0xFFFFFFFFFFFFFFFF def test_h3_project_artifact_notifies_only_after_new_video_is_in_manifest( monkeypatch, tmp_path ): module = _load_minimax_node(monkeypatch) monkeypatch.setattr( module.folder_paths, "get_output_directory", lambda: str(tmp_path), ) project_dir = tmp_path / "easy_media" / "projects" / "demo" project_dir.mkdir(parents=True) staged = project_dir / ".staged.mp4" staged.write_bytes(b"video") notifications = [] def send_sync(event, payload): manifest = json.loads((project_dir / "project.json").read_text()) active_generation = str( manifest["segments"]["0"]["active_generation"] ) video_name = manifest["segments"]["0"]["generations"][ active_generation ]["video"] notifications.append((event, payload, (project_dir / video_name).is_file())) server = types.ModuleType("server") server.PromptServer = types.SimpleNamespace( instance=types.SimpleNamespace(send_sync=send_sync) ) monkeypatch.setitem(sys.modules, "server", server) module.EasyH3ProjectArtifact.execute( project_name="demo", project_save="new", segment_index=0, context_latent=_h3_context_latent(), video_path=f"output/{staged.relative_to(tmp_path)}", tracks_info=_h3_project_inputs()["tracks_info"][0], ) assert notifications == [ ( "easy_multitrack_project_refresh", {"project_name": "demo", "phase": "after_save", "segment_index": 0}, True, ) ] def test_h3_project_artifact_preserves_complete_first_pass_checkpoint( monkeypatch, tmp_path ): module = _load_minimax_node(monkeypatch) monkeypatch.setattr( module.folder_paths, "get_output_directory", lambda: str(tmp_path), ) info = _h3_project_inputs()["tracks_info"][0] project_dir = tmp_path / "easy_media" / "projects" / "demo" project_dir.mkdir(parents=True) staged = project_dir / ".first-pass.mp4" staged.write_bytes(b"first-pass") module.EasyH3ProjectArtifact.execute( project_name="demo", project_save="new", segment_index=0, context_latent=_h3_context_latent(1, video_steps=12), context_latent_low=_h3_context_latent(1, video_steps=12), video_path=f"output/{staged.relative_to(tmp_path)}", tracks_info=info, sampling_pass="first", ) high = module.EasyH3ProjectContextLatentLoad.execute("demo", 0).values[0] low = module.EasyH3ProjectContextLatentLoad.execute( "demo", 0, resolution="low" ).values[0] for latent in (high, low): video, audio = latent["samples"].unbind() assert video.shape[2] == 12 assert audio.shape[-1] == 65 def test_h3_project_context_round_trip_matches_runtime_context(monkeypatch, tmp_path): module = _load_minimax_node(monkeypatch) monkeypatch.setattr( module.folder_paths, "get_output_directory", lambda: str(tmp_path), ) project_dir = tmp_path / "easy_media" / "projects" / "demo" project_dir.mkdir(parents=True) staged = project_dir / ".context-round-trip.mp4" staged.write_bytes(b"video") high = module.trim_motion_context_latent( _h3_context_latent(3, video_steps=12) ) low = module.trim_motion_context_latent( _h3_context_latent(7, video_steps=12) ) high["anchor_samples"] = _h3_video_anchor_latent(3)["samples"] low["anchor_samples"] = _h3_video_anchor_latent(7)["samples"] module.EasyH3ProjectArtifact.execute( project_name="demo", project_save="new", segment_index=0, context_latent=high, context_latent_low=low, video_path=f"output/{staged.relative_to(tmp_path)}", tracks_info=_h3_project_inputs()["tracks_info"][0], sampling_pass="single", ) for resolution, runtime in (("high", high), ("low", low)): loaded = module.EasyH3ProjectContextLatentLoad.execute( "demo", 0, resolution=resolution ).values[0] for actual, expected in zip( loaded["samples"].unbind(), runtime["samples"].unbind() ): assert torch.equal(actual, expected) assert torch.equal(loaded["anchor_samples"], runtime["anchor_samples"]) def test_h3_project_artifact_override_reuses_latest_generation(monkeypatch, tmp_path): module = _load_minimax_node(monkeypatch) assert module is not None monkeypatch.setattr( module.folder_paths, "get_output_directory", lambda: str(tmp_path), ) info = _h3_project_inputs()["tracks_info"][0] project_dir = tmp_path / "easy_media" / "projects" / "default" project_dir.mkdir(parents=True) for run, project_save in enumerate(("new", "new", "override")): staged = project_dir / f".override_{run}.mp4" staged.write_bytes(f"video-{run}".encode()) module.EasyH3ProjectArtifact.execute( project_name="", project_save=project_save, segment_index=3, context_latent=_h3_context_latent(run), video_path=f"output/{staged.relative_to(tmp_path)}", tracks_info=info, ) assert sorted(path.name for path in project_dir.glob("video_3_*.mp4")) == [ "video_3_1.mp4", "video_3_2.mp4", ] assert (project_dir / "video_3_1.mp4").read_bytes() == b"video-0" assert (project_dir / "video_3_2.mp4").read_bytes() == b"video-2" assert not list(project_dir.glob("latent_3_*")) assert sorted( path.name for path in project_dir.glob("context_latent_3_*.safetensors") ) == [ "context_latent_3_1.safetensors", "context_latent_3_2.safetensors", ] manifest = json.loads((project_dir / "project.json").read_text()) assert manifest["segments"]["3"]["active_generation"] == 2 assert set(manifest["segments"]["3"]["generations"]) == {"1", "2"} @pytest.mark.parametrize("project_save", ["new", "override"]) def test_h3_project_artifact_uses_embedded_audio_without_sidecar( monkeypatch, tmp_path, project_save, ): module = _load_minimax_node(monkeypatch) monkeypatch.setattr( module.folder_paths, "get_output_directory", lambda: str(tmp_path), ) info = _h3_project_inputs()["tracks_info"][0] project_dir = tmp_path / "easy_media" / "projects" / "demo" project_dir.mkdir(parents=True) if project_save == "override": (project_dir / "locked_audio_0_1.wav").write_bytes(b"legacy-audio") (project_dir / "project.json").write_text(json.dumps({ "segments": {"0": {"generations": {"1": { "locked_audio": "locked_audio_0_1.wav", }}}}, })) staged = project_dir / ".staged.mp4" staged.write_bytes(b"video") module.EasyH3ProjectArtifact.execute( project_name="demo", project_save=project_save, segment_index=0, context_latent=_h3_context_latent(1), video_path=f"output/{staged.relative_to(tmp_path)}", tracks_info=info, ) manifest = json.loads((project_dir / "project.json").read_text()) generation = manifest["segments"]["0"]["generations"]["1"] assert "locked_audio" not in generation assert not list(project_dir.glob("locked_audio_*.wav")) assert (project_dir / generation["video"]).read_bytes() == b"video" def test_reference_bridge_uses_fixed_inputs_instead_of_autogrow(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None schema = module.EasyMiniMaxH3ReferenceToVideoBridge.define_schema() input_names = [port.name for port in schema.inputs] assert schema.node_id == module.REFERENCE_BRIDGE_NODE_ID assert "ref_images" not in input_names assert input_names[-18:] == [ *[f"ref_image_{index}" for index in range(9)], *[f"ref_video_{index}" for index in range(3)], *[f"ref_video_audio_{index}" for index in range(3)], *[f"ref_audio_{index}" for index in range(3)], ] def test_multi_frames_routes_first_and_last_expanded_images(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None clip = _Clip() vae = _Vae() output = module.EasyMiniMaxH3ToVideo.execute( **_base_inputs( clip=[clip], vae=[vae], images=[[_image_values(0, 1)], _image_values(2)], ) ) conditioning = _graph_node(output, "MiniMaxH3ImageToVideo") assert conditioning["inputs"]["first_frame"][0, 0, 0, 0].item() == 0 assert conditioning["inputs"]["last_frame"][0, 0, 0, 0].item() == 2 def test_multi_frames_with_one_image_routes_only_first_frame(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None output = module.EasyMiniMaxH3ToVideo.execute( **_base_inputs(images=[_image_values(7)]) ) conditioning = _graph_node(output, "MiniMaxH3ImageToVideo") assert "first_frame" in conditioning["inputs"] assert "last_frame" not in conditioning["inputs"] def test_last_frame_routes_only_last_expanded_image(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None output = module.EasyMiniMaxH3ToVideo.execute( **_base_inputs( mode=["last_frame"], images=[[_image_values(0, 1)], _image_values(2)], ) ) conditioning = _graph_node(output, "MiniMaxH3ImageToVideo") assert "first_frame" not in conditioning["inputs"] assert conditioning["inputs"]["last_frame"][0, 0, 0, 0].item() == 2 def test_last_frame_with_one_image_routes_only_last_frame(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None output = module.EasyMiniMaxH3ToVideo.execute( **_base_inputs(mode=["last_frame"], images=[_image_values(7)]) ) conditioning = _graph_node(output, "MiniMaxH3ImageToVideo") assert "first_frame" not in conditioning["inputs"] assert conditioning["inputs"]["last_frame"].shape == (1, 1, 1, 1) assert conditioning["inputs"]["last_frame"].item() == 7 @pytest.mark.parametrize("mode", ["reference", "multi_frames", "last_frame"]) def test_empty_media_routes_to_text_to_video_for_every_mode(monkeypatch, mode): module = _load_minimax_node(monkeypatch) assert module is not None clip = _Clip() output = module.EasyMiniMaxH3ToVideo.execute( **_base_inputs(clip=[clip], mode=[mode]) ) conditioning = _graph_node(output, "MiniMaxH3ImageToVideo") assert conditioning["inputs"]["prompt"] == "prompt" assert "first_frame" not in conditioning["inputs"] assert "last_frame" not in conditioning["inputs"] @pytest.mark.parametrize( ("mode", "media_kind"), [ ("multi_frames", "video"), ("multi_frames", "audio"), ("last_frame", "video"), ("last_frame", "audio"), ], ) def test_frame_modes_with_video_or_audio_route_to_reference_subgraph( monkeypatch, mode, media_kind ): module = _load_minimax_node(monkeypatch) assert module is not None audio = {"waveform": torch.ones(1, 1, 4), "sample_rate": 32000} overrides = { "audio_vae": [_AudioVae()], "images": [_image_values(1, 2)], "videos": [object()] if media_kind == "video" else [], "audios": [audio] if media_kind == "audio" else [], } output = module.EasyMiniMaxH3ToVideo.execute( **_base_inputs(mode=[mode], **overrides) ) nodes_by_type = {node["class_type"]: node for node in output.expand.values()} conditioning = nodes_by_type[module.REFERENCE_BRIDGE_NODE_ID] assert "MiniMaxH3ImageToVideo" not in nodes_by_type assert "ref_image_0" in conditioning["inputs"] assert "ref_image_1" in conditioning["inputs"] if media_kind == "video": assert "GetVideoComponents" in nodes_by_type assert "ref_video_0" in conditioning["inputs"] else: assert conditioning["inputs"]["ref_audio_0"] is audio def test_reference_video_extraction_is_deferred_to_a_cacheable_subnode(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None video = object() output = module.EasyMiniMaxH3ToVideo.execute( **_base_inputs( mode=["reference"], audio_vae=[_AudioVae()], videos=[video], ) ) assert output.expand is not None nodes_by_type = {node["class_type"]: node for node in output.expand.values()} components = nodes_by_type["GetVideoComponents"] conditioning = nodes_by_type[module.REFERENCE_BRIDGE_NODE_ID] components_id = next( node_id for node_id, node in output.expand.items() if node["class_type"] == "GetVideoComponents" ) assert components["inputs"] == {"video": video} assert "easy minimaxH3ResampleVideoFrames" not in nodes_by_type assert conditioning["inputs"]["ref_video_0"] == [components_id, 0] assert conditioning["inputs"]["ref_video_audio_0"] == [components_id, 1] def test_locked_video_timing_is_forwarded_to_reference_bridge(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None output = module.EasyMiniMaxH3ToVideo.execute( **_base_inputs( mode=["reference"], videos=[object()], locked_video_timing_frames=[56], ) ) conditioning = _graph_node(output, module.REFERENCE_BRIDGE_NODE_ID) assert conditioning["inputs"]["locked_video_timing_frames"] == 56 def test_easy_node_reports_progress_for_each_media_input(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None audio = {"waveform": torch.ones(1, 1, 4), "sample_rate": 32000} module.EasyMiniMaxH3ToVideo.execute( **_base_inputs( mode=["reference"], audio_vae=[_AudioVae()], images=[_image_values(1, 2)], videos=[object(), object()], audios=[audio], ) ) progress = _ProgressBar.instances[-1] assert progress.total == 6 assert progress.updates == [ (1, 6), (2, 6), (3, 6), (4, 6), (5, 6), (6, 6), ] def test_easy_node_reports_complete_progress_for_empty_inputs(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None module.EasyMiniMaxH3ToVideo.execute(**_base_inputs(mode=["reference"])) progress = _ProgressBar.instances[-1] assert progress.total == 1 assert progress.updates == [(1, 1)] def test_multi_frames_reports_progress_for_each_expanded_image(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None module.EasyMiniMaxH3ToVideo.execute( **_base_inputs( mode=["multi_frames"], images=[_image_values(1, 2, 3)], ) ) progress = _ProgressBar.instances[-1] assert progress.total == 4 assert progress.updates == [(1, 4), (2, 4), (3, 4), (4, 4)] def test_reference_encodes_images_video_audio_and_standalone_audio(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None clip = _Clip() vae = _Vae() audio_vae = _AudioVae() video_audio = {"waveform": torch.ones(1, 1, 8), "sample_rate": 32000} class _Video: def get_components(self): return types.SimpleNamespace( images=_image_values(0, 1, 2, 3, 4), audio=video_audio, frame_rate=Fraction(12), ) standalone_audio = {"waveform": torch.zeros(1, 1, 4), "sample_rate": 16000} output = module.EasyMiniMaxH3ToVideo.execute( **_base_inputs( clip=[clip], vae=[vae], mode=["reference"], audio_vae=[audio_vae], images=[_image_values(9, 10)], videos=[[_Video()]], audios=[[standalone_audio]], ) ) conditioning = _graph_node(output, module.REFERENCE_BRIDGE_NODE_ID) components = _Video().get_components() fallback_output = module.MiniMaxH3ReferenceToVideoFallback.execute( clip=clip, vae=vae, audio_vae=audio_vae, prompt="prompt", width=32, height=32, length=5, ref_images={ "ref_image_0": _image_values(9), "ref_image_1": _image_values(10), }, ref_videos={"ref_video_0": components.images}, ref_video_audios={"ref_video_audio_0": components.audio}, ref_audios={"ref_audio_0": standalone_audio}, ) refs = fallback_output.values[0][0][1]["minimax_refs"] assert [ref["kind"] for ref in refs] == ["image", "image", "video_audio", "audio"] assert [ item["type"] for item in clip.tokenize_calls[0]["kwargs"]["minimax_ref_items"] ] == [ "image", "image", "audio", "video", "audio", ] assert vae.encoded[-1].shape[0] == 5 assert len(audio_vae.encoded) == 2 assert conditioning["inputs"]["ref_video_0"][1] == 0 def test_reference_bridge_groups_fixed_inputs_before_direct_execute(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None calls = [] class _NativeReferenceNode: @classmethod def execute(cls, **kwargs): calls.append(kwargs) return _NodeOutput("conditioning", "latent") module.comfy_nodes.NODE_CLASS_MAPPINGS[ "MiniMaxH3ReferenceToVideo" ] = _NativeReferenceNode audio = {"waveform": torch.ones(1, 1, 8), "sample_rate": 32000} output = module.EasyMiniMaxH3ReferenceToVideoBridge.execute( clip=_Clip(), vae=_Vae(), audio_vae=_AudioVae(), prompt="prompt", width=32, height=32, length=5, ref_image_0=_image_values(9), ref_video_0=_image_values(0, 1, 2, 3, 4), ref_video_audio_0=audio, ref_audio_0=audio, ) assert output.values == ("conditioning", "latent") assert list(calls[0]["ref_images"]) == ["ref_image_0"] assert list(calls[0]["ref_videos"]) == ["ref_video_0"] assert list(calls[0]["ref_video_audios"]) == ["ref_video_audio_0"] assert list(calls[0]["ref_audios"]) == ["ref_audio_0"] def test_reference_bridge_pads_video_tail_without_resizing(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None calls = [] def unexpected_resize(*_args, **_kwargs): raise AssertionError("reference bridge must leave spatial resizing to core") monkeypatch.setattr(module, "_resize", unexpected_resize) class _NativeReferenceNode: @classmethod def execute(cls, **kwargs): calls.append(kwargs) return _NodeOutput("conditioning", "latent") module.comfy_nodes.NODE_CLASS_MAPPINGS[ "MiniMaxH3ReferenceToVideo" ] = _NativeReferenceNode frames = _image_values(*range(120)) module.EasyMiniMaxH3ReferenceToVideoBridge.execute( clip=_Clip(), vae=_Vae(), prompt="prompt", width=32, height=32, length=120, ref_video_0=frames, ) aligned = calls[0]["ref_videos"]["ref_video_0"] assert aligned.shape == (124, 1, 1, 1) assert aligned[:120, 0, 0, 0].tolist() == list(map(float, range(120))) assert aligned[120:, 0, 0, 0].tolist() == [119.0] * 4 @pytest.mark.parametrize("source_frame_count", [39, 40, 41, 55, 56, 57]) def test_reference_bridge_uniformly_fits_locked_video_timing( monkeypatch, source_frame_count, ): module = _load_minimax_node(monkeypatch) assert module is not None calls = [] class _NativeReferenceNode: @classmethod def execute(cls, **kwargs): calls.append(kwargs) return _NodeOutput("conditioning", "latent") module.comfy_nodes.NODE_CLASS_MAPPINGS[ "MiniMaxH3ReferenceToVideo" ] = _NativeReferenceNode target_frame_count = module._align_frame_count(source_frame_count) frames = _image_values(*range(source_frame_count)) module.EasyMiniMaxH3ReferenceToVideoBridge.execute( clip=_Clip(), vae=_Vae(), prompt="prompt", width=32, height=32, length=target_frame_count, locked_video_timing_frames=target_frame_count, ref_video_0=frames, ) fitted = calls[0]["ref_videos"]["ref_video_0"] expected_indexes = torch.linspace( 0, source_frame_count - 1, target_frame_count, ).round().long() assert fitted.shape == (target_frame_count, 1, 1, 1) assert torch.equal(fitted[:, 0, 0, 0], expected_indexes.float()) restored_indexes = torch.linspace( 0, target_frame_count - 1, source_frame_count, ).round().long() assert fitted.index_select(0, restored_indexes)[:, 0, 0, 0].tolist() == list( map(float, range(source_frame_count)) ) def test_reference_fallback_pads_video_tail_instead_of_dropping_frames(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None vae = _Vae() frames = _image_values(*range(120)) module.MiniMaxH3ReferenceToVideoFallback.execute( clip=_Clip(), vae=vae, audio_vae=None, prompt="prompt", width=32, height=32, length=120, ref_videos={"ref_video_0": frames}, ) encoded = vae.encoded[-1] assert encoded.shape[0] == 124 assert encoded[119:, 0, 0, 0].tolist() == [119.0] * 5 def test_reference_bridge_directly_executes_fallback_when_native_is_missing(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None output = module.EasyMiniMaxH3ReferenceToVideoBridge.execute( clip=_Clip(), vae=_Vae(), prompt="prompt", width=32, height=32, length=5, ref_image_0=_image_values(9), ) refs = output.values[0][0][1]["minimax_refs"] assert [ref["kind"] for ref in refs] == ["image"] def test_reference_audio_requires_audio_vae(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None audio = {"waveform": torch.ones(1, 1, 4), "sample_rate": 32000} with pytest.raises(ValueError, match="audio_vae is required"): module.EasyMiniMaxH3ToVideo.execute( **_base_inputs(mode=["reference"], audios=[audio]) ) def test_silent_reference_video_does_not_require_audio_vae(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None output = module.EasyMiniMaxH3ToVideo.execute( **_base_inputs(mode=["reference"], videos=[object()]) ) conditioning = _graph_node(output, module.REFERENCE_BRIDGE_NODE_ID) assert conditioning["inputs"]["audio_vae"] is None @pytest.mark.parametrize( ("overrides", "media_name", "limit"), [ ({"images": [_image_values(*range(10))]}, "images", 9), ({"videos": [object()] * 4, "audio_vae": [_AudioVae()]}, "videos", 3), ( { "audios": [{"waveform": torch.ones(1, 1, 4), "sample_rate": 32000}] * 4, "audio_vae": [_AudioVae()], }, "audios", 3, ), ], ) def test_reference_media_over_native_limits_is_rejected( monkeypatch, overrides, media_name, limit ): module = _load_minimax_node(monkeypatch) assert module is not None with pytest.raises( ValueError, match=rf"reference mode supports at most {limit} {media_name}", ): module.EasyMiniMaxH3ToVideo.execute( **_base_inputs(mode=["reference"], **overrides) ) def test_minimax_node_has_complete_chinese_localization(): locale_path = Path(__file__).parents[1] / "locales" / "zh" / "nodeDefs.json" locale_path = Path(__file__).parents[1] / "locales" / "zh" / "nodeDefs.json" if not locale_path.exists(): pytest.skip("locales/zh/nodeDefs.json is not bundled in this repository") node_defs = json.loads(locale_path.read_text(encoding="utf-8")) translation = node_defs["easy minimaxH3ToVideo"] assert translation["display_name"] == "简易 MiniMax H3 视频生成" assert set(translation["inputs"]) == { "clip", "vae", "audio_vae", "images", "videos", "audios", "prompt", "mode", "width", "height", "length", "ref_image_size", } assert translation["inputs"]["mode"]["options"] == { "reference": "参考生视频", "multi_frames": "首尾帧生视频", "last_frame": "尾帧生视频", } assert translation["inputs"]["ref_image_size"]["options"] == { "match": "匹配生成尺寸", "max": "最大参考尺寸", } assert translation["outputs"] == { "0": {"name": "正向条件"}, "1": {"name": "潜空间"}, } assert "重采样" not in translation["description"] assert "重采样" not in translation["inputs"]["videos"]["tooltip"] assert "任何参考视频" in translation["inputs"]["audio_vae"]["tooltip"] assert "仅缩小，不放大" in translation["inputs"]["ref_image_size"]["tooltip"] assert "输入后会自动改走参考生视频" in translation["inputs"]["videos"]["tooltip"] assert "输入后会自动改走参考生视频" in translation["inputs"]["audios"]["tooltip"] for node_id in ["MiniMaxH3ImageToVideo", "MiniMaxH3ReferenceToVideo"]: fallback_translation = node_defs[node_id] assert fallback_translation["display_name"] assert fallback_translation["description"] assert fallback_translation["inputs"] assert fallback_translation["outputs"] == { "0": {"name": "正向条件"}, "1": {"name": "潜空间"}, } def test_remove_h3_motion_context_latent_schema(monkeypatch): module = _load_minimax_node(monkeypatch) assert module is not None schema = module.EasyRemoveH3MotionContextLatent.define_schema() assert schema.node_id == "easy removeH3MotionContextLatent" assert schema.display_name == "!!Remove h3 motion context latent" assert schema.is_output_node is True assert schema.not_idempotent is True assert schema.inputs[0].name == "filename_path" assert schema.inputs[0].kwargs["default"] == "h3_context/clip" assert schema.inputs[1].name == "input" assert [output.name for output in schema.outputs] == ["output", "deleted_count"] def test_remove_h3_motion_context_latent_has_matching_chinese_localization(): locale_path = Path(__file__).parents[1] / "locales" / "zh" / "nodeDefs.json" if not locale_path.exists(): pytest.skip("locales/zh/nodeDefs.json is not bundled in this repository") node_defs = json.loads( (Path(__file__).parents[1] / "locales" / "zh" / "nodeDefs.json").read_text() ) translation = node_defs["easy removeH3MotionContextLatent"] assert translation["inputs"]["input"]["name"] == "输入" assert translation["outputs"] == { "0": {"name": "输出"}, "1": {"name": "已删除数量"}, } def test_remove_h3_motion_context_latent_deletes_matching_output_files( monkeypatch, tmp_path ): module = _load_minimax_node(monkeypatch) assert module is not None latent_directory = tmp_path / "h3_context" latent_directory.mkdir() matching_files = [ latent_directory / "clip_00001.safetensors", latent_directory / "clip_00002_.safetensors", ] for path in matching_files: path.write_bytes(b"latent") preserved_file = latent_directory / "other_00001.safetensors" preserved_file.write_bytes(b"latent") folder_paths = types.ModuleType("folder_paths") folder_paths.get_output_directory = lambda: str(tmp_path) monkeypatch.setitem(sys.modules, "folder_paths", folder_paths) monkeypatch.setattr(module, "folder_paths", folder_paths) passthrough = object() output = module.EasyRemoveH3MotionContextLatent.execute( passthrough, "h3_context/clip", ) assert output.values == (passthrough, 2) assert not any(path.exists() for path in matching_files) assert preserved_file.exists() @pytest.mark.parametrize( "filename_path", ["", "../clip", "h3_context/../clip", "/h3_context/clip"], ) def test_remove_h3_motion_context_latent_rejects_unsafe_paths( monkeypatch, tmp_path, filename_path ): module = _load_minimax_node(monkeypatch) assert module is not None folder_paths = types.ModuleType("folder_paths") folder_paths.get_output_directory = lambda: str(tmp_path) monkeypatch.setitem(sys.modules, "folder_paths", folder_paths) with pytest.raises(ValueError, match="filename_path"): module.EasyRemoveH3MotionContextLatent.execute(object(), filename_path) @pytest.mark.parametrize("mode,first_only", [("single", False), ("dual", False), ("dual", True)]) @pytest.mark.parametrize("with_context", [False, True]) def test_audio_only_project_never_decodes_or_saves_video(monkeypatch, mode, first_only, with_context): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs( sampling_mode=_h3_sampling_mode(mode, **{"1st_pass_only": first_only}), upscale_model=["unused-upscaler"], ) info = inputs["tracks_info"][0] info.update(width=32, height=32) if with_context: info["tracks"][0]["segments"].append({ "start_frame": 120, "end_frame": 240, "content": {"continuity_mode": "context", "user_prompt": "continue"}, }) result = module.EasyMultiTrackProject.execute(**inputs) types_in_graph = {node["class_type"] for node in result.expand.values()} assert "VAEDecodeAudio" in types_in_graph assert not types_in_graph.intersection({ "VAEDecode", "VAEEncode", "ImageResizeKJv2", "easy minimaxH3LatentUpscaler", "easy saveVideo", "easy h3SegmentEncodingStart", "easy h3SegmentSaveEnd", "easy h3LockedAudioDurationAlign", }) artifacts = [n for n in result.expand.values() if n["class_type"] == "easy h3ProjectArtifact"] for artifact in artifacts: assert "audio" in artifact["inputs"] assert "video_path" not in artifact["inputs"] if with_context and not first_only: assert "easy h3AudioContextLatent" in types_in_graph trims = [n for n in result.expand.values() if n["class_type"] == "easy h3ContextMediaTrim"] assert trims assert all("images" not in n["inputs"] for n in trims) @pytest.mark.parametrize("dimensions", [(32, 64), (64, 32), (64, 64)]) def test_audio_only_project_requires_both_dimensions_to_be_32(monkeypatch, dimensions): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs(sampling_mode=_h3_sampling_mode("single")) inputs["tracks_info"][0].update(width=dimensions[0], height=dimensions[1]) result = module.EasyMultiTrackProject.execute(**inputs) assert _graph_node(result, "VAEDecode") assert _graph_node(result, "easy saveVideo") def test_audio_only_project_saves_original_locked_audio(monkeypatch): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs() inputs["tracks_info"][0].update(width=32, height=32) inputs["tracks_info"][0]["tracks"].append({ "type": "audio", "audio_locked": True, "segments": [{"start_frame": 0, "end_frame": 120, "content": {"media_type": "audio"}}], }) result = module.EasyMultiTrackProject.execute(**inputs) artifact = _graph_node(result, "easy h3ProjectArtifact") selector = result.expand[artifact["inputs"]["audio"][0]] assert selector["class_type"] == "easy h3LockedAudioSelect" assert selector["inputs"]["locked_audio"] == {"prepared_locked_audio": True} def test_audio_only_context_trims_samples_and_retains_encoded_audio(monkeypatch): module = _load_minimax_node(monkeypatch) audio = {"waveform": torch.arange(80).reshape(1, 1, 80).float(), "sample_rate": 48} trimmed = module.EasyH3ContextMediaTrim.execute( audio=audio, trim_frames=5, output_frames=22, fps=24, ) assert trimmed.values[0] is None assert trimmed.values[1]["waveform"].flatten().tolist() == list(range(10, 54)) encoded = torch.randn(1, 32, 2, 37) context = module.EasyH3AudioContextLatent.execute({"samples": encoded}, 22) video, retained_audio = context.values[0]["samples"].unbind() assert video.shape == (1, 24, 7, 2, 2) assert torch.count_nonzero(video) == 0 assert retained_audio is encoded def test_audio_only_artifact_writes_wav_manifest_and_cleans_up(monkeypatch, tmp_path): import soundfile as sf module = _load_minimax_node(monkeypatch) monkeypatch.setattr(module.folder_paths, "get_output_directory", lambda: str(tmp_path)) info = _h3_project_inputs()["tracks_info"][0] info.update(width=32, height=32) audio = {"waveform": torch.linspace(-0.5, 0.5, 4800).reshape(1, 1, -1), "sample_rate": 48000} project_dir = tmp_path / "easy_media" / "projects" / "audio-demo" for mode in ("new", "new", "override"): module.EasyH3ProjectArtifact.execute( project_name="audio-demo", project_save=mode, segment_index=0, context_latent=_h3_context_latent(), tracks_info=info, audio=audio, ) assert len(list(project_dir.glob("audio_0_*.wav"))) == 2 assert not list(project_dir.glob("*.mp4")) manifest = json.loads((project_dir / "project.json").read_text()) active = manifest["segments"]["0"]["generations"]["1"] assert "video" not in active waveform, sr = sf.read(project_dir / active["audio"]) assert sr == 48000 assert len(waveform) == 4800 assert torch.allclose(torch.from_numpy(waveform), audio["waveform"].flatten().double(), atol=1e-6) loaded_video, loaded_audio = module.EasyH3ProjectContextLatentLoad.execute( "audio-demo", 0 ).values[0]["samples"].unbind() assert loaded_video.shape[2] == 7 assert loaded_audio.shape[-1] == 37 module._project_module.clear_h3_project_segments_from("audio-demo", 0, tmp_path) assert not list(project_dir.glob("*.wav")) assert not list(project_dir.glob("*.safetensors")) @pytest.mark.parametrize("hidden_id", ["15", ["15"]]) def test_audio_only_project_rejects_video_combine_before_any_project_changes(monkeypatch, hidden_id): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs(project_save=["override"], segment_count=[-1]) inputs["tracks_info"][0].update(width=32, height=32) module.EasyMultiTrackProject.hidden = types.SimpleNamespace( unique_id=hidden_id, prompt={"17": { "class_type": "easy multitrackProjectVideoCombine", "inputs": {"project_name": ["15", 0]}, }}, ) def unexpected_work(*args, **kwargs): pytest.fail("Unsupported audio/video combination must fail before project or sampling work") for name in ("initialize_h3_project", "clear_h3_project_segments_from", "GraphBuilder", "detect_turbo_model"): monkeypatch.setattr(module._project_module, name, unexpected_work) with pytest.raises(ValueError, match="Project audio merging is not implemented yet"): module.EasyMultiTrackProject.execute(**inputs) @pytest.mark.parametrize("width,height,source", [ (32, 32, ["99", 0]), (32, 32, "another-project"), (32, 64, ["15", 0]), (64, 32, ["15", 0]), ]) def test_audio_project_preflight_does_not_block_other_projects_or_video(monkeypatch, width, height, source): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs(sampling_mode=_h3_sampling_mode("single")) inputs["tracks_info"][0].update(width=width, height=height) module.EasyMultiTrackProject.hidden = types.SimpleNamespace( unique_id="15", prompt={"17": { "class_type": "easy multitrackProjectVideoCombine", "inputs": {"project_name": source}, }}, ) result = module.EasyMultiTrackProject.execute(**inputs) assert _graph_node(result, "easy h3ProjectArtifact") def test_project_timing_keeps_native_nodes_and_inputs(monkeypatch): module = _load_minimax_node(monkeypatch) module.comfy_nodes.NODE_CLASS_MAPPINGS["ImageResizeKJv2"] = _ImageResizeKJWithNvidia inputs = _h3_project_inputs( project_name="timing-demo", sampling_mode=_h3_sampling_mode("dual"), ) inputs["tracks_info"][0]["tracks"][0]["segments"].append({ "start_frame": 120, "end_frame": 240, "content": { "task_mode": "l2v", "continuity_mode": "context", "images": [], "user_prompt": "continue", }, }) module.GraphBuilder.set_default_prefix("timing", 0, 0) result = module.EasyMultiTrackProject.execute(**inputs) tagged = [node for node in result.expand.values() if "easy_media_timing" in node.get("_meta", {})] assert len(tagged) >= 5 assert {node["class_type"] for node in tagged} == { "SamplerCustomAdvanced", "VAEEncode", "VAEEncodeAudio", "VAEDecode", "VAEDecodeAudio", } labels = [node["_meta"]["easy_media_timing"] for node in tagged] assert len(set(labels)) == len(labels) assert all(label.startswith("timing-demo / ") for label in labels) assert any("first_pass_sample_0" in label for label in labels) assert any("second_pass_sample_0" in label for label in labels) assert any("hires_context_1_video_encode" in label for label in labels) assert any("hires_context_1_audio_encode" in label for label in labels) assert any("low_context_1_video_encode" in label for label in labels) assert any("low_context_1_audio_encode" in label for label in labels) monkeypatch.setattr( module._project_module, "_timed_h3_project_graph", lambda graph, _name, _segments=None: graph.finalize(), ) module.GraphBuilder.set_default_prefix("timing", 0, 0) without_timing = module.EasyMultiTrackProject.execute(**inputs) graph_without_metadata = { node_id: {key: value for key, value in node.items() if key != "_meta"} for node_id, node in result.expand.items() } assert graph_without_metadata == without_timing.expand @pytest.mark.parametrize("duration", [120, 125, 124]) @pytest.mark.parametrize("continuity", ["shot", "context", "context_swap"]) @pytest.mark.parametrize("sampling", ["single", "dual"]) @pytest.mark.parametrize("track_type", ["audio", "video"]) def test_locked_media_preserves_source_span_in_both_passes( monkeypatch, duration, continuity, sampling, track_type, ): module = _load_minimax_node(monkeypatch) module.comfy_nodes.NODE_CLASS_MAPPINGS["ImageResizeKJv2"] = _ImageResizeKJWithNvidia inputs = _h3_project_inputs(sampling_mode=_h3_sampling_mode(sampling)) info = inputs["tracks_info"][0] info["tracks"][0]["segments"] = [ { "start_frame": index * duration, "end_frame": (index + 1) * duration, "content": {"task_mode": "ref", "continuity_mode": continuity}, } for index in range(2) ] info["tracks"].append({ "type": track_type, "audio_locked": True, "segments": [{ "start_frame": 0, "end_frame": duration * 2, "content": {"media_type": track_type}, }], }) result = module.EasyMultiTrackProject.execute(**inputs) graph = result.expand expected_generated = {120: 124, 125: 141, 124: 124}[duration] conditioning = [n for node_id, n in graph.items() if n["class_type"] == "easy minimaxH3ToVideo" and "second_pass_conditioning" not in node_id] for index, node in enumerate(conditioning): length = node["inputs"]["length"] if index == 1 and continuity != "shot": expression = graph[length[0]]["inputs"] assert expression["values.a"] == expected_generated assert expression["expression"] == "a + 34" else: assert length == expected_generated if track_type == "video": assert node["inputs"]["locked_video_timing_frames"] == expected_generated else: assert "locked_video_timing_frames" not in node["inputs"] trims = [ n for n in graph.values() if n["class_type"] == "easy h3ContextMediaTrim" and not n["inputs"].get("phase_align_video_encode") ] assert len(trims) == (4 if sampling == "dual" else 2) assert all(n["inputs"]["output_frames"] == duration for n in trims) assert all( n["inputs"].get("fit_video_duration", False) is (track_type == "video") for n in trims ) saves = [n for n in graph.values() if n["class_type"] == "easy saveVideo"] assert len(saves) == 2 for node in saves: trim = graph[node["inputs"]["input_mode.images"][0]] assert trim["inputs"]["output_frames"] == duration selector = graph[node["inputs"]["input_mode.audio"][0]] assert selector["class_type"] == "easy h3LockedAudioSelect" assert selector["inputs"]["locked_audio"] == {"prepared_locked_audio": True} # Every delivered segment, including the initial shot, supplies fresh context. encodes = [n for n in graph.values() if n["class_type"] == "VAEEncode" and "hires_context" in n["inputs"]["pixels"][0]] assert len(encodes) == 2 def test_audio_locked_context_mv_delivers_exact_timeline_spans(monkeypatch): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs(sampling_mode=_h3_sampling_mode("single")) info = inputs["tracks_info"][0] durations = [187, 185, 197, 223, 203, 191, 228, 204, 195, 222, 209] cursor = 0 task_segments = [] for duration in durations: task_segments.append({ "start_frame": cursor, "end_frame": cursor + duration, "content": { "task_mode": "default", "continuity_mode": "context", }, }) cursor += duration info["tracks"][0]["segments"] = task_segments info["tracks"].append({ "type": "audio", "audio_locked": True, "segments": [{ "start_frame": 0, "end_frame": cursor, "content": {"media_type": "audio"}, }], }) result = module.EasyMultiTrackProject.execute(**inputs) delivery_trims = sorted( ( node for node in result.expand.values() if node["class_type"] == "easy h3ContextMediaTrim" and not node["inputs"].get("phase_align_video_encode") ), key=lambda node: node["_meta"]["easy_media_segment"], ) assert [node["inputs"]["output_frames"] for node in delivery_trims] == durations assert all("fit_video_duration" not in node["inputs"] for node in delivery_trims) assert sum(node["inputs"]["output_frames"] for node in delivery_trims) == 2244 def test_audio_lock_priority_keeps_locked_video_timing(monkeypatch): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs(sampling_mode=_h3_sampling_mode("single")) info = inputs["tracks_info"][0] info["tracks"][0]["segments"][0]["end_frame"] = 125 info["tracks"].extend([ { "type": "video", "audio_locked": True, "segments": [{ "start_frame": 0, "end_frame": 125, "content": {"media_type": "video"}, }], }, { "type": "audio", "audio_locked": True, "segments": [{ "start_frame": 0, "end_frame": 125, "content": {"media_type": "audio"}, }], }, ]) result = module.EasyMultiTrackProject.execute(**inputs) conditioning = _graph_node(result, "easy minimaxH3ToVideo") assert conditioning["inputs"]["length"] == 141 audio_lock = _graph_node(result, "easy minimaxH3AudioLock") assert audio_lock["inputs"]["audio"] == {"prepared_locked_audio": True} @pytest.mark.parametrize("track_type,locked,audio_only,preserves_timing,fits_video", [ ("video", False, False, False, False), ("audio", False, False, False, False), ("audio", True, False, True, False), ("video", True, False, True, True), ("video", True, True, False, False), ("audio", True, True, False, False), ]) @pytest.mark.parametrize("duration", [120, 125, 131]) def test_source_timing_policy_leaves_other_tasks_unchanged( monkeypatch, track_type, locked, audio_only, preserves_timing, fits_video, duration, ): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs(sampling_mode=_h3_sampling_mode("single")) info = inputs["tracks_info"][0] info["tracks"][0]["segments"][0]["end_frame"] = duration if audio_only: info.update(width=32, height=32) info["tracks"].append({ "type": track_type, "audio_locked": locked, "segments": [{"start_frame": 0, "end_frame": duration, "content": {"media_type": track_type}}], }) result = module.EasyMultiTrackProject.execute(**inputs) length = _graph_node(result, "easy minimaxH3ToVideo")["inputs"]["length"] if preserves_timing: assert length == module._align_frame_count(duration) trims = [ node for node in result.expand.values() if node["class_type"] == "easy h3ContextMediaTrim" and not node["inputs"].get("phase_align_video_encode") ] assert len(trims) == 1 assert trims[0]["inputs"]["output_frames"] == duration assert trims[0]["inputs"].get("fit_video_duration", False) is fits_video else: assert isinstance(length, list) assert length[1] == 3 assert result.expand[length[0]]["class_type"] == "easy multiTrackTaskOutput" assert not any( node["class_type"] == "easy h3ContextMediaTrim" for node in result.expand.values() ) @pytest.mark.parametrize("duration", [120, 125, 124]) @pytest.mark.parametrize("prefix", [0, 22]) def test_video_locked_trim_fits_full_generated_span_without_dropping_tail( monkeypatch, duration, prefix, ): module = _load_minimax_node(monkeypatch) generated = module._align_frame_count(duration) + (34 if prefix else 0) images = torch.arange(generated, dtype=torch.float32).reshape(-1, 1, 1, 1) audio = {"waveform": torch.arange(generated * 2).reshape(1, 1, -1), "sample_rate": 48} result = module.EasyH3ContextMediaTrim.execute( images, audio, trim_frames=prefix, output_frames=duration, pad_audio=False, fit_video_duration=True, fps=24, ) available = images[prefix:] expected_indexes = torch.linspace(0, len(available) - 1, duration).round().long() assert torch.equal(result.values[0], available.index_select(0, expected_indexes)) assert result.values[0][0].item() == images[prefix].item() assert result.values[0][-1].item() == images[-1].item() assert torch.equal(result.values[1]["waveform"], audio["waveform"][..., prefix * 2:(prefix + duration) * 2]) def test_audio_locked_trim_discards_generated_tail_without_time_compression( monkeypatch, ): module = _load_minimax_node(monkeypatch) duration = 187 prefix = 22 generated = module._align_frame_count(duration) + 34 images = torch.arange(generated, dtype=torch.float32).reshape(-1, 1, 1, 1) audio = { "waveform": torch.arange(generated * 2).reshape(1, 1, -1), "sample_rate": 48, } result = module.EasyH3ContextMediaTrim.execute( images, audio, trim_frames=prefix, output_frames=duration, pad_audio=False, fps=24, ) assert torch.equal(result.values[0], images[prefix:prefix + duration]) assert result.values[0][-1].item() == images[prefix + duration - 1].item() assert result.values[0][-1].item() != images[-1].item() assert torch.equal( result.values[1]["waveform"], audio["waveform"][..., prefix * 2:(prefix + duration) * 2], ) @pytest.mark.parametrize("track_type", ["video"]) def test_audio_lock_first_pass_checkpoint_keeps_full_sampling_latent(monkeypatch, track_type): module = _load_minimax_node(monkeypatch) inputs = _h3_project_inputs( sampling_mode=_h3_sampling_mode("dual", **{"1st_pass_only": True}), ) info = inputs["tracks_info"][0] info["tracks"][0]["segments"][0]["end_frame"] = 125 info["tracks"].append({ "type": track_type, "audio_locked": True, "segments": [{"start_frame": 0, "end_frame": 125, "content": {"media_type": track_type}}], }) result = module.EasyMultiTrackProject.execute(**inputs) artifact = _graph_node(result, "easy h3ProjectArtifact")["inputs"] assert artifact["sampling_pass"] == "first" checkpoint = artifact["context_latent"] assert result.expand[checkpoint[0]]["class_type"] == "SamplerCustomAdvanced" assert checkpoint[1] == 1 # The clean low context is still used for continuing the next segment. low = artifact["context_latent_low"] assert result.expand[low[0]]["class_type"] == "LTXVConcatAVLatent" def test_minimax_prompt_override_node_serializes_ordered_prompts(monkeypatch): module = _load_minimax_node(monkeypatch) result = module.EasyMinimaxPromptOverride.execute( { "prompt_1": "second prompt", "prompt_0": "@图片1 first prompt", "prompt_2": None, }, duration="10,5,10", generation_type="r2v,i2v,l2v", continuity_mode="shot,context,context_swap", system_prompt="custom system prompt", video_track_lock=1, audio_track_lock=2, ) payload = json.loads(result.values[0]) assert payload["type"] == "minimax_prompt_override" assert payload["prompts"] == ["@图片1 first prompt", "second prompt"] assert payload["duration"] == "10,5,10" assert payload["generation_type"] == "r2v,i2v,l2v" assert payload["continuity_mode"] == "shot,context,context_swap" assert payload["system_prompt"] == "custom system prompt" assert payload["video_track_lock"] == 1 assert payload["audio_track_lock"] == 2 def test_minimax_prompt_override_node_has_complete_chinese_localization(): locale_path = Path(__file__).parents[1] / "locales" / "zh" / "nodeDefs.json" if not locale_path.exists(): pytest.skip("locales/zh/nodeDefs.json is not bundled in this repository") node_defs = json.loads( (Path(__file__).parents[1] / "locales" / "zh" / "nodeDefs.json").read_text() ) translation = node_defs["easy minimaxPromptOverride"] assert translation["display_name"] == "MiniMax 提示词覆盖" assert set(translation["inputs"]) == { "prompts", "duration", "generation_type", "continuity_mode", "system_prompt", "video_track_lock", "audio_track_lock", } assert translation["inputs"]["duration"]["name"] == "生成时长" assert translation["inputs"]["generation_type"]["name"] == "生成类型" assert translation["inputs"]["continuity_mode"]["name"] == "衔接模式" assert translation["inputs"]["system_prompt"]["name"] == "系统提示词" assert translation["inputs"]["video_track_lock"]["name"] == "视频轨锁定" assert translation["inputs"]["audio_track_lock"]["name"] == "音频轨锁定" assert translation["outputs"] == {"0": {"name": "提示词覆盖"}}