Fast_deploy / ComfyUI-MiniMaxH3-Context-Loop /tests /_motion_context_upstream_unit_test.py

Upload folder using huggingface_hub (part 2)

bc50ea0 verified 30 days ago

8.94 kB

#!/usr/bin/env python3
"""Standalone tests for transparent upstream Motion Context delegation."""

import importlib.util
import pathlib
import sys
import types


ROOT = pathlib.Path(__file__).resolve().parents[1]
PACKAGE = "h3_motion_context_upstream_unit"

package = types.ModuleType(PACKAGE)
package.__path__ = [str(ROOT)]
sys.modules[PACKAGE] = package

shared_nodes = types.ModuleType(PACKAGE + ".nodes")
shared_nodes.FRAME_RESCALE = 5.0 / 3.0
shared_nodes._pixel_frames = lambda _steps: 100
shared_nodes._video_from_latent = lambda _latent: types.SimpleNamespace(
    shape=(1, 16, 30, 2, 2))
shared_nodes._audio_tail_from_latent = lambda _latent, _frames: (
    "tail", 37, -1.0 / 3.0)
shared_nodes.payload_merge_calls = []


def ensure_payload_merge(**requirements):
    shared_nodes.payload_merge_calls.append(requirements)
    return "test native payload verified"


shared_nodes._ensure_native_payload_merge = ensure_payload_merge
sys.modules[shared_nodes.__name__] = shared_nodes

spec = importlib.util.spec_from_file_location(
    PACKAGE + ".motion_context_upstream",
    ROOT / "motion_context_upstream.py")
adapter = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = adapter
spec.loader.exec_module(adapter)


class Frames:
    def __init__(self, count):
        self.shape = (count, 8, 8, 3)


class UpstreamMotionContext:
    calls = []
    fail_audio_merge = False
    FUNCTION = "apply"

    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                name: ("ANY", {}) for name in (
                    "conditioning", "vae", "latent", "context_frames",
                    "context_length", "encode_mode", "anchor_mode", "crop",
                    "audio_context_length", "audio_mode", "target_start")
            },
            "optional": {
                "context_latent": ("LATENT", {}),
                "audio_vae": ("VAE", {}),
                "context_audio": ("AUDIO", {}),
            },
        }

    def apply(self, **kwargs):
        type(self).calls.append(kwargs)
        if type(self).fail_audio_merge:
            raise RuntimeError(
                "h3_motion_context: this graph requires native MiniMax H3 "
                "guide/MultiRef support from ComfyUI PR #15439, but the live "
                "runtime is missing native_keyframe_ref_audio_merge."
            )
        existing = list(kwargs["conditioning"][0][1].get(
            "minimax_keyframes", []))
        visual = {
            "resolved_frame_index": 0,
            "latent": "upstream-visual",
        }
        if kwargs.get("context_latent") is not None:
            visual["audio_latent"] = "upstream-audio"
        existing.append(visual)
        metadata = kwargs["conditioning"][0][1].copy()
        metadata["minimax_keyframes"] = existing
        return ([[kwargs["conditioning"][0][0], metadata]],
                int(kwargs["context_length"]))


class FallbackMotionContext:
    calls = []

    def apply(self, **kwargs):
        type(self).calls.append(kwargs)
        return ("fallback-conditioning", int(kwargs["context_length"]))


def apply(**overrides):
    values = {
        "conditioning": [["embedding", {}]],
        "vae": "video-vae",
        "latent": "target-latent",
        "context_frames": Frames(22),
        "context_length": 22,
        "encode_mode": "video",
        "anchor_mode": "head",
        "crop": "disabled",
        "audio_context_length": 0,
        "audio_mode": "timeline",
        "context_latent": None,
        "audio_vae": None,
        "context_audio": None,
        "video_context_latent": None,
    }
    values.update(overrides)
    return adapter.apply_motion_context(FallbackMotionContext, **values)


def main():
    original_nodes = sys.modules.get("nodes")
    comfy_nodes = types.ModuleType("nodes")
    comfy_nodes.NODE_CLASS_MAPPINGS = {
        adapter.UPSTREAM_NODE_ID: UpstreamMotionContext,
    }
    sys.modules["nodes"] = comfy_nodes
    original_native = adapter._native_guides_available
    adapter._native_guides_available = lambda: True
    try:
        # Normal Guide scenes delegate. The adapter arbitrates an incoming
        # first-frame collision and restores Loop-private output metadata.
        conditioning = [["embedding", {
            "minimax_keyframes": [
                {"resolved_frame_index": 0, "latent": "first"},
                {"resolved_frame_index": 70, "latent": "last"},
            ],
            "minimax_frame_count": 100,
        }]]
        output, trim = apply(conditioning=conditioning)
        assert trim == 22
        assert not FallbackMotionContext.calls
        assert len(UpstreamMotionContext.calls) == 1
        delegated = UpstreamMotionContext.calls[-1]
        assert delegated["target_start"] == 0
        assert [value["resolved_frame_index"] for value in
                delegated["conditioning"][0][1]["minimax_keyframes"]] == [70]
        metadata = output[0][1]
        assert [value["resolved_frame_index"] for value in
                metadata["minimax_keyframes"]] == [70, 0]
        assert metadata["minimax_keyframes"][-1][
            "h3_chain_context_visual"] is True
        assert conditioning[0][1]["minimax_keyframes"][0]["latent"] == "first"

        # The upstream latent-audio slice is retained, while the adapter
        # restores the signed H3 grid position without moving its visual Guide.
        output, _trim = apply(context_latent="previous-av-latent")
        keyframes = output[0][1]["minimax_keyframes"]
        assert len(keyframes) == 2
        assert keyframes[0]["resolved_frame_index"] == 0
        assert "audio_latent" not in keyframes[0]
        assert keyframes[0]["h3_chain_context_visual"] is True
        assert abs(keyframes[1]["resolved_frame_index"] + 0.4) < 1e-9
        assert keyframes[1]["audio_latent"] == "upstream-audio"

        # Provider versions predating their own compatibility guard must not
        # bypass Chain's live payload verification. Ref-mode audio adds a
        # keyframe payload that must merge with Ref2VA video; visual-only
        # Ref2VA needs video merge; timeline carry needs both payload merges.
        apply(context_latent="previous-av-latent", audio_mode="ref")
        assert shared_nodes.payload_merge_calls[-1] == {
            "require_video": True, "require_audio": False}
        ref_conditioning = [["embedding", {
            "minimax_refs": [{"kind": "image", "latent": "reference"}],
        }]]
        apply(conditioning=ref_conditioning)
        assert shared_nodes.payload_merge_calls[-1] == {
            "require_video": True, "require_audio": False}
        apply(
            conditioning=ref_conditioning,
            context_latent="previous-av-latent")
        assert shared_nodes.payload_merge_calls[-1] == {
            "require_video": True, "require_audio": True}

        # Loop-only modes remain lossless through the internal engine.
        result = apply(video_context_latent="previous-video-latent")
        assert result == ("fallback-conditioning", 22)
        assert FallbackMotionContext.calls[-1][
            "video_context_latent"] == "previous-video-latent"

        result = apply(
            context_frames=Frames(0), context_length=0,
            audio_context_length=33, context_latent="previous-av-latent")
        assert result == ("fallback-conditioning", 0)

        result = apply(
            audio_context_length=39,
            context_latent="previous-av-latent")
        assert result == ("fallback-conditioning", 22)

        original_repair = adapter._enable_internal_payload_merge
        adapter._enable_internal_payload_merge = lambda: (
            True, "test payload compatibility enabled")
        UpstreamMotionContext.fail_audio_merge = True
        result = apply(context_latent="previous-av-latent")
        assert result == ("fallback-conditioning", 22)
        assert FallbackMotionContext.calls[-1][
            "context_latent"] == "previous-av-latent"
        UpstreamMotionContext.fail_audio_merge = False
        adapter._enable_internal_payload_merge = original_repair

        # Older ComfyUI falls back even when the provider itself is installed.
        adapter._native_guides_available = lambda: False
        result = apply()
        assert result == ("fallback-conditioning", 22)

        # Missing provider is also a safe internal fallback.
        adapter._native_guides_available = lambda: True
        comfy_nodes.NODE_CLASS_MAPPINGS.clear()
        result = apply()
        assert result == ("fallback-conditioning", 22)
    finally:
        adapter._native_guides_available = original_native
        if original_nodes is None:
            sys.modules.pop("nodes", None)
        else:
            sys.modules["nodes"] = original_nodes

    print("upstream Motion Context: transparent delegation and guarded "
          "feature fallbacks passed")


if __name__ == "__main__":
    main()

	#!/usr/bin/env python3
	"""Standalone tests for transparent upstream Motion Context delegation."""

	import importlib.util
	import pathlib
	import sys
	import types


	ROOT = pathlib.Path(__file__).resolve().parents[1]
	PACKAGE = "h3_motion_context_upstream_unit"

	package = types.ModuleType(PACKAGE)
	package.__path__ = [str(ROOT)]
	sys.modules[PACKAGE] = package

	shared_nodes = types.ModuleType(PACKAGE + ".nodes")
	shared_nodes.FRAME_RESCALE = 5.0 / 3.0
	shared_nodes._pixel_frames = lambda _steps: 100
	shared_nodes._video_from_latent = lambda _latent: types.SimpleNamespace(
	shape=(1, 16, 30, 2, 2))
	shared_nodes._audio_tail_from_latent = lambda _latent, _frames: (
	"tail", 37, -1.0 / 3.0)
	shared_nodes.payload_merge_calls = []


	def ensure_payload_merge(**requirements):
	shared_nodes.payload_merge_calls.append(requirements)
	return "test native payload verified"


	shared_nodes._ensure_native_payload_merge = ensure_payload_merge
	sys.modules[shared_nodes.__name__] = shared_nodes

	spec = importlib.util.spec_from_file_location(
	PACKAGE + ".motion_context_upstream",
	ROOT / "motion_context_upstream.py")
	adapter = importlib.util.module_from_spec(spec)
	sys.modules[spec.name] = adapter
	spec.loader.exec_module(adapter)


	class Frames:
	def __init__(self, count):
	self.shape = (count, 8, 8, 3)


	class UpstreamMotionContext:
	calls = []
	fail_audio_merge = False
	FUNCTION = "apply"

	@classmethod
	def INPUT_TYPES(cls):
	return {
	"required": {
	name: ("ANY", {}) for name in (
	"conditioning", "vae", "latent", "context_frames",
	"context_length", "encode_mode", "anchor_mode", "crop",
	"audio_context_length", "audio_mode", "target_start")
	},
	"optional": {
	"context_latent": ("LATENT", {}),
	"audio_vae": ("VAE", {}),
	"context_audio": ("AUDIO", {}),
	},
	}

	def apply(self, **kwargs):
	type(self).calls.append(kwargs)
	if type(self).fail_audio_merge:
	raise RuntimeError(
	"h3_motion_context: this graph requires native MiniMax H3 "
	"guide/MultiRef support from ComfyUI PR #15439, but the live "
	"runtime is missing native_keyframe_ref_audio_merge."
	)
	existing = list(kwargs["conditioning"][0][1].get(
	"minimax_keyframes", []))
	visual = {
	"resolved_frame_index": 0,
	"latent": "upstream-visual",
	}
	if kwargs.get("context_latent") is not None:
	visual["audio_latent"] = "upstream-audio"
	existing.append(visual)
	metadata = kwargs["conditioning"][0][1].copy()
	metadata["minimax_keyframes"] = existing
	return ([[kwargs["conditioning"][0][0], metadata]],
	int(kwargs["context_length"]))


	class FallbackMotionContext:
	calls = []

	def apply(self, **kwargs):
	type(self).calls.append(kwargs)
	return ("fallback-conditioning", int(kwargs["context_length"]))


	def apply(**overrides):
	values = {
	"conditioning": [["embedding", {}]],
	"vae": "video-vae",
	"latent": "target-latent",
	"context_frames": Frames(22),
	"context_length": 22,
	"encode_mode": "video",
	"anchor_mode": "head",
	"crop": "disabled",
	"audio_context_length": 0,
	"audio_mode": "timeline",
	"context_latent": None,
	"audio_vae": None,
	"context_audio": None,
	"video_context_latent": None,
	}
	values.update(overrides)
	return adapter.apply_motion_context(FallbackMotionContext, **values)


	def main():
	original_nodes = sys.modules.get("nodes")
	comfy_nodes = types.ModuleType("nodes")
	comfy_nodes.NODE_CLASS_MAPPINGS = {
	adapter.UPSTREAM_NODE_ID: UpstreamMotionContext,
	}
	sys.modules["nodes"] = comfy_nodes
	original_native = adapter._native_guides_available
	adapter._native_guides_available = lambda: True
	try:
	# Normal Guide scenes delegate. The adapter arbitrates an incoming
	# first-frame collision and restores Loop-private output metadata.
	conditioning = [["embedding", {
	"minimax_keyframes": [
	{"resolved_frame_index": 0, "latent": "first"},
	{"resolved_frame_index": 70, "latent": "last"},
	],
	"minimax_frame_count": 100,
	}]]
	output, trim = apply(conditioning=conditioning)
	assert trim == 22
	assert not FallbackMotionContext.calls
	assert len(UpstreamMotionContext.calls) == 1
	delegated = UpstreamMotionContext.calls[-1]
	assert delegated["target_start"] == 0
	assert [value["resolved_frame_index"] for value in
	delegated["conditioning"][0][1]["minimax_keyframes"]] == [70]
	metadata = output[0][1]
	assert [value["resolved_frame_index"] for value in
	metadata["minimax_keyframes"]] == [70, 0]
	assert metadata["minimax_keyframes"][-1][
	"h3_chain_context_visual"] is True
	assert conditioning[0][1]["minimax_keyframes"][0]["latent"] == "first"

	# The upstream latent-audio slice is retained, while the adapter
	# restores the signed H3 grid position without moving its visual Guide.
	output, _trim = apply(context_latent="previous-av-latent")
	keyframes = output[0][1]["minimax_keyframes"]
	assert len(keyframes) == 2
	assert keyframes[0]["resolved_frame_index"] == 0
	assert "audio_latent" not in keyframes[0]
	assert keyframes[0]["h3_chain_context_visual"] is True
	assert abs(keyframes[1]["resolved_frame_index"] + 0.4) < 1e-9
	assert keyframes[1]["audio_latent"] == "upstream-audio"

	# Provider versions predating their own compatibility guard must not
	# bypass Chain's live payload verification. Ref-mode audio adds a
	# keyframe payload that must merge with Ref2VA video; visual-only
	# Ref2VA needs video merge; timeline carry needs both payload merges.
	apply(context_latent="previous-av-latent", audio_mode="ref")
	assert shared_nodes.payload_merge_calls[-1] == {
	"require_video": True, "require_audio": False}
	ref_conditioning = [["embedding", {
	"minimax_refs": [{"kind": "image", "latent": "reference"}],
	}]]
	apply(conditioning=ref_conditioning)
	assert shared_nodes.payload_merge_calls[-1] == {
	"require_video": True, "require_audio": False}
	apply(
	conditioning=ref_conditioning,
	context_latent="previous-av-latent")
	assert shared_nodes.payload_merge_calls[-1] == {
	"require_video": True, "require_audio": True}

	# Loop-only modes remain lossless through the internal engine.
	result = apply(video_context_latent="previous-video-latent")
	assert result == ("fallback-conditioning", 22)
	assert FallbackMotionContext.calls[-1][
	"video_context_latent"] == "previous-video-latent"

	result = apply(
	context_frames=Frames(0), context_length=0,
	audio_context_length=33, context_latent="previous-av-latent")
	assert result == ("fallback-conditioning", 0)

	result = apply(
	audio_context_length=39,
	context_latent="previous-av-latent")
	assert result == ("fallback-conditioning", 22)

	original_repair = adapter._enable_internal_payload_merge
	adapter._enable_internal_payload_merge = lambda: (
	True, "test payload compatibility enabled")
	UpstreamMotionContext.fail_audio_merge = True
	result = apply(context_latent="previous-av-latent")
	assert result == ("fallback-conditioning", 22)
	assert FallbackMotionContext.calls[-1][
	"context_latent"] == "previous-av-latent"
	UpstreamMotionContext.fail_audio_merge = False
	adapter._enable_internal_payload_merge = original_repair

	# Older ComfyUI falls back even when the provider itself is installed.
	adapter._native_guides_available = lambda: False
	result = apply()
	assert result == ("fallback-conditioning", 22)

	# Missing provider is also a safe internal fallback.
	adapter._native_guides_available = lambda: True
	comfy_nodes.NODE_CLASS_MAPPINGS.clear()
	result = apply()
	assert result == ("fallback-conditioning", 22)
	finally:
	adapter._native_guides_available = original_native
	if original_nodes is None:
	sys.modules.pop("nodes", None)
	else:
	sys.modules["nodes"] = original_nodes

	print("upstream Motion Context: transparent delegation and guarded "
	"feature fallbacks passed")


	if __name__ == "__main__":
	main()