nodes/goodguy1963-ComfyUI-ThinkingLLM
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
main
goodguy1963-ComfyUI-Thinkin…/tests/test_new_features.py
T

4680 lines
210 KiB
Python
Raw Permalink Blame History

"""Lightweight regression tests for the local ThinkingLLM fork.
Run from the ComfyUI-ThinkingLLM folder:
python tests/test_new_features.py
These tests intentionally avoid importing the model modules so they can run in a
plain shell without ML dependencies.
"""
import ast
import base64
import contextlib
import hashlib
import importlib.util
import inspect
import io
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import types
import unittest
import wave
from pathlib import Path
from unittest import mock
import torch
HERE = Path(__file__).resolve().parent
PKG = HERE.parent
WORKFLOW_DIR = PKG / "example_workflows"
STREAMING_FILES = {
"AILab_QwenVL_GGUF_PromptEnhancer.py": {
"stage": "INITIAL GENERATION",
"label": "QwenVL GGUF",
},
"AILab_QwenVL_PromptEnhancer.py": {
"stage": "HF TEXT GENERATION",
"label": "QwenVL HF",
},
"thinkingllm_core/hf_runtime.py": {
"stage": None,
"label": "QwenVL HF",
},
"AILab_QwenVL_GGUF.py": {
"stage": "STREAMING",
"label": "QwenVL GGUF",
},
}
def read_source(filename: str) -> str:
return (PKG / filename).read_text(encoding="utf-8")
def parse_source(filename: str) -> ast.AST:
return ast.parse(read_source(filename), filename=filename)
def build_stub_module(name: str, **attrs):
module = types.ModuleType(name)
for key, value in attrs.items():
setattr(module, key, value)
return module
class _StubQuantization:
FP16 = types.SimpleNamespace(value="fp16")
@classmethod
def get_values(cls):
return [cls.FP16.value]
def build_loader_test_stubs() -> dict[str, types.ModuleType]:
pil_image = build_stub_module("PIL.Image", Image=object)
pil_package = build_stub_module("PIL", Image=pil_image)
comfy_model_management = build_stub_module(
"comfy.model_management",
get_torch_device=lambda: types.SimpleNamespace(type="cuda", index=0),
throw_exception_if_processing_interrupted=lambda: None,
)
comfy_package = build_stub_module("comfy", model_management=comfy_model_management)
output_clean_config = type("OutputCleanConfig", (), {"__init__": lambda self, *args, **kwargs: None})
terminal_stream_display = type("TerminalStreamDisplay", (), {})
stream_degeneration_error = type("StreamDegenerationError", (RuntimeError,), {})
qwen_base = type("QwenVLBase", (), {"__init__": lambda self, *args, **kwargs: None})
return {
"numpy": build_stub_module("numpy", ndarray=object),
"torch": build_stub_module(
"torch",
Tensor=object,
dtype=object,
float16=object,
float32=object,
device=type("device", (), {}),
cuda=types.SimpleNamespace(device_count=lambda: 0, is_available=lambda: False),
backends=types.SimpleNamespace(mps=types.SimpleNamespace(is_available=lambda: False)),
no_grad=lambda func=None, *args, **kwargs: (lambda inner: inner) if func is None else func,
inference_mode=lambda func=None, *args, **kwargs: (lambda inner: inner) if func is None else func,
),
"huggingface_hub": build_stub_module(
"huggingface_hub",
HfApi=type(
"HfApi",
(),
{
"__init__": lambda self, *args, **kwargs: None,
"list_repo_files": lambda self, *args, **kwargs: [
"config.json",
"model.safetensors",
"processor_config.json",
],
},
),
hf_hub_download=lambda *args, **kwargs: "",
snapshot_download=lambda *args, **kwargs: "",
),
"PIL": pil_package,
"PIL.Image": pil_image,
"folder_paths": build_stub_module(
"folder_paths",
models_dir=str(PKG),
folder_names_and_paths={},
get_folder_paths=lambda *args, **kwargs: [],
get_filename_list=lambda *args, **kwargs: [],
get_full_path_or_raise=lambda category, filename: str(PKG / filename),
),
"AILab_StreamDisplay": build_stub_module(
"AILab_StreamDisplay",
StreamDegenerationError=stream_degeneration_error,
StreamDegenerationGuard=type("StreamDegenerationGuard", (), {"__init__": lambda self, *args, **kwargs: None, "push": lambda self, text: None}),
TerminalStreamDisplay=terminal_stream_display,
compose_streamed_model_output=lambda reasoning, content: (
f"<think>\n{reasoning.strip()}\n</think>\n\n{content.strip()}" if reasoning and content
else f"<think>\n{reasoning.strip()}\n</think>" if reasoning
else content.strip()
),
extract_stream_token=lambda chunk: {"reasoning": "", "content": (chunk.get("choices") or [{}])[0].get("delta", {}).get("content", "")},
strip_degenerate_repetition=lambda text, *args, **kwargs: text,
),
"AILab_LlamaCppInstaller": build_stub_module(
"AILab_LlamaCppInstaller",
ensure_llama_cpp_backend=lambda *args, **kwargs: object(),
format_llama_cpp_backend_info=lambda info=None: "stub backend",
get_last_llama_cpp_backend_info=lambda: {"gpu_offload": None, "vision_handlers": ["Qwen3VLChatHandler"]},
relax_windows_dll_directory_for_long_paths=contextlib.nullcontext,
),
"AILab_OutputCleaner": build_stub_module(
"AILab_OutputCleaner",
OUTPUT_CLEANER_VERSION=4,
OutputCleanConfig=output_clean_config,
clean_model_output=lambda text, *args, **kwargs: text,
prompt_output_guard=lambda text, *args, **kwargs: text,
),
"transformers": build_stub_module(
"transformers",
AutoModelForCausalLM=object,
AutoModelForVision2Seq=object,
AutoModelForImageTextToText=object,
AutoProcessor=object,
AutoTokenizer=object,
BitsAndBytesConfig=type("BitsAndBytesConfig", (), {}),
),
"comfy": comfy_package,
"comfy.model_management": comfy_model_management,
"AILab_QwenVL": build_stub_module(
"AILab_QwenVL",
ATTENTION_MODES=["auto"],
HF_ALL_MODELS={"stub-model": {}},
HF_TEXT_MODELS={},
HF_VL_MODELS={},
NODE_PROMPT_STATE={},
PROMPT_CACHE={},
MASK_FOCUS_INSTRUCTION="mask focus instruction",
_make_node_state_key=lambda *args, **kwargs: "node-key",
_build_workflow_fingerprint=lambda *args, **kwargs: "workflow-fingerprint",
_default_model_from_config=lambda models, fallback: next(
(
name
for name, info in models.items()
if isinstance(info, dict) and info.get("default")
),
next(iter(models.keys()), fallback),
),
apply_qwen_soft_thinking_directive=lambda *args, **kwargs: None,
apply_mask_highlight=lambda image, mask: (image, "mask-hash" if mask is not None else None),
build_node_input_signature=lambda *args, **kwargs: {},
download_hf_file_to_path=lambda *args, **kwargs: None,
ensure_model=lambda *args, **kwargs: None,
ensure_cuda_vram_headroom=lambda *args, **kwargs: None,
estimate_qwen_text_tokens=lambda *args, **kwargs: 0,
get_cache_key=lambda *args, **kwargs: "cache-key",
get_alternative_cache_key=lambda *args, **kwargs: "alt-cache-key",
get_image_hash=lambda *args, **kwargs: "image-hash",
get_video_hash=lambda *args, **kwargs: "video-hash",
log_llm_input=lambda *args, **kwargs: None,
model_catalog_label=lambda name, info: name,
model_catalog_options=lambda models, predicate=None: sorted(
name for name, info in models.items() if predicate is None or predicate(name, info or {})
),
normalize_commercial_status=lambda info: "local" if (info or {}).get("is_local") else (info or {}).get("commercial_status", "unclear"),
resolve_model_catalog_name=lambda models, selected: selected,
enforce_model_access=lambda *args, **kwargs: None,
save_prompt_cache=lambda *args, **kwargs: None,
get_node_saved_prompt=lambda *args, **kwargs: None,
get_node_saved_prompt_with_seed=lambda *args, **kwargs: None,
resolve_qwen_thinking_mode=lambda *args, **kwargs: False,
resolve_qwen_context_window=lambda *args, **kwargs: 0,
set_node_saved_prompt=lambda *args, **kwargs: None,
apply_video_duration_context=lambda prompt, *args, **kwargs: prompt,
ensure_video_prompt_duration=lambda prompt, *args, **kwargs: prompt,
resolve_video_duration=lambda *args, **kwargs: None,
validate_minimax_source_duration=lambda *args, **kwargs: [],
validate_minimax_reference_request=lambda *args, **kwargs: None,
load_node_prompt_state=lambda *args, **kwargs: None,
QwenVLBase=qwen_base,
Quantization=_StubQuantization,
TOOLTIPS={},
VIDEO_DURATION_INPUT=("FLOAT", {"default": 5.0, "min": 0.2, "max": 150.0, "step": 0.1}),
),
}
def load_thinkingllm_loader_subset(
module_filenames: list[str],
node_filenames: list[str] | None = None,
):
package_name = "thinkingllm_loader_smoke"
node_filenames = node_filenames or []
tracked_names = [
package_name,
*[Path(filename).stem for filename in module_filenames],
*[Path(filename).stem for filename in node_filenames],
]
previous_modules = {
name: sys.modules[name]
for name in tracked_names
if name in sys.modules
}
real_listdir = os.listdir
pkg_path = PKG.resolve()
nodes_path = (PKG / "nodes").resolve()
def fake_listdir(path):
resolved = Path(path).resolve()
if resolved == pkg_path:
return ["__init__.py", *module_filenames]
if resolved == nodes_path:
return node_filenames
return real_listdir(path)
try:
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
with mock.patch("os.listdir", side_effect=fake_listdir):
spec = importlib.util.spec_from_file_location(package_name, PKG / "__init__.py")
if spec is None or spec.loader is None:
raise AssertionError("failed to create loader spec for ThinkingLLM package")
module = importlib.util.module_from_spec(spec)
sys.modules[package_name] = module
spec.loader.exec_module(module)
return module
finally:
for name in tracked_names:
if name in previous_modules:
sys.modules[name] = previous_modules[name]
else:
sys.modules.pop(name, None)
def load_module_from_file(filename: str, module_name: str):
previous_module = sys.modules.get(module_name)
try:
spec = importlib.util.spec_from_file_location(module_name, PKG / filename)
if spec is None or spec.loader is None:
raise AssertionError(f"failed to create loader spec for {filename}")
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
return module
finally:
if previous_module is None:
sys.modules.pop(module_name, None)
else:
sys.modules[module_name] = previous_module
def extract_dict_node(filename: str, name: str) -> ast.Dict:
tree = parse_source(filename)
for node in tree.body:
if not isinstance(node, ast.Assign):
continue
for target in node.targets:
if isinstance(target, ast.Name) and target.id == name:
if isinstance(node.value, ast.Dict):
return node.value
raise AssertionError(f"{name} in {filename} is not a dict literal")
raise AssertionError(f"{name} not found in {filename}")
def extract_mapping_target_names(filename: str, name: str) -> dict[str, str]:
mapping = {}
dict_node = extract_dict_node(filename, name)
for key_node, value_node in zip(dict_node.keys, dict_node.values):
key = ast.literal_eval(key_node)
if isinstance(value_node, ast.Name):
mapping[key] = value_node.id
else:
raise AssertionError(f"{name}[{key}] in {filename} is not a direct name reference")
return mapping
def extract_literal_dict(filename: str, name: str) -> dict[str, str]:
mapping = {}
dict_node = extract_dict_node(filename, name)
for key_node, value_node in zip(dict_node.keys, dict_node.values):
key = ast.literal_eval(key_node)
mapping[key] = ast.literal_eval(value_node)
return mapping
def iter_workflow_types(path: Path) -> list[str]:
payload = json.loads(path.read_text(encoding="utf-8"))
return [
node.get("type")
for node in payload.get("nodes", [])
if isinstance(node, dict) and isinstance(node.get("type"), str)
]
def load_model_recommendations() -> dict:
return json.loads((PKG / "web" / "model_recommendations.json").read_text(encoding="utf-8"))
def matches_recommendation_rule(model_name: str, rules: list[dict]) -> bool:
lowered = model_name.lower()
ordered = sorted(rules, key=lambda rule: rule.get("priority", 0), reverse=True)
for rule in ordered:
include = [token.lower() for token in rule.get("match_any", [])]
exclude = [token.lower() for token in rule.get("exclude_any", [])]
if any(token in lowered for token in include) and not any(token in lowered for token in exclude):
return True
return False
def iter_catalog_display_names() -> list[str]:
names: list[str] = []
hf_payload = json.loads((PKG / "hf_models.json").read_text(encoding="utf-8"))
for section in ("hf_vl_models", "hf_text_models"):
names.extend(list((hf_payload.get(section) or {}).keys()))
gguf_payload = json.loads((PKG / "gguf_models.json").read_text(encoding="utf-8"))
for section in ("Qwen_model", "qwenVL_model"):
entries = gguf_payload.get(section) or {}
for repo_key, repo in entries.items():
if not isinstance(repo, dict):
continue
for model_file in repo.get("model_files") or []:
names.append(str(Path(model_file).name))
return names
class TestMaskFocusedImageAnalysis(unittest.TestCase):
@classmethod
def setUpClass(cls):
function_node = next(
node
for node in parse_source("thinkingllm_core/media.py").body
if isinstance(node, ast.FunctionDef) and node.name == "apply_mask_highlight"
)
namespace = {"hashlib": hashlib, "torch": torch}
exec(compile(ast.fix_missing_locations(ast.Module(body=[function_node], type_ignores=[])), "thinkingllm_core/media.py", "exec"), namespace)
cls.apply_mask_highlight = staticmethod(namespace["apply_mask_highlight"])
def test_mask_dims_rgb_and_preserves_alpha_without_mutating_input(self):
image = torch.ones((1, 2, 2, 4), dtype=torch.float32)
image[..., 3] = 0.5
original = image.clone()
mask = torch.tensor([[[1.0, 0.0], [0.5, 1.0]]])
highlighted, mask_hash = self.apply_mask_highlight(image, mask)
expected_brightness = torch.tensor([[1.0, 0.2], [0.6, 1.0]])
torch.testing.assert_close(highlighted[0, ..., :3], expected_brightness[..., None].expand(-1, -1, 3))
torch.testing.assert_close(highlighted[..., 3], original[..., 3])
torch.testing.assert_close(image, original)
self.assertEqual(len(mask_hash), 16)
def test_mask_resizes_and_hashes_complete_content(self):
image = torch.ones((1, 4, 4, 3), dtype=torch.float32)
first_mask = torch.zeros((1, 4, 4), dtype=torch.float32)
second_mask = first_mask.clone()
first_mask[0, 0, 0] = 1.0
second_mask[0, 0, 0] = 1.0
second_mask[0, -1, -1] = 1.0
small_mask = torch.tensor([[[1.0, 0.0], [0.0, 0.0]]])
resized, _ = self.apply_mask_highlight(image, small_mask)
_, first_hash = self.apply_mask_highlight(image, first_mask)
_, second_hash = self.apply_mask_highlight(image, second_mask)
self.assertEqual(resized.shape, image.shape)
self.assertGreater(float(resized[0, 0, 0, 0]), float(resized[0, -1, -1, 0]))
self.assertNotEqual(first_hash, second_hash)
def test_mask_requires_an_image_and_a_nonempty_selection(self):
with self.assertRaisesRegex(ValueError, "requires an image"):
self.apply_mask_highlight(None, torch.ones((1, 2, 2)))
with self.assertRaisesRegex(ValueError, "MASK is empty"):
self.apply_mask_highlight(torch.ones((1, 2, 2, 3)), torch.zeros((1, 2, 2)))
with self.assertRaisesRegex(ValueError, "mask mode"):
self.apply_mask_highlight(
torch.ones((1, 2, 2, 3)),
torch.ones((1, 2, 2)),
mode="unsupported",
)
def test_reconstruct_mode_conceals_selected_pixels_and_preserves_alpha(self):
image = torch.zeros((1, 16, 16, 4), dtype=torch.float32)
image[..., :3] = 0.25
image[:, 7:9, 7:9, :3] = 1.0
image[..., 3] = 0.75
mask = torch.zeros((1, 16, 16), dtype=torch.float32)
mask[:, 7:9, 7:9] = 1.0
concealed, reconstruct_hash = self.apply_mask_highlight(image, mask, mode="reconstruct")
_, focus_hash = self.apply_mask_highlight(image, mask, mode="focus")
torch.testing.assert_close(concealed[0, 8, 8, :3], torch.full((3,), 0.25))
torch.testing.assert_close(concealed[..., 3], image[..., 3])
self.assertNotEqual(reconstruct_hash, focus_hash)
def test_both_backends_apply_the_shared_focus_instruction(self):
hf_source = read_source("thinkingllm_core/hf_runtime.py")
gguf_source = read_source("AILab_QwenVL_GGUF.py")
self.assertIn("image, mask_hash = apply_mask_highlight(image, mask)", hf_source)
self.assertIn("image, mask_hash = apply_mask_highlight(image, mask, mode=mask_mode)", gguf_source)
self.assertIn('image_hash = f"{image_hash}:{mask_hash}"', hf_source)
self.assertIn('image_hash = f"{image_hash}:{mask_hash}"', gguf_source)
self.assertIn('prompt = f"{prompt}\\n\\n{MASK_FOCUS_INSTRUCTION}".strip()', hf_source)
self.assertIn("MASK_RECONSTRUCTION_INSTRUCTION", gguf_source)
self.assertIn(
"mask_preview, _ = apply_mask_highlight(image, mask, mode=mask_mode)",
gguf_source,
)
class TestTerminalDisplayHelper(unittest.TestCase):
def test_helper_file_exists(self):
self.assertTrue((PKG / "AILab_StreamDisplay.py").exists())
def test_helper_class_and_methods_exist(self):
tree = parse_source("AILab_StreamDisplay.py")
helper_class = None
for node in tree.body:
if isinstance(node, ast.ClassDef) and node.name == "TerminalStreamDisplay":
helper_class = node
break
self.assertIsNotNone(helper_class, "TerminalStreamDisplay class missing")
method_names = {
node.name for node in helper_class.body if isinstance(node, ast.FunctionDef)
}
for method in ["start_stage", "push", "flush", "end_stage", "push_compact", "end_compact"]:
self.assertIn(method, method_names)
class TestBufferedStreamingIntegration(unittest.TestCase):
def test_old_rolling_tail_patterns_removed(self):
old_patterns = [
r'print\(f"\\r\{tail\}"',
r"tail\s*=\s*full_streamed\[-200:\]",
r"tail\s*=\s*full_text\[-200:\]",
]
for filename in STREAMING_FILES:
source = read_source(filename)
for pattern in old_patterns:
with self.subTest(filename=filename, pattern=pattern):
self.assertIsNone(re.search(pattern, source))
def test_all_streaming_files_use_helper(self):
for filename, expectations in STREAMING_FILES.items():
source = read_source(filename)
with self.subTest(filename=filename):
self.assertIn("TerminalStreamDisplay", source)
self.assertIn(expectations["label"], source)
if expectations["stage"]:
self.assertIn(expectations["stage"], source)
self.assertIn(".push_compact(", source, f"{filename} should use compact streaming display")
self.assertIn(".end_compact(", source, f"{filename} should finish compact streaming cleanly")
def test_gguf_prompt_enhancer_has_no_off_toggle_text_stream_fallback(self):
source = read_source("AILab_QwenVL_GGUF_PromptEnhancer.py")
self.assertNotIn("_maybe_emit_prompt_compact_progress", source)
class TestModelRecommendations(unittest.TestCase):
def test_model_catalog_statuses_are_valid_and_assignments_are_conservative(self):
valid = {"cleared", "external_gated", "unclear", "noncommercial"}
hf_payload = json.loads((PKG / "hf_models.json").read_text(encoding="utf-8"))
gguf_payload = json.loads((PKG / "gguf_models.json").read_text(encoding="utf-8"))
hf_entries = [
entry
for section in ("hf_vl_models", "hf_text_models")
for entry in (hf_payload.get(section) or {}).values()
]
gguf_entries = [
entry
for section in ("Qwen_model", "qwenVL_model")
for entry in (gguf_payload.get(section) or {}).values()
]
self.assertTrue(all(entry.get("commercial_status") in valid for entry in hf_entries + gguf_entries))
self.assertEqual(2, sum(entry["commercial_status"] == "cleared" for entry in hf_entries))
self.assertEqual(2, sum(entry["commercial_status"] == "external_gated" for entry in hf_entries))
self.assertFalse(any(entry["commercial_status"] == "noncommercial" for entry in hf_entries + gguf_entries))
self.assertTrue(all(entry["commercial_status"] == "unclear" for entry in gguf_entries))
def test_commercial_catalog_contains_only_release_locked_models(self):
payload = json.loads((PKG / "hf_models.commercial.json").read_text(encoding="utf-8"))
entries = [
entry
for section in ("hf_vl_models", "hf_text_models")
for entry in payload.get(section, {}).values()
]
self.assertEqual({"MODEL-006", "MODEL-014"}, {entry["component_id"] for entry in entries})
self.assertTrue(all(re.fullmatch(r"[0-9a-f]{40}", entry["revision"]) for entry in entries))
self.assertTrue(all(entry["commercial_status"] == "cleared" for entry in entries))
def test_model_status_dropdowns_are_commercial_only_and_legacy_labels_resolve(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file("AILab_QwenVL.py", "thinkingllm_model_status_ui_test")
catalog = {
"unclear": {},
"noncommercial": {"commercial_status": "noncommercial"},
"local": {"is_local": True},
"gated": {"commercial_status": "external_gated"},
"cleared": {"commercial_status": "cleared"},
"invalid": {"commercial_status": "not-a-status"},
}
self.assertEqual(list(catalog), module.model_catalog_options(catalog))
access_globals = module.model_catalog_options.__globals__
with mock.patch.dict(access_globals, {"COMMERCIAL_RELEASE": True}):
options = module.model_catalog_options(catalog)
self.assertEqual(["✅", "🔑", "📁", "⚠", "⚠", "🚫"], [option.split()[0] for option in options])
for raw_name, entry in catalog.items():
decorated = f"{module.MODEL_STATUS_PRESENTATION[module.normalize_commercial_status(entry)][1]}{raw_name}"
self.assertEqual(raw_name, module.resolve_model_catalog_name(catalog, raw_name))
self.assertEqual(raw_name, module.resolve_model_catalog_name(catalog, decorated))
self.assertEqual("unclear", module.normalize_commercial_status({}))
self.assertEqual("unclear", module.normalize_commercial_status({"commercial_status": "invalid"}))
def test_model_catalog_labels_share_installed_and_local_conventions(self):
module = load_module_from_file(
"thinkingllm_core/model_access.py",
"thinkingllm_shared_model_label_test",
)
catalog = {
"catalog model": {"is_installed": True},
"remote model": {},
"[local] custom.gguf": {"is_local": True},
"[ComfyUI] encoder.safetensors": {
"is_local": True,
"local_display_name": "encoder.safetensors",
"storage_label": "text_encoders",
},
}
self.assertEqual(
[
"catalog model [installed]",
"remote model",
"[local] custom.gguf",
"[local] encoder.safetensors (text_encoders)",
],
module.model_catalog_options(catalog),
)
self.assertEqual(
"catalog model",
module.resolve_model_catalog_name(catalog, "catalog model [installed]"),
)
self.assertEqual(
"catalog model",
module.resolve_model_catalog_name(catalog, "catalog model"),
)
self.assertEqual(
"[ComfyUI] encoder.safetensors",
module.resolve_model_catalog_name(catalog, "[local] encoder.safetensors (text_encoders)"),
)
self.assertEqual(
"[ComfyUI] encoder.safetensors",
module.resolve_model_catalog_name(catalog, "[ComfyUI] encoder.safetensors"),
)
def test_qwenvl_facade_preserves_exports_and_shared_object_identity(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file("AILab_QwenVL.py", "thinkingllm_facade_compatibility_test")
self.assertTrue(all(hasattr(module, name) for name in module.__all__))
self.assertIs(module.PROMPT_CACHE, module.get_cache_key.__globals__["PROMPT_CACHE"])
self.assertIs(module.NODE_PROMPT_STATE, module.get_node_saved_prompt.__globals__["NODE_PROMPT_STATE"])
self.assertIs(module.VIDEO_PRESET_METADATA, module.get_video_preset_metadata.__globals__["VIDEO_PRESET_METADATA"])
model_globals = module.load_model_configs.__globals__
exported_catalogs = (module.HF_VL_MODELS, module.HF_TEXT_MODELS, module.HF_ALL_MODELS)
module.load_model_configs()
self.assertEqual(
exported_catalogs,
(model_globals["HF_VL_MODELS"], model_globals["HF_TEXT_MODELS"], model_globals["HF_ALL_MODELS"]),
)
self.assertTrue(all(before is after for before, after in zip(
exported_catalogs,
(model_globals["HF_VL_MODELS"], model_globals["HF_TEXT_MODELS"], model_globals["HF_ALL_MODELS"]),
)))
def test_community_hf_and_gguf_nodes_expose_plain_model_names(self):
package = load_thinkingllm_loader_subset([
"AILab_QwenVL.py",
"AILab_QwenVL_GGUF.py",
"AILab_QwenVL_PromptEnhancer.py",
"AILab_QwenVL_GGUF_PromptEnhancer.py",
])
prefixes = ["✅ Commercial | ", "🔑 External / gated | ", "📁 Local / user-supplied | ", "⚠ Rights unclear | ", "🚫 Non-commercial | "]
node_names = (
"ThinkingLLM_QwenVL",
"ThinkingLLM_QwenVL_Advanced",
"ThinkingLLM_QwenVL_PromptEnhancer",
"ThinkingLLM_QwenVL_GGUF",
"ThinkingLLM_QwenVL_GGUF_Advanced",
"ThinkingLLM_Gemma4_Audio_GGUF",
"ThinkingLLM_QwenVL_GGUF_PromptEnhancer",
)
for node_name in node_names:
with self.subTest(node_name=node_name):
options = package.NODE_CLASS_MAPPINGS[node_name].INPUT_TYPES()["required"]["model_name"][0]
self.assertTrue(all(not option.startswith(tuple(prefixes)) for option in options))
def test_commercial_mode_disables_mutable_runtime_behaviour(self):
init_source = read_source("__init__.py")
access_source = read_source("thinkingllm_core/model_access.py")
hf_models_source = read_source("thinkingllm_core/hf_models.py")
hf_runtime_source = read_source("thinkingllm_core/hf_runtime.py")
enhancer_source = read_source("AILab_QwenVL_PromptEnhancer.py")
self.assertIn('os.environ.get("THINKINGLLM_COMMERCIAL_RELEASE") == "1"', init_source)
self.assertIn('WEB_DIRECTORY = None if COMMERCIAL_RELEASE else "./web"', init_source)
self.assertNotIn('"AILab_QwenVL_GGUF"', re.search(r"COMMERCIAL_MODULES = \{.*?\}", init_source, re.DOTALL).group(0))
commercial_node_modules = re.search(
r"COMMERCIAL_NODE_MODULES\s*=\s*(\{.*?\})",
init_source,
re.DOTALL,
)
self.assertIsNotNone(commercial_node_modules)
self.assertEqual(
ast.literal_eval(commercial_node_modules.group(1)),
{"story_split_node", "system_prompt_preset", "vram_cleanup"},
)
self.assertIn("only permits preinstalled, hash-locked models", access_source)
self.assertIn("if custom.exists() and not COMMERCIAL_RELEASE", hf_models_source)
self.assertIn("if COMMERCIAL_RELEASE:\n _mark_installed_hf_catalog_models()", hf_models_source)
self.assertIn("else:\n _scan_local_hf_models()", hf_models_source)
self.assertIn('"local_files_only": COMMERCIAL_RELEASE', hf_runtime_source)
self.assertIn("trust_remote_code=not COMMERCIAL_RELEASE", hf_runtime_source)
self.assertIn('"local_files_only": COMMERCIAL_RELEASE', enhancer_source)
self.assertIn("commercial release mode forbids runtime package installation", enhancer_source)
def test_recommendation_rules_have_sources(self):
payload = load_model_recommendations()
rules = payload.get("rules") or []
self.assertTrue(rules, "model_recommendations.json should define at least one rule")
for rule in rules:
with self.subTest(rule=rule.get("id")):
self.assertTrue(rule.get("sources"), "each recommendation rule should record its research source")
def test_recommendation_notes_are_creative_work_focused(self):
payload = load_model_recommendations()
banned_terms = ("coding", "programming", "benchmark", "webdev", "math competition")
for rule in payload.get("rules") or []:
for note in rule.get("notes") or []:
lowered = note.lower()
with self.subTest(rule=rule.get("id"), note=note):
self.assertFalse(
any(term in lowered for term in banned_terms),
"recommendation notes should focus on creative prompt generation rather than coding/benchmarking",
)
def test_recommendations_cover_all_catalog_models(self):
payload = load_model_recommendations()
rules = payload.get("rules") or []
uncovered = [name for name in iter_catalog_display_names() if not matches_recommendation_rule(name, rules)]
self.assertEqual([], uncovered)
def test_qwen36_hauhau_exists_in_text_and_vision_catalogs(self):
payload = json.loads((PKG / "gguf_models.json").read_text(encoding="utf-8"))
text_catalog = payload.get("Qwen_model") or {}
vision_catalog = payload.get("qwenVL_model") or {}
text_key = next((key for key in text_catalog if "Qwen3.6-35B-A3B-Uncensored-HauhauCS-Aggressive" in key), None)
vision_key = next((key for key in vision_catalog if "Qwen3.6-35B-A3B-Uncensored-HauhauCS-Aggressive" in key), None)
self.assertIsNotNone(text_key)
self.assertIsNotNone(vision_key)
self.assertIn("mmproj-Qwen3.6-35B-A3B-Uncensored-HauhauCS-Aggressive-f16.gguf", vision_catalog[vision_key]["mmproj_file"])
self.assertGreaterEqual(len(text_catalog[text_key].get("model_files") or []), 11)
self.assertGreaterEqual(len(vision_catalog[vision_key].get("model_files") or []), 11)
def test_qwen38_blackfrost_exists_in_text_and_vision_catalogs(self):
payload = json.loads((PKG / "gguf_models.json").read_text(encoding="utf-8"))
text_catalog = payload.get("Qwen_model") or {}
vision_catalog = payload.get("qwenVL_model") or {}
expected_files = [
"Qwen3.8-27B-ABLITERATED-Q2_K.gguf",
"Qwen3.8-27B-ABLITERATED-Q3_K_S.gguf",
"Qwen3.8-27B-ABLITERATED-Q3_K_M.gguf",
"Qwen3.8-27B-ABLITERATED-Q4_K_S.gguf",
"Qwen3.8-27B-ABLITERATED-Q4_K_M.gguf",
"Qwen3.8-27B-ABLITERATED-Q5_K_S.gguf",
"Qwen3.8-27B-ABLITERATED-Q5_K_M.gguf",
"Qwen3.8-27B-ABLITERATED-Q6_K.gguf",
"Qwen3.8-27B-ABLITERATED-Q8_0.gguf",
]
text_key = next((key for key in text_catalog if "Qwen3.8-27B-ABLITERATED" in key), None)
vision_key = next((key for key in vision_catalog if "Qwen3.8-27B-ABLITERATED" in key), None)
self.assertIsNotNone(text_key)
self.assertIsNotNone(vision_key)
for catalog, key in ((text_catalog, text_key), (vision_catalog, vision_key)):
entry = catalog[key]
self.assertEqual(entry["repo_id"], "Blackfrost-AI/Qwen3.8-27B-ABLITERATED-GGUF")
self.assertEqual(entry["model_files"], expected_files)
self.assertEqual(entry["defaults"]["context_length"], 262144)
self.assertEqual(
vision_catalog[vision_key]["mmproj_file"],
"mmproj-Qwen3.8-27B-ABLITERATED-Q8_0.gguf",
)
self.assertEqual(
vision_catalog[vision_key]["mmproj_local_filenames"],
{
"mmproj-Qwen3.8-27B-ABLITERATED-Q8_0.gguf": [
"mmproj-Qwen3.8-27B-ABLITERATED-F16.gguf"
]
},
)
def test_qwen38_blackfrost_expands_in_both_gguf_node_catalogs(self):
stubs = build_loader_test_stubs()
with mock.patch.dict(sys.modules, stubs, clear=False):
vision_module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_qwen38_vision_catalog_test",
)
with mock.patch.dict(sys.modules, {"AILab_QwenVL_GGUF": vision_module}, clear=False):
prompt_module = load_module_from_file(
"AILab_QwenVL_GGUF_PromptEnhancer.py",
"thinkingllm_qwen38_prompt_catalog_test",
)
vision_models = vision_module.GGUF_VL_CATALOG.get("models") or {}
prompt_models = prompt_module.ThinkingLLM_QwenVL_GGUF_PromptEnhancer.load_gguf_models().get("models") or {}
q3_name = "Qwen3.8-27B-ABLITERATED-Q3_K_S.gguf [~12.1-13.3GB]"
q4_name = "Qwen3.8-27B-ABLITERATED-Q4_K_M.gguf [~15.6-16.5GB]"
self.assertIn(q3_name, vision_models)
self.assertIn(q4_name, vision_models)
self.assertIn(q3_name, prompt_models)
self.assertIn(q4_name, prompt_models)
resolved = vision_module._resolve_model_entry(q4_name)
self.assertEqual(resolved.mmproj_filename, "mmproj-Qwen3.8-27B-ABLITERATED-Q8_0.gguf")
self.assertIn("mmproj-Qwen3.8-27B-ABLITERATED-F16.gguf", resolved.mmproj_local_filenames)
def test_gemma4_12b_is_available_for_hf_and_gguf_vision_and_text(self):
hf_payload = json.loads((PKG / "hf_models.json").read_text(encoding="utf-8"))
gguf_payload = json.loads((PKG / "gguf_models.json").read_text(encoding="utf-8"))
self.assertTrue(any("Gemma-4-12B" in key for key in hf_payload.get("hf_vl_models", {})))
self.assertTrue(any("Gemma-4-12B" in key for key in hf_payload.get("hf_text_models", {})))
gguf_text = gguf_payload.get("Qwen_model") or {}
gguf_vision = gguf_payload.get("qwenVL_model") or {}
text_key = next((key for key in gguf_text if "Gemma-4-12B" in key), None)
vision_key = next((key for key in gguf_vision if "Gemma-4-12B" in key), None)
self.assertIsNotNone(text_key)
self.assertIsNotNone(vision_key)
self.assertIn("mmproj-F16.gguf", gguf_vision[vision_key]["mmproj_file"])
self.assertGreaterEqual(len(gguf_text[text_key].get("model_files") or []), 8)
self.assertGreaterEqual(len(gguf_vision[vision_key].get("model_files") or []), 8)
def test_gemma4_e2b_text_catalog_uses_current_hugging_face_files(self):
payload = json.loads((PKG / "gguf_models.json").read_text(encoding="utf-8"))
entries = [
entry
for key, entry in (payload.get("Qwen_model") or {}).items()
if "Gemma-4-E2B" in key and "Aggressive" not in key
]
self.assertTrue(entries)
for entry in entries:
self.assertEqual(entry["repo_id"], "bartowski/google_gemma-4-E2B-it-GGUF")
self.assertEqual(entry["model_files"], [
"google_gemma-4-E2B-it-Q4_K_S.gguf",
"google_gemma-4-E2B-it-Q8_0.gguf",
"google_gemma-4-E2B-it-bf16.gguf",
])
def test_hauhaucs_gemma4_gguf_only_models_are_not_in_transformers_catalogs(self):
hf_payload = json.loads((PKG / "hf_models.json").read_text(encoding="utf-8"))
gguf_payload = json.loads((PKG / "gguf_models.json").read_text(encoding="utf-8"))
expected = {
"HauhauCS/Gemma-4-E4B-Uncensored-HauhauCS-Aggressive": {
"model": "Gemma-4-E4B-Uncensored-HauhauCS-Aggressive-Q4_K_M.gguf",
"mmproj": "mmproj-Gemma-4-E4B-Uncensored-HauhauCS-Aggressive-f16.gguf",
},
"HauhauCS/Gemma-4-E2B-Uncensored-HauhauCS-Aggressive": {
"model": "Gemma-4-E2B-Uncensored-HauhauCS-Aggressive-Q4_K_P.gguf",
"mmproj": "mmproj-Gemma-4-E2B-Uncensored-HauhauCS-Aggressive-f16.gguf",
},
}
for hf_section in ("hf_vl_models", "hf_text_models"):
repo_ids = {entry.get("repo_id") for entry in (hf_payload.get(hf_section) or {}).values()}
self.assertTrue(expected.keys().isdisjoint(repo_ids))
for repo_id, filenames in expected.items():
text_entries = [entry for entry in (gguf_payload.get("Qwen_model") or {}).values() if entry.get("repo_id") == repo_id]
vision_entries = [entry for entry in (gguf_payload.get("qwenVL_model") or {}).values() if entry.get("repo_id") == repo_id]
self.assertEqual(1, len(text_entries))
self.assertEqual(1, len(vision_entries))
self.assertIn(filenames["model"], text_entries[0]["model_files"])
self.assertIn(filenames["model"], vision_entries[0]["model_files"])
self.assertEqual(filenames["mmproj"], vision_entries[0]["mmproj_file"])
def test_qwen3_asr_gguf_catalog_is_removed(self):
gguf_payload = json.loads((PKG / "gguf_models.json").read_text(encoding="utf-8"))
vision_catalog = gguf_payload.get("qwenVL_model") or {}
asr_entries = {key: value for key, value in vision_catalog.items() if "Qwen3-ASR" in key}
self.assertEqual({}, asr_entries)
def test_legacy_gguf_model_file_entries_expand_to_runnable_models(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_legacy_gguf_model_files_expand_test",
)
models = module.GGUF_VL_CATALOG.get("models") or {}
self.assertIn("gemma-4-12b-it-Q4_K_M.gguf [~7.5GB]", models)
resolved = module._resolve_model_entry("gemma-4-12b-it-Q4_K_M.gguf [~7.5GB]")
self.assertEqual(resolved.model_filename, "gemma-4-12b-it-Q4_K_M.gguf")
self.assertEqual(resolved.mmproj_filename, "mmproj-F16.gguf")
def test_minimax_h3_presets_are_split_by_input_mode(self):
data = json.loads((PKG / "AILab_System_Prompts.json").read_text(encoding="utf-8"))
text_label = "🎬 MiniMax H3 Text-to-Video"
first_label = "🖼️ MiniMax H3 First-Frame-to-Video (I2VA)"
first_last_label = "🎞️ MiniMax H3 First/Last-to-Video (FL2VA)"
last_label = "🏁 MiniMax H3 Last-Frame-to-Video (L2VA)"
reference_label = "🖼️ MiniMax H3 Reference-to-Video"
self.assertNotIn("🎬 MiniMax H3 Multimodal Video", data["_preset_prompts"])
for label in (text_label, first_label, first_last_label, last_label, reference_label):
self.assertIn(label, data["_preset_prompts"])
self.assertIn(label, data["qwenvl"])
self.assertIn("integrated_multimodal_description", data["qwenvl"][text_label])
self.assertNotIn("subject_definitions", data["qwenvl"][text_label])
self.assertIn("literal square brackets", data["qwenvl"][text_label])
self.assertIn("may be visual-only", data["qwenvl"][text_label])
self.assertIn("<d>[English]", data["qwen_text"]["styles"][text_label]["system_prompt"])
self.assertIn("subject_definitions", data["qwenvl"][reference_label])
self.assertIn("retention_analysis", data["qwenvl"][reference_label])
self.assertIn(text_label, data["qwen_text"]["styles"])
self.assertNotIn(reference_label, data["qwen_text"]["styles"])
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
package = load_thinkingllm_loader_subset(
[
"AILab_QwenVL.py",
"AILab_QwenVL_GGUF.py",
"AILab_QwenVL_PromptEnhancer.py",
"AILab_QwenVL_GGUF_PromptEnhancer.py",
]
)
for node_name in ("ThinkingLLM_QwenVL", "ThinkingLLM_QwenVL_Advanced", "ThinkingLLM_QwenVL_GGUF", "ThinkingLLM_QwenVL_GGUF_Advanced"):
values = package.NODE_CLASS_MAPPINGS[node_name].INPUT_TYPES()["required"]["preset_prompt"][0]
for label in (text_label, first_label, first_last_label, last_label, reference_label):
self.assertIn(label, values)
for node_name, widget_name in (("ThinkingLLM_QwenVL_PromptEnhancer", "enhancement_style"), ("ThinkingLLM_QwenVL_GGUF_PromptEnhancer", "preset_system_prompt")):
values = package.NODE_CLASS_MAPPINGS[node_name].INPUT_TYPES()["required"][widget_name][0]
self.assertIn(text_label, values)
for label in (first_label, first_last_label, last_label, reference_label):
self.assertNotIn(label, values)
def test_minimax_reference_requires_a_custom_target_prompt(self):
stub_modules = build_loader_test_stubs()
stub_modules.pop("AILab_QwenVL", None)
with mock.patch.dict(sys.modules, stub_modules, clear=False):
module = load_module_from_file("AILab_QwenVL.py", "thinkingllm_minimax_reference_validation_test")
with self.assertRaisesRegex(ValueError, "requires a custom prompt"):
module.validate_minimax_reference_request("🖼️ MiniMax H3 Reference-to-Video", " ")
module.validate_minimax_reference_request("🖼️ MiniMax H3 Reference-to-Video", "Keep Picture 1 as the character identity.")
module.validate_minimax_reference_request("🎬 MiniMax H3 Text-to-Video", "")
for filename in ("thinkingllm_core/hf_runtime.py", "AILab_QwenVL_GGUF.py"):
self.assertIn(
"validate_minimax_reference_request(preset_prompt, custom_prompt)",
read_source(filename),
)
def test_llm_input_logging_is_complete_and_shared(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file("AILab_QwenVL.py", "thinkingllm_input_logging_test")
long_tail = "END-OF-UNTRUNCATED-PROMPT"
output = io.StringIO()
with contextlib.redirect_stdout(output):
module.log_llm_input(
"Backend",
"FINALIZATION ATTEMPT 2/3",
"Preset Name",
f"user text {long_tail}",
system_text="system text",
formatted_text="<chat>formatted text</chat>",
media={"image": 2, "audio": 1},
)
logged = output.getvalue()
self.assertIn("Preset: Preset Name", logged)
self.assertIn("Stage: FINALIZATION ATTEMPT 2/3", logged)
self.assertIn("system text", logged)
self.assertIn(long_tail, logged)
self.assertIn("<chat>formatted text</chat>", logged)
self.assertIn("image=2, audio=1", logged)
self.assertNotIn("data:image", logged)
self.assertNotIn("base64", logged.lower())
for filename in ("thinkingllm_core/hf_runtime.py", "AILab_QwenVL_GGUF.py", "AILab_QwenVL_PromptEnhancer.py", "AILab_QwenVL_GGUF_PromptEnhancer.py"):
with self.subTest(filename=filename):
source = read_source(filename)
self.assertIn("log_llm_input(", source)
self.assertIn("no LLM call was made", source)
def test_ltx_presets_are_available_for_vision_and_prompt_enhancers(self):
data = json.loads(Path("AILab_System_Prompts.json").read_text(encoding="utf-8"))
self.assertIn("🎦 LTX 2.3 NSFW I2V Scene", data["_preset_prompts"])
self.assertIn("🎦 LTX 2.3 NSFW I2V Scene", data["qwenvl"])
self.assertIn("📖 LTX 2.3 NSFW T2V Scene", data["qwen_text"]["styles"])
vision_prompt = data["qwenvl"]["🎦 LTX 2.3 NSFW I2V Scene"]
text_prompt = data["qwen_text"]["styles"]["📖 LTX 2.3 NSFW T2V Scene"]["system_prompt"]
self.assertIn("LTX 2.3", vision_prompt)
self.assertIn("image first", vision_prompt)
self.assertIn("LTX 2.3", text_prompt)
self.assertIn("Do not change the user input intent", text_prompt)
self.assertNotIn("Example:", vision_prompt)
self.assertNotIn("Examples:", text_prompt)
def test_video_preset_duration_and_recommendation_metadata_are_complete(self):
data = json.loads((PKG / "AILab_System_Prompts.json").read_text(encoding="utf-8"))
metadata = data["_video_presets"]
duration_labels = {
"📖 LTX 2.3 NSFW T2V Scene",
"🎦 LTX 2.3 NSFW I2V Scene",
"🎞️ LTX 2.3 First/Last-to-Video",
"🖼️ LTX 2.3 Reference-to-Video",
"🎬 MiniMax H3 Text-to-Video",
"🖼️ MiniMax H3 First-Frame-to-Video (I2VA)",
"🎞️ MiniMax H3 First/Last-to-Video (FL2VA)",
"🏁 MiniMax H3 Last-Frame-to-Video (L2VA)",
"🖼️ MiniMax H3 Reference-to-Video",
}
wan_labels = {
"🍿 Wan 2.2 NSFW I2V Timeline (3s)",
"🍿 Wan 2.2 NSFW I2V Timeline (5s)",
"🎥 Wan 2.2 NSFW I2V Scene (3s)",
"🎥 Wan 2.2 NSFW I2V Scene (5s)",
"🎬 Wan 2.2 NSFW I2V Timeline (20s)",
"📖 Wan 2.2 NSFW I2V Scene (20s)",
"🍿 Wan 2.2 NSFW T2V Timeline (3s)",
"🍿 Wan 2.2 NSFW T2V Timeline (5s)",
"🎥 Wan 2.2 NSFW T2V Scene (3s)",
"🎥 Wan 2.2 NSFW T2V Scene (5s)",
"🎬 Wan 2.2 NSFW T2V Timeline (20s)",
"📖 Wan 2.2 NSFW T2V Scene (20s)",
}
self.assertEqual(set(metadata), duration_labels | wan_labels)
self.assertNotIn("🖼️ MiniMax H3 Reference-to-Video", data["qwen_text"]["styles"])
for label, entry in metadata.items():
with self.subTest(label=label):
self.assertIn(entry["provider"], {"ltx_2_3", "minimax_h3", "wan_2_2"})
self.assertTrue(entry["video_mode"])
self.assertIs(entry["duration_required"], label in duration_labels)
self.assertIs(entry["recommended_settings"]["enable_thinking"], False)
self.assertIn(entry["recommended_settings"]["max_tokens"], {512, 768, 1024, 1536, 2048, 3072})
self.assertTrue(entry["recommendation_note"])
styles = data["qwen_text"]["styles"]
self.assertIn("🍿 Wan 2.2 NSFW T2V Timeline (3s)", styles)
self.assertNotIn("🍿 Wan 2.2 NSFW T2V Timeline (3)", styles)
wan_t2v_20 = styles["📖 Wan 2.2 NSFW T2V Scene (20s)"]["system_prompt"]
self.assertIn("EXACTLY FOUR continuous 5-second", wan_t2v_20)
self.assertNotIn("5s total) to establish initial scene", wan_t2v_20)
self.assertNotIn("PERFORMANCE OPTIMIZATION", json.dumps(data, ensure_ascii=False))
self.assertNotIn("RECOMMENDED: Use max_tokens", json.dumps(data, ensure_ascii=False))
def test_video_duration_normalization_and_native_output_enforcement(self):
stub_modules = build_loader_test_stubs()
stub_modules.pop("AILab_QwenVL", None)
with mock.patch.dict(sys.modules, stub_modules, clear=False):
module = load_module_from_file("AILab_QwenVL.py", "thinkingllm_video_duration_test")
for requested, expected_frames, expected_seconds, expected_text in (
(5.0, 124, 124 / 24, "5.17"),
(8.0, 192, 8.0, "8.00"),
(12.0, 294, 12.25, "12.25"),
):
with self.subTest(requested=requested):
resolved = module.resolve_video_duration("🎬 MiniMax H3 Text-to-Video", requested)
self.assertEqual(resolved["frames"], expected_frames)
self.assertAlmostEqual(resolved["effective_seconds"], expected_seconds)
self.assertEqual(resolved["duration_text"], expected_text)
ltx = module.resolve_video_duration("📖 LTX 2.3 NSFW T2V Scene", 10.0)
self.assertEqual(ltx["effective_seconds"], 10.0)
self.assertEqual(ltx["duration_text"], "10")
self.assertIsNone(module.resolve_video_duration("🖼️ Detailed Description", 12.0))
ltx_output = module.ensure_video_prompt_duration(
"A dancer turns.\nThe camera follows.",
"📖 LTX 2.3 NSFW T2V Scene",
10.0,
)
self.assertEqual(ltx_output.count("10-second continuous shot"), 1)
self.assertNotIn("\n", ltx_output)
minimax_output = module.ensure_video_prompt_duration(
"integrated_multimodal_description: Target duration: 5.17 seconds. [Shot 1] A dancer turns.\n"
"overall_soundscape: Footsteps.\n"
"non_diegetic_music: N/A",
"🎬 MiniMax H3 Text-to-Video",
5.0,
)
self.assertNotIn("Target duration", minimax_output)
self.assertTrue(minimax_output.startswith("integrated_multimodal_description: [Shot 1]"))
self.assertLess(minimax_output.index("integrated_multimodal_description"), minimax_output.index("overall_soundscape"))
reference_output = module.ensure_video_prompt_duration(
"subject_definitions: <Subject 1> is the actor in <Picture 1>.\n"
"summary: [reference generation] Target duration: 12.25 seconds. The target video uses <Subject 1>.\n"
"retention_analysis: <Subject 1> (appears in [Shot 1]): fully_preserved - identity is retained.\n"
"detailed_description: Cinematic live action.\n[Shot 1] <Subject 1> performs the requested action.\n"
"overall_soundscape: Room tone.\n"
"non_diegetic_music: N/A",
"🖼️ MiniMax H3 Reference-to-Video",
12.0,
)
self.assertNotIn("Target duration", reference_output)
self.assertIn("summary: [reference generation] The target video uses <Subject 1>.", reference_output)
self.assertEqual(
[match.group(1) for match in re.finditer(r"(?m)^(subject_definitions|summary|retention_analysis|detailed_description|overall_soundscape|non_diegetic_music):", reference_output)],
["subject_definitions", "summary", "retention_analysis", "detailed_description", "overall_soundscape", "non_diegetic_music"],
)
ordinary = "A plain image description."
self.assertEqual(module.ensure_video_prompt_duration(ordinary, "🖼️ Detailed Description", 30.0), ordinary)
def test_video_duration_context_and_cache_use_effective_duration(self):
stub_modules = build_loader_test_stubs()
stub_modules.pop("AILab_QwenVL", None)
with mock.patch.dict(sys.modules, stub_modules, clear=False):
module = load_module_from_file("AILab_QwenVL.py", "thinkingllm_video_duration_cache_test")
context = module.apply_video_duration_context(
"Make it 3 seconds long.",
"🎬 MiniMax H3 Text-to-Video",
5.0,
)
self.assertIn("TARGET_DURATION_SECONDS: 5.17", context)
self.assertIn("overrides any conflicting duration", context)
resolved_five = module.resolve_video_duration("🎬 MiniMax H3 Text-to-Video", 5.0)
self.assertEqual(resolved_five["contract_version"], module.VIDEO_PROMPT_CONTRACT_VERSION)
five_seconds = resolved_five["effective_seconds"]
five_point_one_seconds = module.resolve_video_duration("🎬 MiniMax H3 Text-to-Video", 5.1)["effective_seconds"]
eight_seconds = module.resolve_video_duration("🎬 MiniMax H3 Text-to-Video", 8.0)["effective_seconds"]
self.assertEqual(five_seconds, five_point_one_seconds)
self.assertEqual(
module.get_cache_key("m", "p", "x", effective_duration_seconds=five_seconds),
module.get_cache_key("m", "p", "x", effective_duration_seconds=five_point_one_seconds),
)
self.assertNotEqual(
module.get_cache_key("m", "p", "x", effective_duration_seconds=five_seconds),
module.get_cache_key("m", "p", "x", effective_duration_seconds=eight_seconds),
)
self.assertNotEqual(
module.get_cache_key(
"m",
"p",
"x",
effective_duration_seconds=five_seconds,
video_prompt_contract_version=module.VIDEO_PROMPT_CONTRACT_VERSION - 1,
),
module.get_cache_key(
"m",
"p",
"x",
effective_duration_seconds=five_seconds,
video_prompt_contract_version=module.VIDEO_PROMPT_CONTRACT_VERSION,
),
)
self.assertEqual(
module.build_node_input_signature(effective_duration_seconds=five_seconds),
module.build_node_input_signature(effective_duration_seconds=five_point_one_seconds),
)
self.assertNotEqual(
module.build_node_input_signature(effective_duration_seconds=five_seconds),
module.build_node_input_signature(effective_duration_seconds=eight_seconds),
)
with self.assertRaisesRegex(ValueError, "between 0.2 and 150.0"):
module.resolve_video_duration("🎬 MiniMax H3 Text-to-Video", 0.1)
self.assertIsNone(module.resolve_video_duration("🖼️ Detailed Description", 0.1))
def test_minimax_h3_output_is_normalized_without_becoming_a_schema_gate(self):
stub_modules = build_loader_test_stubs()
stub_modules.pop("AILab_QwenVL", None)
with mock.patch.dict(sys.modules, stub_modules, clear=False):
module = load_module_from_file("AILab_QwenVL.py", "thinkingllm_minimax_contract_test")
base_body = (
"integrated_multimodal_description: Target duration: 8.00 seconds. "
"[Shot 1] Live-action, a baker opens the shutters. "
"[Shot 2] At 00:3.5 the camera cuts to the bread on the counter.\n"
"overall_soundscape: Shutters scrape and trays clink.\n"
"non_diegetic_music: An invented electronic drone."
)
t2va = module.ensure_video_prompt_duration(
base_body,
"🎬 MiniMax H3 Text-to-Video",
8.0,
source_prompt_text="A baker opens the shutters and the camera cuts to bread.",
)
self.assertTrue(t2va.startswith("integrated_multimodal_description: [Shot 1]"))
self.assertIn("[Shot 2] At 00:03.500,", t2va)
self.assertNotIn("Target duration", t2va)
self.assertTrue(t2va.endswith("non_diegetic_music: An invented electronic drone."))
for preset in (
"🖼️ MiniMax H3 First-Frame-to-Video (I2VA)",
"🎞️ MiniMax H3 First/Last-to-Video (FL2VA)",
"🏁 MiniMax H3 Last-Frame-to-Video (L2VA)",
):
with self.subTest(preset=preset):
output = module.ensure_video_prompt_duration(base_body, preset, 8.0)
self.assertTrue(output.startswith("integrated_multimodal_description:"))
self.assertEqual(output.count("integrated_multimodal_description:"), 1)
reference = module.ensure_video_prompt_duration(
"subject_definitions: <Subject 1> is the baker from <Picture 1>.\n"
"summary: [reference generation] The target video shows <Subject 1> opening the bakery.\n"
"retention_analysis: <Subject 1> (appears in [Shot 1]): fully_preserved - identity and clothing are retained.\n"
"detailed_description: Live-action with cool dawn light.\n"
"[Shot 1] <Subject 1> opens the shutters.\n"
"overall_soundscape: Shutters scrape over quiet street ambience.\n"
"non_diegetic_music: A fabricated synth cue.",
"🖼️ MiniMax H3 Reference-to-Video",
12.0,
source_prompt_text="Use Picture 1 for the baker opening the bakery, without music.",
)
self.assertEqual(
re.findall(
r"(?m)^(subject_definitions|summary|retention_analysis|detailed_description|overall_soundscape|non_diegetic_music):",
reference,
),
[
"subject_definitions",
"summary",
"retention_analysis",
"detailed_description",
"overall_soundscape",
"non_diegetic_music",
],
)
self.assertTrue(reference.endswith("non_diegetic_music: A fabricated synth cue."))
malformed_dialogue = (
"integrated_multimodal_description: [Shot 1] Peter (S1) says, "
"<d>English</d> This text is outside the dialogue tag.\n"
"overall_soundscape: Room tone.\n"
"non_diegetic_music: N/A"
)
self.assertIn(
"<d>English</d> This text is outside the dialogue tag.",
module.ensure_video_prompt_duration(
malformed_dialogue,
"🎬 MiniMax H3 Text-to-Video",
8.0,
),
)
impossible_dialogue = (
"integrated_multimodal_description: [Shot 1] Peter (S1) says, "
"<d>[English] " + "word " * 40 + "</d>\n"
"overall_soundscape: Room tone.\n"
"non_diegetic_music: N/A"
)
self.assertIn(
"word word word",
module.ensure_video_prompt_duration(
impossible_dialogue,
"🎬 MiniMax H3 Text-to-Video",
5.0,
),
)
missing_speaker_id = (
"integrated_multimodal_description: [Shot 1] Peter says, "
"<d>[English] This is still forwarded without an S1 marker.</d>\n"
"overall_soundscape: Room tone.\n"
"non_diegetic_music: N/A"
)
self.assertIn(
"forwarded without an S1 marker",
module.ensure_video_prompt_duration(
missing_speaker_id,
"🎬 MiniMax H3 Text-to-Video",
8.0,
),
)
self.assertTrue(
module.ensure_video_prompt_duration(
"duration: 8 seconds\n" + base_body,
"🎬 MiniMax H3 Text-to-Video",
8.0,
).startswith("duration: 8 seconds")
)
def test_minimax_dialogue_preflight_fidelity_and_format_recovery(self):
stub_modules = build_loader_test_stubs()
stub_modules.pop("AILab_QwenVL", None)
with mock.patch.dict(sys.modules, stub_modules, clear=False):
module = load_module_from_file("AILab_QwenVL.py", "thinkingllm_minimax_dialogue_preflight_test")
source = (
"Peter looks at the screen.\n\n"
"He says:\n\n"
"“Hello there.”\n\n"
"The screen reads “STATUS”.\n\n"
"“This is still the original second line!”"
)
self.assertEqual(
module.extract_minimax_source_dialogue(source),
["Hello there.", "This is still the original second line!"],
)
too_long_source = "Peter says:\n\n“" + "word " * 100 + "”"
self.assertEqual(
module.validate_minimax_source_duration(
"🎬 MiniMax H3 Text-to-Video",
7.0,
too_long_source,
),
[("word " * 100).strip()],
)
inline_candidate = (
"integrated_multimodal_description: [Shot 1] at 00:00.000, Peter (S1) says, "
"<d>[English] Hello there.</d> Peter (S1) continues, "
"<d>[English] This is still the original second line!</d> "
"[Shot 2] at 00:03.5, the camera cuts to the screen. "
"overall_soundscape: Room tone. non_diegetic_music: N/A"
)
normalized = module.ensure_video_prompt_duration(
inline_candidate,
"🎬 MiniMax H3 Text-to-Video",
8.0,
source_prompt_text=source,
)
self.assertTrue(normalized.startswith("integrated_multimodal_description: [Shot 1] Peter"))
self.assertNotIn("[Shot 1] At", normalized)
self.assertIn("[Shot 2] At 00:03.500,", normalized)
self.assertIn("\n\noverall_soundscape: Room tone.\n\nnon_diegetic_music: N/A", normalized)
unbracketed_language = (
"integrated_multimodal_description: [Shot 1] Peter (S1) says, "
"<d>English Hello there.</d>\n"
"overall_soundscape: Room tone.\n"
"non_diegetic_music: N/A"
)
normalized_language = module.ensure_video_prompt_duration(
unbracketed_language,
"🎬 MiniMax H3 Text-to-Video",
8.0,
source_prompt_text=source,
)
self.assertIn("<d>[English] Hello there.</d>", normalized_language)
self.assertNotIn("<d>English ", normalized_language)
selected_subset = (
"integrated_multimodal_description: [Shot 1] Peter (S1) says, "
"<d>[English] This is still the original second line!</d>\n"
"overall_soundscape: Room tone.\n"
"non_diegetic_music: N/A"
)
self.assertIn(
"This is still the original second line!",
module.ensure_video_prompt_duration(
selected_subset,
"🎬 MiniMax H3 Text-to-Video",
8.0,
source_prompt_text=source,
),
)
missing_dialogue = (
"integrated_multimodal_description: [Shot 1] Peter silently looks at the screen.\n"
"overall_soundscape: Room tone.\n"
"non_diegetic_music: N/A"
)
visual_only = module.ensure_video_prompt_duration(
missing_dialogue,
"🎬 MiniMax H3 Text-to-Video",
8.0,
source_prompt_text=source,
)
self.assertIn("Peter silently looks at the screen", visual_only)
self.assertNotIn("<d>", visual_only)
rewritten_dialogue = selected_subset.replace(
"This is still the original second line!",
"This is a rewritten line!",
)
self.assertIn(
"This is a rewritten line!",
module.ensure_video_prompt_duration(
rewritten_dialogue,
"🎬 MiniMax H3 Text-to-Video",
8.0,
source_prompt_text=source,
),
)
over_duration = (
"integrated_multimodal_description: [Shot 1] Opening. "
"[Shot 2] At 00:10.000, cut to the ending.\n"
"overall_soundscape: Room tone.\n"
"non_diegetic_music: N/A"
)
self.assertIn(
"[Shot 2] At 00:10.000,",
module.ensure_video_prompt_duration(
over_duration,
"🎬 MiniMax H3 Text-to-Video",
8.0,
),
)
def test_minimax_gguf_forwards_best_effort_output_without_repair_pass(self):
source = read_source("AILab_QwenVL_GGUF_PromptEnhancer.py")
self.assertNotIn("is_recoverable_minimax_validation_error", source)
self.assertNotIn("ORIGINAL USER REQUEST:", source)
self.assertNotIn("CANDIDATE TO REPAIR:", source)
self.assertNotIn("MINIMAX STRUCTURE REPAIR", source)
self.assertNotIn("_NON_THINKING_REASONING_CHAR_LIMIT", source)
self.assertIn("final = ensure_video_prompt_duration(", source)
self.assertIn("arch.startswith(\"qwen35\")", source)
self.assertIn('completion_kwargs["chat_template_kwargs"]', source)
self.assertIn('"thinking_mode": "enabled" if current_enable_thinking else "disabled"', source)
for filename in (
"thinkingllm_core/hf_runtime.py",
"AILab_QwenVL_GGUF.py",
"AILab_QwenVL_PromptEnhancer.py",
"AILab_QwenVL_GGUF_PromptEnhancer.py",
):
with self.subTest(filename=filename):
self.assertIn("validate_minimax_source_duration(", read_source(filename))
def test_output_cleaner_preserves_structured_video_prompt_after_thinking(self):
from AILab_OutputCleaner import OutputCleanConfig, clean_model_output
raw = (
"<think>\nFirst, I need to plan the output.\nThen, I will write it.\n</think>\n\n"
"integrated_multimodal_description: Target duration: 7.29 seconds. "
"[Shot 1] Peter clicks to the next slide while the first robot appears, then the camera pans left.\n\n"
"overall_soundscape: Conference ambience and keyboard clicks.\n\n"
"non_diegetic_music: Minimal electronic drone."
)
cleaned = clean_model_output(raw, OutputCleanConfig(mode="prompt"))
self.assertNotIn("<think>", cleaned)
self.assertIn("integrated_multimodal_description:", cleaned)
self.assertIn("[Shot 1] Peter clicks to the next slide", cleaned)
self.assertIn("the first robot appears, then the camera pans left", cleaned)
self.assertLess(cleaned.index("integrated_multimodal_description"), cleaned.index("overall_soundscape"))
self.assertLess(cleaned.index("overall_soundscape"), cleaned.index("non_diegetic_music"))
def test_output_cleaner_still_drops_actual_leading_planning_paragraph(self):
from AILab_OutputCleaner import OutputCleanConfig, clean_model_output
raw = "First: I need to plan this response.\n\nA cinematic close-up of a dancer in rain."
cleaned = clean_model_output(raw, OutputCleanConfig(mode="prompt"))
self.assertEqual(cleaned, "A cinematic close-up of a dancer in rain.")
def test_output_cleaner_returns_only_final_channel_content(self):
from AILab_OutputCleaner import OutputCleanConfig, clean_model_output
raw = (
"<|start|>assistant<|channel|>analysis<|message|>I should inspect the image first."
"<|end|><|start|>assistant<|channel|>final<|message|>The image shows a red fox."
"<|end|>"
)
cleaned = clean_model_output(raw, OutputCleanConfig(mode="text"))
self.assertEqual(cleaned, "The image shows a red fox.")
def test_output_cleaner_removes_mirrored_channel_reasoning_before_plain_final(self):
from AILab_OutputCleaner import OutputCleanConfig, clean_model_output
raw = "<|channel>I should inspect the image first.<channel|>The image shows a red fox."
cleaned = clean_model_output(raw, OutputCleanConfig(mode="text"))
self.assertEqual(cleaned, "The image shows a red fox.")
def test_output_cleaner_drops_channel_reasoning_without_final(self):
from AILab_OutputCleaner import OutputCleanConfig, clean_model_output
raw = "<|start|>assistant<|channel|>analysis<|message|>I should inspect the image first.<|end|>"
cleaned = clean_model_output(raw, OutputCleanConfig(mode="text"))
self.assertEqual(cleaned, "")
def test_output_cleaner_handles_final_only_and_channel_token_variants(self):
from AILab_OutputCleaner import OutputCleanConfig, clean_model_output
cases = (
"<|channel|>final<|message|>Final answer.<|end|>",
"<|CHANNEL>FiNaL<|MESSAGE>Final answer.<|END>",
"<channel|>FINAL<message|>Final answer.<end|>",
)
for raw in cases:
with self.subTest(raw=raw):
cleaned = clean_model_output(raw, OutputCleanConfig(mode="text"))
self.assertEqual(cleaned, "Final answer.")
def test_output_cleaner_keeps_plain_final_after_completed_reasoning(self):
from AILab_OutputCleaner import OutputCleanConfig, clean_model_output
raw = "<|channel|>reasoning<|message|>Private draft.<|end|>Public answer."
cleaned = clean_model_output(raw, OutputCleanConfig(mode="text"))
self.assertEqual(cleaned, "Public answer.")
def test_output_cleaner_uses_last_final_channel(self):
from AILab_OutputCleaner import OutputCleanConfig, clean_model_output
raw = (
"<|channel|>final<|message|>Discarded answer.<|end|>"
"<|channel|>analysis<|message|>I need to correct it.<|end|>"
"<|channel|>final<|message|>Correct answer.<|end|>"
)
cleaned = clean_model_output(raw, OutputCleanConfig(mode="text"))
self.assertEqual(cleaned, "Correct answer.")
def test_output_cleaner_drops_unclosed_channel_reasoning(self):
from AILab_OutputCleaner import OutputCleanConfig, clean_model_output
raw = "<|channel|>thinking<|message|>An unfinished private draft"
cleaned = clean_model_output(raw, OutputCleanConfig(mode="text"))
self.assertEqual(cleaned, "")
def test_output_cleaner_preserves_ambiguous_or_literal_channel_text(self):
from AILab_OutputCleaner import OutputCleanConfig, clean_model_output
cases = (
"<|channel>Final answer only.<channel|>",
"Explain this token: <|channel|>final<|message|> without treating it as protocol.",
)
for raw in cases:
with self.subTest(raw=raw):
cleaned = clean_model_output(raw, OutputCleanConfig(mode="text"))
self.assertEqual(cleaned, raw)
def test_output_cleaner_preserves_channel_protocol_when_think_stripping_is_disabled(self):
from AILab_OutputCleaner import OutputCleanConfig, clean_model_output
raw = "<|channel|>analysis<|message|>Private draft.<|end|>"
cleaned = clean_model_output(raw, OutputCleanConfig(mode="text", strip_think=False))
self.assertEqual(cleaned, raw)
def test_gguf_channel_cleaning_keeps_raw_trace_assignment(self):
source = read_source("AILab_QwenVL_GGUF.py")
self.assertIn("cleaned = clean_model_output(raw_full, OutputCleanConfig(mode=\"text\"))", source)
self.assertIn("raw_for_trace = raw_full", source)
self.assertIn("cleaned = clean_model_output(raw_content, OutputCleanConfig(mode=\"text\"))", source)
self.assertIn("raw_for_trace = raw_content", source)
def test_llm_input_logging_is_complete_and_shared(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file("AILab_QwenVL.py", "thinkingllm_input_logging_test")
long_tail = "END-OF-UNTRUNCATED-PROMPT"
output = io.StringIO()
with contextlib.redirect_stdout(output):
module.log_llm_input(
"Backend",
"FINALIZATION ATTEMPT 2/3",
"Preset Name",
f"user text {long_tail}",
system_text="system text",
formatted_text="<chat>formatted text</chat>",
media={"image": 2, "audio": 1},
)
logged = output.getvalue()
self.assertIn("Preset: Preset Name", logged)
self.assertIn("Stage: FINALIZATION ATTEMPT 2/3", logged)
self.assertIn("system text", logged)
self.assertIn(long_tail, logged)
self.assertIn("<chat>formatted text</chat>", logged)
self.assertIn("image=2, audio=1", logged)
self.assertNotIn("data:image", logged)
self.assertNotIn("base64", logged.lower())
for filename in ("thinkingllm_core/hf_runtime.py", "AILab_QwenVL_GGUF.py", "AILab_QwenVL_PromptEnhancer.py", "AILab_QwenVL_GGUF_PromptEnhancer.py"):
with self.subTest(filename=filename):
source = read_source(filename)
self.assertIn("log_llm_input(", source)
self.assertIn("no LLM call was made", source)
def test_custom_prompt_image_no_preset_alias_is_available_for_saved_workflows(self):
label = "💬 Custom prompt + image (no preset)"
data = json.loads(Path("AILab_System_Prompts.json").read_text(encoding="utf-8"))
self.assertIn(label, data["_preset_prompts"])
self.assertEqual(data["qwenvl"][label], "")
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
package = load_thinkingllm_loader_subset(["AILab_QwenVL_GGUF.py"])
values = package.NODE_CLASS_MAPPINGS["ThinkingLLM_QwenVL_GGUF"].INPUT_TYPES()["required"]["preset_prompt"][0]
self.assertIn("🚫 No preset (image-only)", values)
self.assertIn(label, values)
def test_ltx_presets_are_exposed_by_relevant_nodes(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
package = load_thinkingllm_loader_subset(
[
"AILab_QwenVL.py",
"AILab_QwenVL_GGUF.py",
"AILab_QwenVL_PromptEnhancer.py",
"AILab_QwenVL_GGUF_PromptEnhancer.py",
]
)
expectations = {
"ThinkingLLM_QwenVL": "preset_prompt",
"ThinkingLLM_QwenVL_Advanced": "preset_prompt",
"ThinkingLLM_QwenVL_GGUF": "preset_prompt",
"ThinkingLLM_QwenVL_GGUF_Advanced": "preset_prompt",
"ThinkingLLM_QwenVL_PromptEnhancer": "enhancement_style",
"ThinkingLLM_QwenVL_GGUF_PromptEnhancer": "preset_system_prompt",
}
for node_name, widget_name in expectations.items():
with self.subTest(node_name=node_name):
values = package.NODE_CLASS_MAPPINGS[node_name].INPUT_TYPES()["required"][widget_name][0]
self.assertTrue(any("LTX 2.3" in value for value in values))
def test_preset_tooltip_payload_matches_system_prompts(self):
system_prompts = json.loads(Path("AILab_System_Prompts.json").read_text(encoding="utf-8"))
tooltip_payload = json.loads(Path("web/preset_tooltips.json").read_text(encoding="utf-8"))
expected = {
"preset_prompt": system_prompts["qwenvl"],
"enhancement_style": {
name: entry["system_prompt"]
for name, entry in system_prompts["qwen_text"]["styles"].items()
},
"preset_system_prompt": {
name: entry["system_prompt"]
for name, entry in system_prompts["qwen_text"]["styles"].items()
},
"video_presets": system_prompts["_video_presets"],
}
self.assertEqual(tooltip_payload, expected)
def test_standalone_system_prompt_preset_node(self):
data = json.loads((PKG / "AILab_System_Prompts.json").read_text(encoding="utf-8"))
module = load_module_from_file(
"nodes/system_prompt_preset.py",
"thinkingllm_system_prompt_preset_test",
)
node_name = "ThinkingLLM_SystemPromptPreset"
node_cls = module.NODE_CLASS_MAPPINGS[node_name]
combo = node_cls.INPUT_TYPES()["required"]["preset_prompt"][0]
expected_presets = [
name for name in data["_preset_prompts"] if name in data["qwenvl"]
]
self.assertEqual(combo, expected_presets)
self.assertEqual(node_cls.RETURN_TYPES, ("STRING",))
self.assertEqual(node_cls.RETURN_NAMES, ("system_prompt",))
self.assertEqual(
node_cls().get_system_prompt("🖼️ Detailed Description"),
(data["qwenvl"]["🖼️ Detailed Description"],),
)
self.assertEqual(
node_cls().get_system_prompt(module.NO_PRESET_PROMPT),
("",),
)
legacy_preset = "🍿 Wan 2.2 NSFW T2V Timeline (3s)"
self.assertNotIn(legacy_preset, data["qwenvl"])
self.assertNotIn(legacy_preset, combo)
self.assertEqual(
node_cls().get_system_prompt(legacy_preset),
("",),
)
def test_standalone_system_prompt_preset_node_is_registered_by_all_loaders(self):
node_name = "ThinkingLLM_SystemPromptPreset"
node_files = ["system_prompt_preset.py"]
community_package = load_thinkingllm_loader_subset([], node_files)
self.assertIn(node_name, community_package.NODE_CLASS_MAPPINGS)
with mock.patch.dict(os.environ, {"THINKINGLLM_COMMERCIAL_RELEASE": "1"}):
commercial_package = load_thinkingllm_loader_subset([], node_files)
self.assertIn(node_name, commercial_package.NODE_CLASS_MAPPINGS)
def test_appearance_js_hooks_preset_dropdown_tooltips(self):
source = read_source("web/js/appearance.js")
self.assertIn("preset_tooltips.json", source)
self.assertIn("hookPresetTooltipPreviews(node)", source)
self.assertIn("PRESET_TOOLTIP_NODE_NAMES.has(node.comfyClass)", source)
inference_nodes = re.search(
r"const THINKINGLLM_NODE_NAMES = new Set\(\[(.*?)\]\);",
source,
re.DOTALL,
).group(1)
tooltip_nodes = re.search(
r"const PRESET_TOOLTIP_NODE_NAMES = new Set\(\[(.*?)\]\);",
source,
re.DOTALL,
).group(1)
self.assertNotIn("ThinkingLLM_SystemPromptPreset", inference_nodes)
self.assertIn("ThinkingLLM_SystemPromptPreset", tooltip_nodes)
for widget_name in ["preset_prompt", "enhancement_style", "preset_system_prompt"]:
with self.subTest(widget_name=widget_name):
self.assertIn(f'"{widget_name}"', source)
self.assertIn("updateDurationVisibility(node, promptMap)", source)
self.assertIn('findWidget(node, "duration_seconds")', source)
self.assertIn('fetch(PRESET_TOOLTIPS_URL, { cache: "no-store" })', source)
self.assertIn("findVideoPresetMetadata(payload, selectedPreset)", source)
self.assertIn("metadataAvailable: false", source)
self.assertIn('replaceAll("\\uFE0F", "")', source)
self.assertIn("Boolean(metadata?.duration_required)", source)
self.assertIn("buildVideoPresetRecommendationText(node, payload)", source)
self.assertIn("metadata?.recommended_settings", source)
self.assertIn("Info only: your saved widget values are not changed.", source)
self.assertIn('["model_name", "enable_thinking", ...PRESET_WIDGET_NAMES]', source)
def test_duration_visibility_preserves_widget_type_and_serialization_slot(self):
source = read_source("web/js/appearance.js")
visibility_block = source[
source.index("function setWidgetVisible("):
source.index("function readComboValue(")
]
self.assertNotIn("widget.type =", visibility_block)
self.assertNotIn("widget.draw =", visibility_block)
self.assertIn("widget.computedHeight = 0", visibility_block)
self.assertIn("delete widget.computedHeight", visibility_block)
self.assertIn("repairDurationWidgetValue(durationWidget)", source)
self.assertIn("DEFAULT_DURATION_SECONDS", source)
def test_hf_default_flag_controls_default_model(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
package = load_thinkingllm_loader_subset(["AILab_QwenVL.py"])
required = package.NODE_CLASS_MAPPINGS["ThinkingLLM_QwenVL"].INPUT_TYPES()["required"]
_, meta = required["model_name"]
self.assertEqual(meta["default"], "Qwen3-VL-4B-Instruct-Abliterated [DL: 7.5GB, VRAM: 6.0GB]")
def test_appearance_js_loads_recommendation_widget(self):
source = read_source("web/js/appearance.js")
self.assertIn("model_recommendations.json", source)
self.assertIn("ComfyWidgets", source)
self.assertIn("recommended_settings", source)
self.assertIn("findWidgets(node, RECOMMENDATION_WIDGET_NAME)", source)
self.assertIn("removeDuplicateRecommendationWidgets(node, widget)", source)
self.assertIn("node.widgets.splice(index, 1)", source)
self.assertIn("support_notes?.audio", source)
self.assertIn("AUDIO_UNSUPPORTED_MESSAGE", source)
self.assertIn("else if (hasAudioInput(node))", source)
self.assertIn("ThinkingLLM_Gemma4_Audio_GGUF", source)
self.assertIn("ThinkingLLM_Whisper_ASR", source)
self.assertIn("Whisper ASR transcription", source)
self.assertNotIn("Prompt enhancer compact terminal progress is active", source)
self.assertIn("queueRecommendationRefresh", source)
def test_appearance_js_styles_all_visible_plugin_nodes(self):
source = read_source("web/js/appearance.js")
self.assertIn("if (!node.color && theme.nodeColor)", source)
self.assertIn("if (!node.bgcolor && theme.nodeBgColor)", source)
expected_node_names = [
"AILab_QwenVL",
"AILab_QwenVL_Advanced",
"AILab_QwenVL_PromptEnhancer",
"AILab_QwenVL_GGUF",
"AILab_QwenVL_GGUF_Advanced",
"AILab_QwenVL_GGUF_PromptEnhancer",
"ThinkingLLM_QwenVL",
"ThinkingLLM_QwenVL_Advanced",
"ThinkingLLM_QwenVL_PromptEnhancer",
"ThinkingLLM_QwenVL_GGUF",
"ThinkingLLM_QwenVL_GGUF_Advanced",
"ThinkingLLM_Gemma4_Audio_GGUF",
"ThinkingLLM_Whisper_ASR",
"ThinkingLLM_QwenVL_GGUF_PromptEnhancer",
"ThinkingLLM_SystemPromptPreset",
"VRAMCleanup",
"StorySplitNode",
]
for node_name in expected_node_names:
with self.subTest(node_name=node_name):
self.assertIn(f'"{node_name}"', source)
def test_gguf_prompt_enhancer_caps_context_and_safe_constructs_llama(self):
source = read_source("AILab_QwenVL_GGUF_PromptEnhancer.py")
self.assertIn("_PROMPT_ENHANCER_MAX_CONTEXT_LENGTH = 32768", source)
self.assertIn("_resolve_prompt_enhancer_context_length", source)
self.assertIn("construct_llama_safely(Llama, kwargs", source)
def test_gguf_nodes_guard_failed_llama_construction(self):
source = read_source("AILab_QwenVL_GGUF.py")
self.assertIn("def construct_llama_safely", source)
self.assertIn("_thinkingllm_llama_init_complete", source)
self.assertIn("construct_llama_safely(Llama, llm_kwargs_filtered", source)
def test_gguf_streaming_files_use_shared_extractor(self):
for filename in ("AILab_QwenVL_GGUF.py", "AILab_QwenVL_GGUF_PromptEnhancer.py"):
source = read_source(filename)
with self.subTest(filename=filename):
self.assertIn("extract_stream_token", source, f"{filename} should use shared chunk extraction")
self.assertIn("compose_streamed_model_output", source)
self.assertIn("display_token = reasoning_token + content_token", source)
self.assertNotIn("display_token = reasoning_token or content_token", source)
self.assertNotIn("stream_display.push(display_token)", source)
def test_stream_token_extractor_handles_reasoning_content(self):
from AILab_StreamDisplay import extract_stream_token
# Reasoning chunk
chunk = {"choices": [{"delta": {"reasoning_content": "Let me think...", "content": ""}}]}
self.assertEqual(extract_stream_token(chunk), {"reasoning": "Let me think...", "content": ""})
# Content-only chunk
chunk2 = {"choices": [{"delta": {"reasoning_content": "", "content": "The answer is 42."}}]}
self.assertEqual(extract_stream_token(chunk2), {"reasoning": "", "content": "The answer is 42."})
# Mixed chunk
chunk3 = {"choices": [{"delta": {"reasoning_content": "Hmm...", "content": "OK"}}]}
self.assertEqual(extract_stream_token(chunk3), {"reasoning": "Hmm...", "content": "OK"})
# Empty chunk
self.assertEqual(extract_stream_token({}), {"reasoning": "", "content": ""})
# llama.cpp/OpenAI-compatible variants seen across releases
top_level_reasoning = {"reasoning_content": "thinking..."}
self.assertEqual(extract_stream_token(top_level_reasoning), {"reasoning": "thinking...", "content": ""})
message_chunk = {"choices": [{"message": {"content": "final answer"}}]}
self.assertEqual(extract_stream_token(message_chunk), {"reasoning": "", "content": "final answer"})
object_chunk = types.SimpleNamespace(
choices=[types.SimpleNamespace(delta=types.SimpleNamespace(reasoning_content="why", content=" then"))]
)
self.assertEqual(extract_stream_token(object_chunk), {"reasoning": "why", "content": " then"})
def test_stream_channel_composer_keeps_reasoning_out_of_clean_response(self):
cleaner = load_module_from_file(
"AILab_OutputCleaner.py",
"thinkingllm_output_cleaner_channel_test",
)
stream_helpers = load_module_from_file(
"AILab_StreamDisplay.py",
"thinkingllm_stream_channel_test",
)
reasoning = (
"integrated_multimodal_description: obsolete draft\n\n"
"Wait, I need to rethink the structure."
)
final_prompt = (
"integrated_multimodal_description: [Shot 1] The actual shot.\n\n"
"overall_soundscape: N/A\n\n"
"non_diegetic_music: N/A"
)
raw_trace = stream_helpers.compose_streamed_model_output(reasoning, final_prompt)
cleaned = cleaner.clean_model_output(
raw_trace,
cleaner.OutputCleanConfig(mode="prompt"),
)
self.assertIn(reasoning, raw_trace)
self.assertEqual(cleaned, final_prompt)
self.assertNotIn("obsolete draft", cleaned)
def test_output_cleaner_drops_unclosed_thinking_instead_of_salvaging_it(self):
cleaner = load_module_from_file(
"AILab_OutputCleaner.py",
"thinkingllm_unclosed_thinking_test",
)
truncated = (
"<think>\nThinking Process:\n\n"
"integrated_multimodal_description: obsolete example\n\n"
"1. Analyze the request: I need to"
)
cleaned = cleaner.clean_model_output(
truncated,
cleaner.OutputCleanConfig(mode="prompt"),
)
self.assertEqual(cleaned, "")
def test_compact_stream_display_does_not_flush_unstyled_tail(self):
from AILab_StreamDisplay import TerminalStreamDisplay
import io
display = TerminalStreamDisplay("UnitTest", flush_interval=999, min_chunk_chars=999, compact=True, line_width=24)
stream = io.StringIO()
with contextlib.redirect_stdout(stream):
display.start_stage("STREAMING")
display.push_compact("partial compact token output with visible words")
display.end_stage()
output = stream.getvalue()
self.assertNotIn("[UnitTest]", output)
self.assertNotIn("\r", output)
self.assertNotIn("…", output)
self.assertIn("partial compact token", output)
self.assertIn("output with visible", output)
self.assertIn("words", output)
def test_terminal_display_uses_ascii_safe_typographic_punctuation(self):
from AILab_StreamDisplay import TerminalStreamDisplay
import io
original = "\u201cThat\u2019s useful\u2026 really\u201d \u2014 Peter"
display = TerminalStreamDisplay("UnitTest", compact=True, line_width=80)
stream = io.StringIO()
with contextlib.redirect_stdout(stream):
display.start_stage("STREAMING")
display.push_compact(original)
display.end_stage()
self.assertEqual(stream.getvalue(), '"That\'s useful... really" -- Peter\n')
self.assertEqual(original, "\u201cThat\u2019s useful\u2026 really\u201d \u2014 Peter")
def test_stream_degeneracy_guard_catches_repeated_word_loop(self):
from AILab_StreamDisplay import StreamDegenerationError, StreamDegenerationGuard, strip_degenerate_repetition
guard = StreamDegenerationGuard(repeated_word_limit=5)
with self.assertRaises(StreamDegenerationError):
guard.push("mightiest, mightiest, mightiest, mightiest, mightiest, ")
self.assertEqual(
strip_degenerate_repetition(
"A strong start, mightiest, mightiest, mightiest, mightiest, mightiest",
repeated_word_limit=5,
),
"A strong start",
)
def test_visible_streaming_has_no_reasoning_heartbeat_pollution(self):
for filename in STREAMING_FILES:
source = read_source(filename)
with self.subTest(filename=filename):
if "GGUF" in filename or filename in {"thinkingllm_core/hf_runtime.py", "AILab_QwenVL_PromptEnhancer.py"}:
stream_pushes = [line for line in source.splitlines() if "push_compact" in line]
self.assertTrue(stream_pushes)
self.assertNotIn("last_status_at = _maybe_emit_answer_stream_heartbeat", source)
self.assertNotIn("last_status_at = _maybe_emit_hf_stream_heartbeat", source)
self.assertNotIn("last_status_at = _maybe_emit_hf_prompt_stream_heartbeat", source)
self.assertNotIn("last_status_at = _maybe_emit_prompt_stream_heartbeat", source)
self.assertNotIn("terminal stream shows readable progress", source)
class TestPromptEnhancerMetadata(unittest.TestCase):
def test_hf_prompt_enhancer_metadata_matches_runtime(self):
source = read_source("AILab_QwenVL_PromptEnhancer.py")
self.assertIn('RETURN_TYPES = ("STRING", "STRING")', source)
self.assertIn('RETURN_NAMES = ("ENHANCED_OUTPUT", "RAW_TRACE")', source)
self.assertIn('"stream_tokens_to_terminal": ("BOOLEAN"', source)
self.assertIn('"unique_id": "UNIQUE_ID"', source)
self.assertIn('"extra_pnginfo": "EXTRA_PNGINFO"', source)
def test_raw_trace_still_exposed_across_nodes(self):
expectations = {
"AILab_QwenVL_PromptEnhancer.py": 'RETURN_NAMES = ("ENHANCED_OUTPUT", "RAW_TRACE")',
"AILab_QwenVL_GGUF_PromptEnhancer.py": 'RETURN_NAMES = ("ENHANCED_OUTPUT", "RAW_TRACE")',
"AILab_QwenVL.py": 'RETURN_NAMES = ("RESPONSE", "RAW_TRACE")',
"AILab_QwenVL_GGUF.py": 'RETURN_NAMES = ("RESPONSE", "RAW_TRACE")',
}
for filename, marker in expectations.items():
with self.subTest(filename=filename):
self.assertIn(marker, read_source(filename))
class TestSharedComfyUITextEncoders(unittest.TestCase):
def test_discovers_and_runs_supported_native_text_encoders(self):
filenames = [
"qwen3vl_4b_convrot.safetensors",
"qwen3vl_8b_fp8_scaled.safetensors",
"gemma-3-12b-it-heretic-v2_int8.safetensors",
"qwen3vl_32b_minimax_h3_nvfp4_awq.safetensors",
"qwen3vl_32b_minimax_h3_nvfp4_awq.safetensor",
"clip_l.safetensors",
]
calls = {}
calls["decoded_text"] = "native answer"
class FakeClip:
def tokenize(self, text, **kwargs):
calls["tokenize"] = (text, kwargs)
return "tokens"
def generate(self, tokens, **kwargs):
calls["generate"] = (tokens, kwargs)
return [1, 2, 3]
def decode(self, tokens):
return calls["decoded_text"]
def load_clip(**kwargs):
calls["load"] = kwargs
return FakeClip()
stub_modules = build_loader_test_stubs()
stub_modules.pop("AILab_QwenVL", None)
stub_modules["folder_paths"].get_filename_list = lambda category: filenames if category == "text_encoders" else []
stub_modules["folder_paths"].get_full_path_or_raise = lambda category, filename: str(PKG / filename)
comfy_sd = build_stub_module("comfy.sd", load_clip=load_clip)
stub_modules["comfy"].sd = comfy_sd
stub_modules["comfy.sd"] = comfy_sd
with mock.patch.dict(sys.modules, stub_modules, clear=False):
module = load_module_from_file("AILab_QwenVL.py", "thinkingllm_native_text_encoder_test")
hf_folder_paths = module.load_model_configs.__globals__["folder_paths"]
with mock.patch.object(
hf_folder_paths,
"get_filename_list",
side_effect=lambda category: filenames if category == "text_encoders" else [],
), mock.patch.object(
hf_folder_paths,
"get_full_path_or_raise",
side_effect=lambda category, filename: str(PKG / filename),
):
module.load_model_configs()
qwen_name = "[ComfyUI] qwen3vl_8b_fp8_scaled.safetensors"
gemma_name = "[ComfyUI] gemma-3-12b-it-heretic-v2_int8.safetensors"
self.assertIn("[ComfyUI] qwen3vl_4b_convrot.safetensors", module.HF_VL_MODELS)
self.assertIn(qwen_name, module.HF_VL_MODELS)
self.assertIn(gemma_name, module.HF_VL_MODELS)
self.assertNotIn("[ComfyUI] qwen3vl_32b_minimax_h3_nvfp4_awq.safetensors", module.HF_VL_MODELS)
self.assertNotIn("[ComfyUI] qwen3vl_32b_minimax_h3_nvfp4_awq.safetensor", module.HF_VL_MODELS)
node = module.QwenVLBase()
node.load_model(qwen_name, "unused", "auto", False, "auto", True)
text, raw = node.generate("hello", None, None, 4, 32, 0.7, 0.9, 1, 1.1, model_name=qwen_name, seed=7)
self.assertEqual(("native answer", "native answer"), (text, raw))
self.assertEqual([str(PKG / "qwen3vl_8b_fp8_scaled.safetensors")], calls["load"]["ckpt_paths"])
self.assertTrue(calls["tokenize"][1]["thinking"])
self.assertEqual(7, calls["generate"][1]["seed"])
calls["decoded_text"] = " "
with self.assertRaisesRegex(RuntimeError, "without producing text"):
node.generate("hello", None, None, 4, 32, 0.7, 0.9, 1, 1.1, model_name=qwen_name, seed=8)
with self.assertRaisesRegex(ValueError, "not video"):
node.generate("hello", None, object(), 4, 32, 0.7, 0.9, 1, 1.1, model_name=gemma_name)
class TestHuggingFaceTokenSupport(unittest.TestCase):
TOKEN_NODE_FILES = [
"AILab_QwenVL.py",
"AILab_QwenVL_GGUF.py",
"AILab_QwenVL_PromptEnhancer.py",
"AILab_QwenVL_GGUF_PromptEnhancer.py",
]
def _load_qwenvl_with_stubs(self):
stub_modules = build_loader_test_stubs()
stub_modules.pop("AILab_QwenVL", None)
with mock.patch.dict(sys.modules, stub_modules, clear=False):
return load_module_from_file("AILab_QwenVL.py", "thinkingllm_qwenvl_hf_token_test")
def test_hf_token_widget_exists_in_download_nodes(self):
for filename in self.TOKEN_NODE_FILES:
source = read_source(filename)
with self.subTest(filename=filename):
self.assertIn('"hf_token": ("STRING"', source)
self.assertRegex(source, r'"hf_token": \("STRING", \{[^\n]+"tooltip"')
def test_community_access_policy_only_requires_auth_for_gated_models(self):
module = self._load_qwenvl_with_stubs()
gated = {"commercial_status": "external_gated"}
module.enforce_model_access(gated, "private/repo", local_exists=True)
with mock.patch.dict(module.os.environ, {}, clear=True):
with self.assertRaisesRegex(PermissionError, "no Hugging Face access token"):
module.enforce_model_access(gated, "private/repo", local_exists=False)
module.enforce_model_access(gated, "private/repo", local_exists=False, hf_token="hf_test")
for status in ("unclear", "noncommercial"):
with self.subTest(status=status):
module.enforce_model_access({"commercial_status": status}, "example/model", local_exists=False)
module.enforce_model_access({"commercial_status": status}, "example/model", local_exists=True)
def test_commercial_mode_rejects_every_status_except_cleared(self):
module = self._load_qwenvl_with_stubs()
access_globals = module.enforce_model_access.__globals__
with mock.patch.dict(access_globals, {"COMMERCIAL_RELEASE": True}):
module.enforce_model_access({"commercial_status": "cleared"}, "official/model", local_exists=True)
for info in (
{"commercial_status": "external_gated"},
{"commercial_status": "unclear"},
{"commercial_status": "noncommercial"},
{"is_local": True},
):
with self.subTest(info=info):
with self.assertRaisesRegex(PermissionError, "Commercial mode rejects"):
module.enforce_model_access(info, "blocked/model", local_exists=True)
def test_gated_hf_snapshot_uses_local_copy_without_token_and_requires_token_before_download(self):
module = self._load_qwenvl_with_stubs()
models_globals = module.ensure_model.__globals__
model_folder_paths = models_globals["folder_paths"]
entry = {"repo_id": "private/repo", "commercial_status": "external_gated"}
with tempfile.TemporaryDirectory() as temp_dir:
models_dir = Path(temp_dir)
target = models_dir / "LLM" / "Qwen-VL" / "repo"
target.mkdir(parents=True)
(target / "config.json").write_text("{}", encoding="utf-8")
(target / "model.safetensors").write_bytes(b"weights")
with mock.patch.dict(models_globals, {"HF_ALL_MODELS": {"gated": entry}}), mock.patch.object(
model_folder_paths, "models_dir", str(models_dir)
), mock.patch.object(model_folder_paths, "folder_names_and_paths", {}), mock.patch.dict(
models_globals, {"_download_model_snapshot": mock.Mock(side_effect=AssertionError("local gated model must not download"))}
):
self.assertEqual(str(target), module.ensure_model("gated"))
shutil.rmtree(target)
with mock.patch.dict(models_globals, {"HF_ALL_MODELS": {"gated": entry}}), mock.patch.object(
model_folder_paths, "models_dir", str(models_dir)
), mock.patch.object(model_folder_paths, "folder_names_and_paths", {}), mock.patch.dict(
models_globals, {"_download_model_snapshot": mock.Mock(side_effect=AssertionError("missing token must stop before download"))}
), mock.patch.dict(module.os.environ, {}, clear=True):
with self.assertRaisesRegex(PermissionError, "no Hugging Face access token"):
module.ensure_model("gated", hf_token="")
def test_gated_hf_snapshot_forwards_token_only_to_download(self):
module = self._load_qwenvl_with_stubs()
models_globals = module.ensure_model.__globals__
model_folder_paths = models_globals["folder_paths"]
captured = []
def fake_download(repo_id, target, **kwargs):
captured.append((repo_id, kwargs.get("hf_token")))
target.mkdir(parents=True, exist_ok=True)
(target / "config.json").write_text("{}", encoding="utf-8")
(target / "model.safetensors").write_bytes(b"weights")
with tempfile.TemporaryDirectory() as temp_dir, mock.patch.dict(
models_globals,
{
"HF_ALL_MODELS": {"gated": {"repo_id": "private/repo", "commercial_status": "external_gated"}},
"_download_model_snapshot": mock.Mock(side_effect=fake_download),
},
), mock.patch.object(model_folder_paths, "models_dir", temp_dir), mock.patch.object(
model_folder_paths, "folder_names_and_paths", {}
):
module.ensure_model("gated", hf_token="hf_test_secret")
self.assertEqual([("private/repo", "hf_test_secret")], captured)
def test_hf_catalog_model_is_labeled_and_reused_from_extra_llm_path(self):
module = self._load_qwenvl_with_stubs()
models_globals = module.ensure_model.__globals__
model_folder_paths = models_globals["folder_paths"]
entry = {"repo_id": "example/example-model", "commercial_status": "unclear"}
with tempfile.TemporaryDirectory() as temp_dir:
temp_root = Path(temp_dir)
shared_llm = temp_root / "shared" / "LLM"
target = shared_llm / "Qwen-VL" / "example-model"
target.mkdir(parents=True)
(target / "config.json").write_text("{}", encoding="utf-8")
(target / "model.safetensors").write_bytes(b"weights")
with mock.patch.dict(models_globals, {"HF_ALL_MODELS": {"Example model": entry}}), mock.patch.object(
model_folder_paths,
"models_dir",
str(temp_root / "dev" / "models"),
), mock.patch.object(
model_folder_paths,
"folder_names_and_paths",
{"LLM": ([str(shared_llm)], set())},
), mock.patch.object(
model_folder_paths,
"get_folder_paths",
side_effect=lambda name: [str(shared_llm)] if name == "LLM" else [],
), mock.patch.dict(
models_globals,
{"_download_model_snapshot": mock.Mock(side_effect=AssertionError("installed model must not download"))},
):
models_globals["_mark_installed_hf_catalog_models"]()
self.assertTrue(entry["is_installed"])
self.assertEqual(
["Example model [installed]"],
module.model_catalog_options({"Example model": entry}),
)
self.assertEqual(str(target), module.ensure_model("Example model [installed]"))
def test_gated_gguf_stops_before_download_when_token_is_missing(self):
hf_module = self._load_qwenvl_with_stubs()
stubs = build_loader_test_stubs()
stubs["AILab_QwenVL"] = hf_module
with mock.patch.dict(sys.modules, stubs, clear=False):
gguf_module = load_module_from_file("AILab_QwenVL_GGUF.py", "thinkingllm_gated_gguf_policy_test")
with tempfile.TemporaryDirectory() as temp_dir, mock.patch.object(
gguf_module,
"GGUF_VL_CATALOG",
{
"base_dir": temp_dir,
"models": {
"gated.gguf": {
"filename": "gated.gguf",
"repo_id": "private/repo",
"commercial_status": "external_gated",
}
},
},
), mock.patch.object(gguf_module.QwenVLGGUFBase, "_load_backend", return_value=object()), mock.patch.object(
gguf_module, "_download_single_file", side_effect=AssertionError("missing token must stop before GGUF download")
), mock.patch.dict(hf_module.os.environ, {}, clear=True):
with self.assertRaisesRegex(PermissionError, "no Hugging Face access token"):
gguf_module.QwenVLGGUFBase()._load_model(
model_name="gated.gguf",
device="cpu",
ctx=None,
n_batch=None,
n_ubatch=None,
gpu_layers=None,
image_max_tokens=None,
top_k=None,
pool_size=None,
n_threads=None,
n_threads_batch=None,
flash_attn=False,
offload_kqv=False,
ctx_checkpoints=None,
hf_token="",
)
def test_hf_token_is_not_part_of_prompt_or_cache_signatures(self):
for filename in self.TOKEN_NODE_FILES:
tree = parse_source(filename)
with self.subTest(filename=filename):
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
if isinstance(node.func, ast.Name):
function_name = node.func.id
elif isinstance(node.func, ast.Attribute):
function_name = node.func.attr
else:
continue
if function_name not in {"build_node_input_signature", "get_cache_key"}:
continue
keyword_names = {keyword.arg for keyword in node.keywords if keyword.arg}
self.assertNotIn("hf_token", keyword_names)
for arg in node.args:
self.assertNotEqual(getattr(arg, "id", None), "hf_token")
def test_hf_download_prefers_exact_file_and_passes_token(self):
module = self._load_qwenvl_with_stubs()
access_globals = module.download_hf_file_to_path.__globals__
captured_kwargs = []
def fake_hf_hub_download(**kwargs):
captured_kwargs.append(dict(kwargs))
local_dir = Path(kwargs["local_dir"])
local_dir.mkdir(parents=True, exist_ok=True)
downloaded_path = local_dir / kwargs["filename"]
downloaded_path.parent.mkdir(parents=True, exist_ok=True)
downloaded_path.write_bytes(b"GGUF")
return str(downloaded_path)
def fail_snapshot_download(**kwargs):
raise AssertionError("snapshot_download should not run when exact file download succeeds")
with tempfile.TemporaryDirectory() as temp_dir:
target_path = Path(temp_dir) / "model.gguf"
with mock.patch.dict(
access_globals,
{
"hf_hub_download": mock.Mock(side_effect=fake_hf_hub_download),
"snapshot_download": mock.Mock(side_effect=fail_snapshot_download),
},
):
module.download_hf_file_to_path(
["private/repo"],
"model.gguf",
target_path,
hf_token="hf_secret_test_token",
)
self.assertEqual(captured_kwargs[0]["token"], "hf_secret_test_token")
self.assertEqual(captured_kwargs[0]["filename"], "model.gguf")
def test_hf_download_errors_redact_token_and_add_access_hint(self):
module = self._load_qwenvl_with_stubs()
secret = "hf_secret_test_token"
message = module._hf_download_error_message(
RuntimeError(f"401 Unauthorized: {secret}"),
repo_id="private/repo",
hf_token=secret,
)
self.assertNotIn(secret, message)
self.assertIn("<redacted HF token>", message)
self.assertIn("access was rejected", message)
def test_gated_hf_download_without_token_names_both_token_options(self):
module = self._load_qwenvl_with_stubs()
with mock.patch.dict(module.os.environ, {}, clear=True):
message = module._hf_download_error_message(
RuntimeError("401 Unauthorized"),
repo_id="private/repo",
hf_token="",
)
self.assertIn("no Hugging Face access token was supplied", message)
self.assertIn("hf_token field", message)
self.assertIn("HF_TOKEN environment variable", message)
def test_hf_token_can_fall_back_to_environment(self):
module = self._load_qwenvl_with_stubs()
with mock.patch.dict(module.os.environ, {"HF_TOKEN": "hf_env_secret"}, clear=True):
self.assertEqual(module._clean_hf_token(""), "hf_env_secret")
message = module._hf_download_error_message(
RuntimeError("401 Unauthorized: hf_env_secret"),
repo_id="private/repo",
hf_token="",
)
self.assertNotIn("hf_env_secret", message)
self.assertIn("<redacted HF token>", message)
self.assertIn("access was rejected", message)
def test_hf_snapshot_rejects_gguf_only_repo_before_download_and_preserves_target(self):
module = self._load_qwenvl_with_stubs()
access_globals = module._download_model_snapshot.__globals__
snapshot_download_mock = mock.Mock(side_effect=AssertionError("GGUF-only repository must not be downloaded as Transformers"))
class FakeHfApi:
def __init__(self, *args, **kwargs):
pass
def list_repo_files(self, **kwargs):
return ["README.md", "model-Q4_K_M.gguf", "mmproj-f16.gguf"]
with tempfile.TemporaryDirectory() as temp_dir:
target = Path(temp_dir) / "existing-model"
target.mkdir()
sentinel = target / "user-file.gguf"
sentinel.write_bytes(b"preserve me")
with mock.patch.dict(access_globals, {"HfApi": FakeHfApi, "snapshot_download": snapshot_download_mock}):
with self.assertRaisesRegex(RuntimeError, "ThinkingLLM GGUF node"):
module._download_model_snapshot("example/gguf-only", target)
self.assertEqual(b"preserve me", sentinel.read_bytes())
snapshot_download_mock.assert_not_called()
def test_hf_snapshot_stages_validation_and_preserves_existing_files(self):
module = self._load_qwenvl_with_stubs()
access_globals = module._download_model_snapshot.__globals__
download_targets = []
class FakeHfApi:
def __init__(self, *args, **kwargs):
pass
def list_repo_files(self, **kwargs):
return ["config.json", "model.safetensors", "processor_config.json"]
def fake_snapshot_download(**kwargs):
staging = Path(kwargs["local_dir"])
download_targets.append(staging)
staging.mkdir(parents=True)
(staging / "config.json").write_text("{}", encoding="utf-8")
(staging / "processor_config.json").write_text("{}", encoding="utf-8")
(staging / "model.safetensors").write_bytes(b"weights")
with tempfile.TemporaryDirectory() as temp_dir:
target = Path(temp_dir) / "existing-model"
target.mkdir()
sentinel = target / "user-note.txt"
sentinel.write_text("preserve me", encoding="utf-8")
with mock.patch.dict(access_globals, {"HfApi": FakeHfApi, "snapshot_download": mock.Mock(side_effect=fake_snapshot_download)}):
module._download_model_snapshot("example/transformers", target, require_processor=True)
self.assertEqual("preserve me", sentinel.read_text(encoding="utf-8"))
self.assertEqual(b"weights", (target / "model.safetensors").read_bytes())
self.assertEqual(1, len(download_targets))
self.assertNotEqual(target, download_targets[0])
def test_hf_snapshot_validation_failure_does_not_modify_existing_target(self):
module = self._load_qwenvl_with_stubs()
access_globals = module._download_model_snapshot.__globals__
class FakeHfApi:
def __init__(self, *args, **kwargs):
pass
def list_repo_files(self, **kwargs):
return ["config.json", "model.safetensors"]
def fake_incomplete_download(**kwargs):
staging = Path(kwargs["local_dir"])
staging.mkdir(parents=True)
(staging / "config.json").write_text("{}", encoding="utf-8")
with tempfile.TemporaryDirectory() as temp_dir:
target = Path(temp_dir) / "existing-model"
target.mkdir()
sentinel = target / "user-note.txt"
sentinel.write_text("preserve me", encoding="utf-8")
with mock.patch.dict(
access_globals,
{"HfApi": FakeHfApi, "snapshot_download": mock.Mock(side_effect=fake_incomplete_download)},
):
with self.assertRaisesRegex(RuntimeError, "existing target was preserved"):
module._download_model_snapshot("example/incomplete", target)
self.assertEqual([sentinel], list(target.iterdir()))
self.assertEqual("preserve me", sentinel.read_text(encoding="utf-8"))
def test_hf_file_download_failure_raises_redacted_message(self):
module = self._load_qwenvl_with_stubs()
access_globals = module.download_hf_file_to_path.__globals__
secret = "hf_secret_test_token"
def fake_snapshot_download(**kwargs):
raise RuntimeError(f"403 Forbidden: {secret}")
with tempfile.TemporaryDirectory() as temp_dir:
target_path = Path(temp_dir) / "missing.gguf"
with mock.patch.dict(
access_globals,
{
"hf_hub_download": mock.Mock(side_effect=fake_snapshot_download),
"snapshot_download": mock.Mock(side_effect=fake_snapshot_download),
},
):
with self.assertRaises(FileNotFoundError) as raised:
module.download_hf_file_to_path(
["private/repo"],
"missing.gguf",
target_path,
hf_token=secret,
)
message = str(raised.exception)
self.assertNotIn(secret, message)
self.assertIn("<redacted HF token>", message)
self.assertIn("access was rejected", message)
def test_frontend_hides_only_legacy_gguf_advanced_shims(self):
source = read_source("web/js/appearance.js")
for widget_name in [
"legacy_seed_mode",
"legacy_unload_after_run",
]:
with self.subTest(widget_name=widget_name):
self.assertIn(f'"{widget_name}"', source)
internal_widgets_block = source[
source.index("const GGUF_ADVANCED_INTERNAL_WIDGETS = new Set(["):
source.index("]);", source.index("const GGUF_ADVANCED_INTERNAL_WIDGETS = new Set([")):
]
for widget_name in [
"n_ubatch",
"n_threads",
"n_threads_batch",
"flash_attn",
"offload_kqv",
"ctx_checkpoints",
]:
with self.subTest(widget_name=widget_name):
self.assertNotIn(f'"{widget_name}"', internal_widgets_block)
self.assertNotIn('GGUF_ADVANCED_INTERNAL_WIDGETS = new Set([\n "hf_token"', source)
self.assertIn("hideStableWidget", source)
self.assertIn("clearHfTokenAfterExecution", source)
class TestInterruptSupport(unittest.TestCase):
INTERRUPT_FILES = [
"thinkingllm_core/hf_runtime.py",
"AILab_QwenVL_GGUF.py",
"AILab_QwenVL_PromptEnhancer.py",
"AILab_QwenVL_GGUF_PromptEnhancer.py",
]
def test_interrupt_import_present(self):
for filename in self.INTERRUPT_FILES:
source = read_source(filename)
with self.subTest(filename=filename):
self.assertIn("from comfy.model_management import throw_exception_if_processing_interrupted", source)
def test_interrupt_check_in_loops(self):
for filename in self.INTERRUPT_FILES:
source = read_source(filename)
with self.subTest(filename=filename):
self.assertIn("throw_exception_if_processing_interrupted()", source)
def test_interrupt_after_blocking_calls(self):
"""Verify interrupt checks exist after non-streaming generate/create_chat_completion calls."""
for filename in self.INTERRUPT_FILES:
source = read_source(filename)
with self.subTest(filename=filename):
# Every file should have at least one interrupt check outside streaming loops
# (after blocking calls + in stream loops = at least 2 checks per file,
# but simple nodes may have only 1 non-streaming path)
count = source.count("throw_exception_if_processing_interrupted()")
self.assertGreaterEqual(
count, 1,
f"{filename} has only {count} interrupt check(s); expected >= 1"
)
class TestNoResidualGlobals(unittest.TestCase):
def test_no_global_last_saved_prompt_statement(self):
for filename in [
"AILab_QwenVL.py",
"AILab_QwenVL_GGUF.py",
"AILab_QwenVL_PromptEnhancer.py",
"AILab_QwenVL_GGUF_PromptEnhancer.py",
]:
tree = parse_source(filename)
visitor = _GlobalCheck()
visitor.visit(tree)
with self.subTest(filename=filename):
self.assertFalse(visitor.found)
class TestLoaderModuleRegistration(unittest.TestCase):
def test_commercial_loader_registers_only_reviewed_hf_modules(self):
with mock.patch.dict(os.environ, {"THINKINGLLM_COMMERCIAL_RELEASE": "1"}):
package = load_thinkingllm_loader_subset(
[
"AILab_QwenVL.py",
"AILab_QwenVL_PromptEnhancer.py",
"AILab_QwenVL_GGUF.py",
"AILab_WhisperASR.py",
]
)
self.assertIn("ThinkingLLM_QwenVL", package.NODE_CLASS_MAPPINGS)
self.assertIn("ThinkingLLM_QwenVL_PromptEnhancer", package.NODE_CLASS_MAPPINGS)
self.assertNotIn("ThinkingLLM_QwenVL_GGUF", package.NODE_CLASS_MAPPINGS)
self.assertNotIn("ThinkingLLM_Whisper_ASR", package.NODE_CLASS_MAPPINGS)
self.assertIsNone(package.WEB_DIRECTORY)
def test_loader_registers_qwenvl_modules_with_stubbed_dependencies(self):
package = load_thinkingllm_loader_subset(
[
"AILab_QwenVL_GGUF.py",
"AILab_WhisperASR.py",
"AILab_QwenVL_PromptEnhancer.py",
"AILab_QwenVL_GGUF_PromptEnhancer.py",
]
)
for node_name in [
"ThinkingLLM_QwenVL_GGUF",
"ThinkingLLM_QwenVL_GGUF_Advanced",
"ThinkingLLM_Gemma4_Audio_GGUF",
"ThinkingLLM_Whisper_ASR",
"ThinkingLLM_QwenVL_PromptEnhancer",
"ThinkingLLM_QwenVL_GGUF_PromptEnhancer",
]:
with self.subTest(node_name=node_name):
self.assertIn(node_name, package.NODE_CLASS_MAPPINGS)
self.assertNotIn("ThinkingLLM_Qwen3_ASR_GGUF", package.NODE_CLASS_MAPPINGS)
class TestLegacyNodeNameCompatibility(unittest.TestCase):
CURRENT_TO_LEGACY = {
"ThinkingLLM_QwenVL": "AILab_QwenVL",
"ThinkingLLM_QwenVL_Advanced": "AILab_QwenVL_Advanced",
"ThinkingLLM_QwenVL_GGUF": "AILab_QwenVL_GGUF",
"ThinkingLLM_QwenVL_GGUF_Advanced": "AILab_QwenVL_GGUF_Advanced",
"ThinkingLLM_QwenVL_PromptEnhancer": "AILab_QwenVL_PromptEnhancer",
"ThinkingLLM_QwenVL_GGUF_PromptEnhancer": "AILab_QwenVL_GGUF_PromptEnhancer",
}
MAPPING_FILES = {
"AILab_QwenVL.py": [
"ThinkingLLM_QwenVL",
"ThinkingLLM_QwenVL_Advanced",
],
"AILab_QwenVL_GGUF.py": [
"ThinkingLLM_QwenVL_GGUF",
"ThinkingLLM_QwenVL_GGUF_Advanced",
],
"AILab_QwenVL_PromptEnhancer.py": [
"ThinkingLLM_QwenVL_PromptEnhancer",
],
"AILab_QwenVL_GGUF_PromptEnhancer.py": [
"ThinkingLLM_QwenVL_GGUF_PromptEnhancer",
],
}
def test_current_names_remain_canonical(self):
for filename, canonical_names in self.MAPPING_FILES.items():
mappings = extract_mapping_target_names(filename, "NODE_CLASS_MAPPINGS")
for canonical_name in canonical_names:
with self.subTest(filename=filename, canonical_name=canonical_name):
self.assertIn(canonical_name, mappings)
def test_legacy_aliases_map_to_same_runtime_class_name(self):
for filename, canonical_names in self.MAPPING_FILES.items():
mappings = extract_mapping_target_names(filename, "NODE_CLASS_MAPPINGS")
displays = extract_literal_dict(filename, "NODE_DISPLAY_NAME_MAPPINGS")
for canonical_name in canonical_names:
legacy_name = self.CURRENT_TO_LEGACY[canonical_name]
with self.subTest(filename=filename, legacy_name=legacy_name):
self.assertEqual(mappings[canonical_name], mappings[legacy_name])
self.assertEqual(displays[canonical_name], displays[legacy_name])
def test_package_export_search_surface_has_no_duplicate_thinkingllm_rows(self):
package = load_thinkingllm_loader_subset([
"AILab_QwenVL.py",
"AILab_QwenVL_GGUF.py",
"AILab_QwenVL_PromptEnhancer.py",
"AILab_QwenVL_GGUF_PromptEnhancer.py",
"AILab_WhisperASR.py",
])
exported_names = set(package.NODE_CLASS_MAPPINGS)
exported_displays = package.NODE_DISPLAY_NAME_MAPPINGS
self.assertFalse(
[name for name in exported_names if name.startswith("AILab_QwenVL")],
"legacy AILab aliases should not be exported as visible search rows",
)
seen = {}
duplicates = {}
for node_name, display_name in exported_displays.items():
if not str(display_name).startswith("ThinkingLLM"):
continue
if display_name in seen:
duplicates.setdefault(display_name, [seen[display_name]]).append(node_name)
else:
seen[display_name] = node_name
self.assertEqual(duplicates, {})
def test_package_exposes_legacy_replacements_for_old_workflows(self):
package = load_thinkingllm_loader_subset([
"AILab_QwenVL.py",
"AILab_QwenVL_GGUF.py",
"AILab_QwenVL_PromptEnhancer.py",
"AILab_QwenVL_GGUF_PromptEnhancer.py",
])
replacements = getattr(package, "LEGACY_NODE_REPLACEMENTS", {})
expected = {legacy: current for current, legacy in self.CURRENT_TO_LEGACY.items()}
self.assertEqual(replacements, expected)
def test_package_registers_legacy_replacements_when_comfy_manager_is_available(self):
class StubNodeReplace:
def __init__(self, new_node_id, old_node_id):
self.new_node_id = new_node_id
self.old_node_id = old_node_id
class StubManager:
def __init__(self):
self.registered = []
def register(self, node_replace):
self.registered.append((node_replace.old_node_id, node_replace.new_node_id))
manager = StubManager()
prompt_server = type("PromptServer", (), {"instance": types.SimpleNamespace(node_replace_manager=manager)})
stubs = build_loader_test_stubs()
stubs.update({
"server": build_stub_module("server", PromptServer=prompt_server),
"comfy_api": build_stub_module("comfy_api"),
"comfy_api.latest": build_stub_module("comfy_api.latest"),
"comfy_api.latest._io": build_stub_module("comfy_api.latest._io", NodeReplace=StubNodeReplace),
})
with mock.patch.dict(sys.modules, stubs, clear=False):
package = load_thinkingllm_loader_subset([
"AILab_QwenVL.py",
"AILab_QwenVL_GGUF.py",
"AILab_QwenVL_PromptEnhancer.py",
"AILab_QwenVL_GGUF_PromptEnhancer.py",
])
self.assertEqual(dict(manager.registered), package.LEGACY_NODE_REPLACEMENTS)
def test_example_workflow_legacy_types_are_covered(self):
resolved_types = set()
for filename in self.MAPPING_FILES:
mappings = extract_mapping_target_names(filename, "NODE_CLASS_MAPPINGS")
resolved_types.update(mappings.keys())
workflow_files = sorted(WORKFLOW_DIR.glob("*.json"))
self.assertTrue(workflow_files, "expected shipped example workflows")
for workflow_file in workflow_files:
for node_type in iter_workflow_types(workflow_file):
if node_type.startswith("AILab_QwenVL"):
with self.subTest(workflow=workflow_file.name, node_type=node_type):
self.assertIn(node_type, resolved_types)
class TestGGUFAdvancedWorkflowCompatibility(unittest.TestCase):
LEGACY_REQUIRED_PREFIX = [
"model_name",
"device",
"preset_prompt",
"custom_prompt",
"max_tokens",
"temperature",
"top_p",
"repetition_penalty",
"frame_count",
"ctx",
"n_batch",
"gpu_layers",
"image_max_tokens",
"top_k",
"pool_size",
"keep_model_loaded",
"seed",
]
def _load_gguf_advanced_class(self):
package = load_thinkingllm_loader_subset(["AILab_QwenVL_GGUF.py"])
return package.NODE_CLASS_MAPPINGS["ThinkingLLM_QwenVL_GGUF_Advanced"]
def test_gguf_advanced_required_widgets_preserve_legacy_order(self):
node_cls = self._load_gguf_advanced_class()
required_keys = list(node_cls.INPUT_TYPES()["required"].keys())
self.assertEqual(
required_keys[: len(self.LEGACY_REQUIRED_PREFIX)],
self.LEGACY_REQUIRED_PREFIX,
)
self.assertEqual(
required_keys[len(self.LEGACY_REQUIRED_PREFIX): len(self.LEGACY_REQUIRED_PREFIX) + 2],
["legacy_seed_mode", "legacy_unload_after_run"],
)
self.assertGreater(required_keys.index("n_ubatch"), required_keys.index("legacy_unload_after_run"))
def test_gguf_advanced_process_signature_matches_required_widget_order(self):
node_cls = self._load_gguf_advanced_class()
signature = inspect.signature(node_cls.process)
parameter_names = [
name
for name, parameter in signature.parameters.items()
if name != "self" and parameter.kind in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY)
]
expected_prefix = self.LEGACY_REQUIRED_PREFIX + [
"legacy_seed_mode",
"legacy_unload_after_run",
"n_ubatch",
"n_threads",
"n_threads_batch",
"flash_attn",
"offload_kqv",
"ctx_checkpoints",
]
self.assertEqual(parameter_names[: len(expected_prefix)], expected_prefix)
def test_gguf_advanced_accepts_legacy_serialized_values(self):
node_cls = self._load_gguf_advanced_class()
required = node_cls.INPUT_TYPES()["required"]
legacy_seed_mode_options, legacy_seed_mode_meta = required["legacy_seed_mode"]
n_ubatch_type, n_ubatch_meta = required["n_ubatch"]
self.assertIn(False, legacy_seed_mode_options)
self.assertEqual(legacy_seed_mode_meta["default"], "fixed")
self.assertEqual(n_ubatch_type, "INT")
self.assertEqual(n_ubatch_meta["min"], 0)
def test_gguf_runtime_normalizes_legacy_zero_ubatch(self):
source = read_source("AILab_QwenVL_GGUF.py")
self.assertIn("n_ubatch_val = int(n_ubatch) if n_ubatch is not None else 0", source)
self.assertIn("if n_ubatch_val <= 0:", source)
self.assertIn("n_ubatch_val = min(n_batch_val, 512)", source)
def test_gguf_vision_nodes_accept_optional_audio_without_legacy_order_change(self):
node_cls = self._load_gguf_advanced_class()
input_types = node_cls.INPUT_TYPES()
self.assertEqual(input_types["optional"]["audio"], ("AUDIO",))
simple_cls = load_thinkingllm_loader_subset(["AILab_QwenVL_GGUF.py"]).NODE_CLASS_MAPPINGS["ThinkingLLM_QwenVL_GGUF"]
self.assertEqual(simple_cls.INPUT_TYPES()["optional"]["audio"], ("AUDIO",))
def test_thinkingllm_node_input_capabilities_are_intentional(self):
package = load_thinkingllm_loader_subset([
"AILab_QwenVL.py",
"AILab_QwenVL_GGUF.py",
"AILab_QwenVL_PromptEnhancer.py",
"AILab_QwenVL_GGUF_PromptEnhancer.py",
"AILab_WhisperASR.py",
])
expected_optional_inputs = {
"ThinkingLLM_QwenVL": {"image", "video", "mask", "duration_seconds"},
"ThinkingLLM_QwenVL_Advanced": {"image", "video", "mask", "duration_seconds"},
"ThinkingLLM_QwenVL_PromptEnhancer": {"duration_seconds"},
"ThinkingLLM_QwenVL_GGUF": {"image", "video", "mask", "mask_mode", "audio", "audio_file_path", "duration_seconds"},
"ThinkingLLM_QwenVL_GGUF_Advanced": {"image", "video", "mask", "mask_mode", "audio", "audio_file_path", "duration_seconds"},
"ThinkingLLM_Gemma4_Audio_GGUF": {"audio", "audio_file_path"},
"ThinkingLLM_Whisper_ASR": {"audio", "audio_file_path"},
"ThinkingLLM_QwenVL_GGUF_PromptEnhancer": {"duration_seconds"},
}
for node_name, expected in expected_optional_inputs.items():
with self.subTest(node_name=node_name):
input_types = package.NODE_CLASS_MAPPINGS[node_name].INPUT_TYPES()
self.assertEqual(set(input_types.get("optional", {})), expected)
if "duration_seconds" in expected:
duration_type, duration_meta = input_types["optional"]["duration_seconds"]
self.assertEqual(duration_type, "FLOAT")
self.assertEqual(duration_meta["default"], 5.0)
self.assertEqual(duration_meta["min"], 0.2)
self.assertEqual(duration_meta["max"], 150.0)
self.assertEqual(duration_meta["step"], 0.1)
def test_gguf_vision_selector_includes_google_gemma4_e4b_image_model(self):
package = load_thinkingllm_loader_subset(["AILab_QwenVL_GGUF.py"])
expected_model_file = "google_gemma-4-E4B-it-Q4_K_M.gguf"
for node_name in ("ThinkingLLM_QwenVL_GGUF", "ThinkingLLM_QwenVL_GGUF_Advanced"):
with self.subTest(node_name=node_name):
model_keys = package.NODE_CLASS_MAPPINGS[node_name].INPUT_TYPES()["required"]["model_name"][0]
self.assertTrue(
any(expected_model_file in key for key in model_keys),
f"{expected_model_file} should be available in the image-capable GGUF selector",
)
def test_catalog_model_is_marked_installed_and_old_display_name_still_resolves(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file("AILab_QwenVL_GGUF.py", "thinkingllm_installed_catalog_test")
with tempfile.TemporaryDirectory() as temp_dir:
models_dir = Path(temp_dir)
gguf_dir = models_dir / "LLM" / "GGUF"
gguf_dir.mkdir(parents=True)
(gguf_dir / "model.gguf").write_bytes(b"GGUF")
config_path = models_dir / "gguf_models.json"
config_path.write_text(json.dumps({
"base_dir": "LLM/GGUF",
"repos": {
"test": {
"repo_id": "example/test",
"mmproj_file": "mmproj.gguf",
"model_files": ["model.gguf"],
}
},
}), encoding="utf-8")
with mock.patch.object(module, "GGUF_CONFIG_PATH", config_path):
with mock.patch.object(module.folder_paths, "models_dir", str(models_dir)):
catalog = module._load_gguf_vl_catalog()
self.assertIn("model.gguf", catalog["models"])
self.assertTrue(catalog["models"]["model.gguf"]["is_installed"])
self.assertNotIn("[local] model.gguf", catalog["models"])
self.assertIn("model.gguf [installed]", module.model_catalog_options(catalog["models"]))
with mock.patch.object(module, "GGUF_VL_CATALOG", catalog):
resolved = module._resolve_model_entry("model.gguf")
resolved_from_label = module._resolve_model_entry("model.gguf [installed]")
self.assertEqual(resolved.model_filename, "model.gguf")
self.assertEqual(resolved_from_label.model_filename, "model.gguf")
def test_gguf_vision_advanced_allows_larger_reasoning_output_budget(self):
node_cls = self._load_gguf_advanced_class()
max_tokens_type, max_tokens_meta = node_cls.INPUT_TYPES()["required"]["max_tokens"]
self.assertEqual(max_tokens_type, "INT")
self.assertGreaterEqual(max_tokens_meta["max"], 32768)
def test_simple_gguf_node_exposes_raw_trace(self):
node_cls = load_thinkingllm_loader_subset(["AILab_QwenVL_GGUF.py"]).NODE_CLASS_MAPPINGS["ThinkingLLM_QwenVL_GGUF"]
self.assertEqual(node_cls.RETURN_TYPES, ("STRING", "STRING"))
self.assertEqual(node_cls.RETURN_NAMES, ("RESPONSE", "RAW_TRACE"))
def test_advanced_vision_nodes_expose_mask_preview(self):
package = load_thinkingllm_loader_subset(["AILab_QwenVL.py", "AILab_QwenVL_GGUF.py"])
for node_name in ("ThinkingLLM_QwenVL_Advanced", "ThinkingLLM_QwenVL_GGUF_Advanced"):
with self.subTest(node_name=node_name):
node_cls = package.NODE_CLASS_MAPPINGS[node_name]
self.assertEqual(node_cls.RETURN_TYPES, ("STRING", "STRING", "IMAGE"))
self.assertEqual(node_cls.RETURN_NAMES, ("RESPONSE", "RAW_TRACE", "MASK_PREVIEW"))
def test_gemma4_audio_node_is_audio_only_and_model_filtered(self):
node_cls = load_thinkingllm_loader_subset(["AILab_QwenVL_GGUF.py"]).NODE_CLASS_MAPPINGS["ThinkingLLM_Gemma4_Audio_GGUF"]
input_types = node_cls.INPUT_TYPES()
optional = input_types["optional"]
self.assertEqual(optional["audio"], ("AUDIO",))
self.assertIn("audio_file_path", optional)
self.assertNotIn("image", optional)
self.assertNotIn("video", optional)
model_keys, model_meta = input_types["required"]["model_name"]
self.assertTrue(model_keys, "audio node should expose at least one Gemma 4 audio model")
self.assertTrue(all("gemma-4" in key.lower() for key in model_keys))
self.assertTrue(all("26b" not in key.lower() and "31b" not in key.lower() for key in model_keys))
self.assertIn("gemma-4-12b", model_meta["default"].lower())
self.assertIn("Audio", input_types["required"]["custom_prompt"][1]["default"])
def test_gguf_nodes_disable_auto_finalization_retry_by_default(self):
package = load_thinkingllm_loader_subset([
"AILab_QwenVL_GGUF.py",
"AILab_QwenVL_GGUF_PromptEnhancer.py",
])
for node_name in (
"ThinkingLLM_QwenVL_GGUF",
"ThinkingLLM_QwenVL_GGUF_Advanced",
"ThinkingLLM_Gemma4_Audio_GGUF",
"ThinkingLLM_QwenVL_GGUF_PromptEnhancer",
):
with self.subTest(node_name=node_name):
required = package.NODE_CLASS_MAPPINGS[node_name].INPUT_TYPES()["required"]
self.assertIn("auto_finalization_retry", required)
self.assertEqual(required["auto_finalization_retry"][0], "BOOLEAN")
self.assertEqual(required["auto_finalization_retry"][1]["default"], False)
def test_qwen3_asr_node_is_not_registered(self):
package = load_thinkingllm_loader_subset(["AILab_QwenVL_GGUF.py", "AILab_WhisperASR.py"])
self.assertNotIn("ThinkingLLM_Qwen3_ASR_GGUF", package.NODE_CLASS_MAPPINGS)
def test_whisper_asr_node_is_audio_only(self):
node_cls = load_thinkingllm_loader_subset(["AILab_WhisperASR.py"]).NODE_CLASS_MAPPINGS["ThinkingLLM_Whisper_ASR"]
input_types = node_cls.INPUT_TYPES()
self.assertEqual(input_types["optional"]["audio"], ("AUDIO",))
self.assertIn("audio_file_path", input_types["optional"])
model_keys, model_meta = input_types["required"]["model_size"]
self.assertIn("base", model_keys)
self.assertIn("large-v3", model_keys)
self.assertEqual(model_meta["default"], "small")
self.assertEqual(input_types["required"]["device"][1]["default"], "cpu")
self.assertEqual(input_types["required"]["compute_type"][1]["default"], "int8")
self.assertNotIn("image", input_types["optional"])
self.assertNotIn("video", input_types["optional"])
def test_whisper_asr_returns_install_message_when_backend_missing(self):
node_cls = load_thinkingllm_loader_subset(["AILab_WhisperASR.py"]).NODE_CLASS_MAPPINGS["ThinkingLLM_Whisper_ASR"]
with mock.patch.dict(sys.modules, {"faster_whisper": None}, clear=False):
transcript, segments_json, raw_trace = node_cls().process(
model_size="base",
language="auto",
task="transcribe",
device="auto",
compute_type="auto",
beam_size=5,
vad_filter=True,
audio_file_path="dummy.wav",
)
self.assertIn("faster-whisper is not installed", transcript)
self.assertEqual(segments_json, "[]")
self.assertIn("pip install faster-whisper", raw_trace)
def test_whisper_asr_transcribes_with_mock_backend(self):
module = load_module_from_file(
"AILab_WhisperASR.py",
"thinkingllm_whisper_mock_backend_test",
)
class FakeSegment:
def __init__(self, start, end, text):
self.start = start
self.end = end
self.text = text
fake_info = types.SimpleNamespace(language="de", language_probability=0.99, duration=1.0)
captured = {}
class FakeWhisperModel:
def __init__(self, model_size, device, compute_type, local_files_only=False):
captured["init"] = (model_size, device, compute_type)
def transcribe(self, path, language, task, beam_size, vad_filter):
captured["transcribe"] = (Path(path).suffix, language, task, beam_size, vad_filter)
return [FakeSegment(0.0, 1.0, " Hallo Welt. ")], fake_info
fake_backend = build_stub_module("faster_whisper", WhisperModel=FakeWhisperModel)
with tempfile.TemporaryDirectory() as temp_dir:
source = Path(temp_dir) / "input.wav"
with wave.open(str(source), "wb") as wav:
wav.setnchannels(1)
wav.setsampwidth(2)
wav.setframerate(16000)
wav.writeframes(b"\0\0" * 1600)
with mock.patch.dict(sys.modules, {"faster_whisper": fake_backend}, clear=False):
transcript, segments_json, raw_trace = module.ThinkingLLM_Whisper_ASR().process(
model_size="base",
language="de",
task="transcribe",
device="cpu",
compute_type="int8",
beam_size=3,
vad_filter=True,
audio_file_path=str(source),
)
self.assertEqual(transcript, "Hallo Welt.")
self.assertEqual(json.loads(segments_json)[0]["text"], "Hallo Welt.")
self.assertEqual(captured["init"], ("base", "cpu", "int8"))
self.assertEqual(captured["transcribe"], (".wav", "de", "transcribe", 3, True))
self.assertIn("language=de", raw_trace)
def test_whisper_asr_prefers_cached_model_without_hf_ping(self):
module = load_module_from_file(
"AILab_WhisperASR.py",
"thinkingllm_whisper_cached_model_test",
)
class FakeSegment:
start = 0.0
end = 1.0
text = " cached transcript "
fake_info = types.SimpleNamespace(language="en", language_probability=0.99, duration=1.0)
captured = {}
class FakeWhisperModel:
def __init__(self, model_size, device, compute_type, local_files_only=False):
captured["init"] = (model_size, device, compute_type, local_files_only)
def transcribe(self, path, language, task, beam_size, vad_filter):
return [FakeSegment()], fake_info
fake_backend = build_stub_module("faster_whisper", WhisperModel=FakeWhisperModel)
with tempfile.TemporaryDirectory() as temp_dir:
source = Path(temp_dir) / "input.wav"
with wave.open(str(source), "wb") as wav:
wav.setnchannels(1)
wav.setsampwidth(2)
wav.setframerate(16000)
wav.writeframes(b"\0\0" * 1600)
with mock.patch.dict(sys.modules, {"faster_whisper": fake_backend}, clear=False):
transcript, _, raw_trace = module.ThinkingLLM_Whisper_ASR().process(
model_size="small",
language="auto",
task="transcribe",
device="cpu",
compute_type="int8",
beam_size=5,
vad_filter=True,
audio_file_path=str(source),
)
self.assertEqual(transcript, "cached transcript")
self.assertEqual(captured["init"], ("small", "cpu", "int8", True))
self.assertIn("local_files_only=True", raw_trace)
def test_whisper_asr_downloads_once_when_cache_is_missing(self):
module = load_module_from_file(
"AILab_WhisperASR.py",
"thinkingllm_whisper_cache_miss_test",
)
class FakeSegment:
start = 0.0
end = 1.0
text = " downloaded transcript "
fake_info = types.SimpleNamespace(language="en", language_probability=0.99, duration=1.0)
attempts = []
class FakeWhisperModel:
def __init__(self, model_size, device, compute_type, local_files_only=False):
attempts.append(local_files_only)
if local_files_only:
raise RuntimeError("local_files_only=True and no cached snapshot was found")
def transcribe(self, path, language, task, beam_size, vad_filter):
return [FakeSegment()], fake_info
fake_backend = build_stub_module("faster_whisper", WhisperModel=FakeWhisperModel)
with tempfile.TemporaryDirectory() as temp_dir:
source = Path(temp_dir) / "input.wav"
with wave.open(str(source), "wb") as wav:
wav.setnchannels(1)
wav.setsampwidth(2)
wav.setframerate(16000)
wav.writeframes(b"\0\0" * 1600)
with mock.patch.dict(sys.modules, {"faster_whisper": fake_backend}, clear=False):
transcript, _, raw_trace = module.ThinkingLLM_Whisper_ASR().process(
model_size="small",
language="auto",
task="transcribe",
device="cpu",
compute_type="int8",
beam_size=5,
vad_filter=True,
audio_file_path=str(source),
)
self.assertEqual(transcript, "downloaded transcript")
self.assertEqual(attempts, [True, False])
self.assertIn("local_files_only=False", raw_trace)
self.assertIn("Local Whisper cache miss", raw_trace)
def test_audio_to_wav_base64_resamples_to_16khz_mono(self):
import numpy as real_numpy
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_audio_resample_test",
)
module.np = real_numpy
waveform = real_numpy.stack([
real_numpy.linspace(-0.5, 0.5, 800, dtype=real_numpy.float32),
real_numpy.linspace(0.5, -0.5, 800, dtype=real_numpy.float32),
])
encoded = module._audio_to_wav_base64({"waveform": waveform, "sample_rate": 8000})
self.assertEqual(len(encoded), 1)
with wave.open(io.BytesIO(base64.b64decode(encoded[0])), "rb") as wav:
self.assertEqual(wav.getframerate(), 16000)
self.assertEqual(wav.getnchannels(), 1)
self.assertEqual(wav.getsampwidth(), 2)
self.assertEqual(wav.getnframes(), 1600)
def test_audio_to_wav_base64_rejects_invalid_audio_metadata(self):
import numpy as real_numpy
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_audio_validation_test",
)
module.np = real_numpy
self.assertEqual(module._audio_to_wav_base64({"waveform": real_numpy.zeros((1, 10)), "sample_rate": 0}), [])
self.assertEqual(module._audio_to_wav_base64({"sample_rate": 16000}), [])
def test_audio_file_path_to_wav_base64_decodes_m4a_with_ffmpeg(self):
if shutil.which("ffmpeg") is None:
self.skipTest("ffmpeg is required for audio file path decoding")
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_audio_file_path_test",
)
with tempfile.TemporaryDirectory(prefix="thinkingllm audio ") as temp_dir:
m4a_path = Path(temp_dir) / "sample voice memo.m4a"
subprocess.run(
[
"ffmpeg",
"-hide_banner",
"-v",
"error",
"-f",
"lavfi",
"-i",
"sine=frequency=440:duration=0.25",
"-c:a",
"aac",
str(m4a_path),
],
check=True,
capture_output=True,
)
encoded = module._audio_file_path_to_wav_base64(str(m4a_path))
self.assertEqual(len(encoded), 1)
with wave.open(io.BytesIO(base64.b64decode(encoded[0])), "rb") as wav:
self.assertEqual(wav.getframerate(), 16000)
self.assertEqual(wav.getnchannels(), 1)
self.assertEqual(wav.getsampwidth(), 2)
self.assertGreater(wav.getnframes(), 3000)
self.assertLess(wav.getnframes(), 5000)
def test_audio_nodes_skip_saved_prompt_cache(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_audio_nodes_skip_saved_prompt_cache_test",
)
class DummyLoader(module.QwenVLGGUFBase):
def _load_model(self, *args, **kwargs):
self.chat_handler = object()
self.last_backend_trace = ""
def _encode_media(self, image, video, frame_count):
return []
def _invoke(self, *args, **kwargs):
return "real transcript", "raw transcript"
with mock.patch.object(
module,
"get_node_saved_prompt_with_seed",
side_effect=AssertionError("audio nodes must not reuse saved prompts"),
), mock.patch.object(
module,
"set_node_saved_prompt",
side_effect=AssertionError("audio nodes must not store saved prompts"),
), mock.patch.object(
module,
"_audio_inputs_to_wav_base64",
return_value=(["BASE64WAV"], []),
), mock.patch.object(
module,
"_resolve_model_entry",
return_value=types.SimpleNamespace(context_length=4096),
):
text, raw_trace = DummyLoader().run(
model_name="Gemma-4-12B-it-Q4_K_M.gguf [~7.5GB]",
preset_prompt="🚫 No preset (image-only)",
custom_prompt="Analyze the audio.",
image=None,
video=None,
frame_count=1,
max_tokens=64,
temperature=0.0,
top_p=1.0,
repetition_penalty=1.0,
seed=1,
keep_model_loaded=False,
device="auto",
node_class="ThinkingLLM_Gemma4_Audio_GGUF",
)
self.assertEqual(text, "real transcript")
self.assertEqual(raw_trace, "raw transcript")
class TestLlamaCppInstaller(unittest.TestCase):
def _resolve_linux_install_spec(self, installer, *, py_minor, cuda_version, machine="x86_64"):
with mock.patch.dict(installer.os.environ, {}, clear=True):
with mock.patch.object(
installer.sys,
"version_info",
types.SimpleNamespace(major=3, minor=py_minor),
):
with mock.patch.object(installer.platform, "machine", return_value=machine):
cuda_versions = {
"toolkit": "",
"torch": cuda_version,
"nvidia_smi": "",
"selected": cuda_version,
"selected_source": "torch" if cuda_version else "unknown",
}
with mock.patch.object(installer, "_detect_cuda_versions", return_value=cuda_versions):
return installer._resolve_linux_install_spec()
def test_linux_install_spec_prefers_known_cuda_128_wheels_for_verified_python_tags(self):
installer = load_module_from_file(
"AILab_LlamaCppInstaller.py",
"thinkingllm_llama_installer_linux_spec_test",
)
for py_tag in ("cp310", "cp311", "cp312", "cp313", "cp314"):
with self.subTest(py_tag=py_tag):
install_spec, install_reason = self._resolve_linux_install_spec(
installer,
py_minor=int(py_tag[-2:]),
cuda_version="12.8",
)
expected_spec = (
"https://github.com/JamePeng/llama-cpp-python/releases/download/"
"v0.3.35-cu128-Basic-linux-20260406/"
f"llama_cpp_python-0.3.35+cu128.basic-{py_tag}-{py_tag}-linux_x86_64.whl"
)
self.assertEqual(install_spec, expected_spec)
self.assertEqual(install_reason, "known Linux wheel (torch CUDA 12.8)")
self.assertNotEqual(install_spec, installer.DEFAULT_JAMEPENG_GIT_SPEC)
def test_linux_install_spec_keeps_existing_known_cuda_124_wheel(self):
installer = load_module_from_file(
"AILab_LlamaCppInstaller.py",
"thinkingllm_llama_installer_linux_spec_124_test",
)
install_spec, install_reason = self._resolve_linux_install_spec(
installer,
py_minor=12,
cuda_version="12.4",
)
self.assertEqual(
install_spec,
"https://github.com/JamePeng/llama-cpp-python/releases/download/"
"v0.3.34-cu124-Basic-linux-20260331/"
"llama_cpp_python-0.3.34+cu124.basic-cp312-cp312-linux_x86_64.whl",
)
self.assertEqual(install_reason, "known Linux wheel (torch CUDA 12.4)")
self.assertNotEqual(install_spec, installer.DEFAULT_JAMEPENG_GIT_SPEC)
def test_linux_install_spec_blocks_unverified_combinations_by_default(self):
installer = load_module_from_file(
"AILab_LlamaCppInstaller.py",
"thinkingllm_llama_installer_linux_spec_fallback_test",
)
install_spec, install_reason = self._resolve_linux_install_spec(
installer,
py_minor=11,
cuda_version="12.4",
)
self.assertEqual(install_spec, "")
self.assertEqual(install_reason, "no verified Linux wheel match")
def test_linux_install_spec_source_build_requires_explicit_opt_in(self):
installer = load_module_from_file(
"AILab_LlamaCppInstaller.py",
"thinkingllm_llama_installer_linux_spec_source_opt_in_test",
)
with mock.patch.dict(installer.os.environ, {"THINKINGLLM_LLAMA_CPP_ALLOW_SOURCE_BUILD": "1"}, clear=True):
with mock.patch.object(installer.sys, "version_info", types.SimpleNamespace(major=3, minor=11)):
with mock.patch.object(installer.platform, "machine", return_value="x86_64"):
with mock.patch.object(
installer,
"_detect_cuda_versions",
return_value={"toolkit": "", "torch": "12.4", "nvidia_smi": "", "selected": "12.4", "selected_source": "torch"},
):
install_spec, install_reason = installer._resolve_linux_install_spec()
self.assertEqual(install_spec, installer.DEFAULT_JAMEPENG_GIT_SPEC)
self.assertEqual(install_reason, "source-build fallback (explicit opt-in)")
def test_source_build_env_enables_cuda_when_toolkit_exists(self):
installer = load_module_from_file(
"AILab_LlamaCppInstaller.py",
"thinkingllm_llama_installer_source_env_test",
)
with mock.patch.dict(installer.os.environ, {"CMAKE_ARGS": "-DOTHER=1"}, clear=True):
with mock.patch.object(installer, "_detect_cuda_home_version", return_value="12.8"):
install_env = installer._build_source_install_env()
self.assertIn("-DOTHER=1", install_env["CMAKE_ARGS"])
self.assertIn("-DGGML_CUDA=on", install_env["CMAKE_ARGS"])
self.assertEqual(install_env["FORCE_CMAKE"], "1")
def test_windows_runtime_dll_dirs_include_torch_lib(self):
installer = load_module_from_file(
"AILab_LlamaCppInstaller.py",
"thinkingllm_llama_installer_dirs_test",
)
torch_module = types.SimpleNamespace(__file__=r"C:\stub\torch\__init__.py")
with mock.patch.dict(sys.modules, {"torch": torch_module}, clear=False):
with mock.patch.object(installer.os.path, "isdir", return_value=True):
self.assertEqual(
installer._get_windows_runtime_dll_dirs(),
[r"C:\stub\torch\lib"],
)
def test_windows_long_path_dll_fallback_uses_path(self):
installer = load_module_from_file(
"AILab_LlamaCppInstaller.py",
"thinkingllm_llama_installer_test",
)
with tempfile.TemporaryDirectory() as temp_dir:
original_path = os.environ.get("PATH", "")
def raise_long_path_error(path):
error = FileNotFoundError(f"[WinError 206] path too long: {path}")
error.winerror = 206
raise error
with mock.patch.object(installer.platform, "system", return_value="Windows"):
with mock.patch.object(installer.os, "add_dll_directory", side_effect=raise_long_path_error):
with mock.patch.dict(installer.os.environ, {"PATH": original_path}, clear=False):
with installer._relax_windows_dll_directory_for_long_paths():
handle = installer.os.add_dll_directory(temp_dir)
self.assertTrue(installer.os.environ["PATH"].startswith(temp_dir))
self.assertFalse(installer.os.environ["PATH"].startswith(temp_dir + ";"))
self.assertTrue(hasattr(handle, "close"))
class TestGGUFWindllRuntimeFallback(unittest.TestCase):
def test_gguf_prompt_enhancer_reuses_catalog_model_from_extra_llm_path_without_download(self):
calls = []
class DummyLlama:
def __init__(self, **kwargs):
calls.append(kwargs.get("model_path"))
installer_module = build_stub_module(
"AILab_LlamaCppInstaller",
ensure_llama_cpp_backend=lambda *args, **kwargs: DummyLlama,
format_llama_cpp_backend_info=lambda info=None: "stub backend",
get_last_llama_cpp_backend_info=lambda: {"gpu_offload": True, "vision_handlers": []},
relax_windows_dll_directory_for_long_paths=contextlib.nullcontext,
)
with tempfile.TemporaryDirectory() as temp_dir:
temp_root = Path(temp_dir)
dev_models = temp_root / "dev" / "models"
shared_llm = temp_root / "shared" / "models" / "LLM"
shared_model_dir = shared_llm / "GGUF" / "Qwen" / "Qwen3-4B-GGUF"
shared_model_dir.mkdir(parents=True)
model_filename = "Qwen3-4B-Q4_K_M.gguf"
(shared_model_dir / model_filename).write_bytes(b"GGUF")
stubs = build_loader_test_stubs()
stubs.update(
{
"folder_paths": build_stub_module(
"folder_paths",
models_dir=str(dev_models),
folder_names_and_paths={"LLM": ([str(shared_llm)], set())},
get_folder_paths=lambda name: [str(shared_llm)] if name == "LLM" else [],
),
"AILab_LlamaCppInstaller": installer_module,
}
)
with mock.patch.dict(sys.modules, stubs, clear=False):
vision_module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gguf_prompt_extra_llm_vision_dependency_test",
)
with mock.patch.dict(sys.modules, {"AILab_QwenVL_GGUF": vision_module}, clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF_PromptEnhancer.py",
"thinkingllm_gguf_prompt_extra_llm_reuse_test",
)
model_name = next(
name
for name in module.ThinkingLLM_QwenVL_GGUF_PromptEnhancer.INPUT_TYPES()["required"]["model_name"][0]
if name.startswith(model_filename)
)
self.assertTrue(model_name.endswith(" [installed]"))
loader = module.ThinkingLLM_QwenVL_GGUF_PromptEnhancer()
with mock.patch.dict(sys.modules, stubs, clear=False), mock.patch.object(
module,
"download_hf_file_to_path",
side_effect=AssertionError("download should not be called for shared local GGUF"),
), mock.patch.object(
module,
"read_gguf_architecture",
return_value="qwen3",
):
loader._load_model(
model_name=model_name,
device="cpu",
enable_thinking=True,
unique_id=None,
context_length_override=128,
quiet=True,
)
self.assertEqual(len(calls), 1)
self.assertTrue(str(calls[0]).endswith(model_filename))
def test_gguf_loader_reuses_catalog_model_from_extra_llm_path_without_download(self):
calls = []
class DummyChatHandler:
def __init__(self, **kwargs):
calls.append(("chat", kwargs.get("clip_model_path")))
class DummyLlama:
def __init__(self, **kwargs):
calls.append(("llama", kwargs.get("model_path")))
chat_format_module = build_stub_module(
"llama_cpp.llama_chat_format",
Qwen35VLChatHandler=DummyChatHandler,
Qwen3VLChatHandler=DummyChatHandler,
)
installer_module = build_stub_module(
"AILab_LlamaCppInstaller",
ensure_llama_cpp_backend=lambda *args, **kwargs: DummyLlama,
format_llama_cpp_backend_info=lambda info=None: "stub backend",
get_last_llama_cpp_backend_info=lambda: {
"gpu_offload": True,
"vision_handlers": ["Qwen35VLChatHandler"],
},
relax_windows_dll_directory_for_long_paths=contextlib.nullcontext,
)
with tempfile.TemporaryDirectory() as temp_dir:
temp_root = Path(temp_dir)
dev_models = temp_root / "dev" / "models"
shared_llm = temp_root / "shared" / "models" / "LLM"
shared_model_dir = shared_llm / "GGUF" / "HauhauCS" / "Qwen3.5-4B-Uncensored-HauhauCS-Aggressive"
shared_model_dir.mkdir(parents=True)
model_filename = "Qwen3.5-4B-Uncensored-HauhauCS-Aggressive-Q8_0.gguf"
mmproj_filename = "mmproj-Qwen3.5-4B-Uncensored-HauhauCS-Aggressive-BF16.gguf"
(shared_model_dir / model_filename).write_bytes(b"GGUF")
(shared_model_dir / mmproj_filename).write_bytes(b"mmproj")
stubs = build_loader_test_stubs()
stubs.update(
{
"folder_paths": build_stub_module(
"folder_paths",
models_dir=str(dev_models),
folder_names_and_paths={"LLM": ([str(shared_llm)], set())},
get_folder_paths=lambda name: [str(shared_llm)] if name == "LLM" else [],
),
"AILab_LlamaCppInstaller": installer_module,
"llama_cpp": build_stub_module("llama_cpp", llama_chat_format=chat_format_module),
"llama_cpp.llama_chat_format": chat_format_module,
}
)
with mock.patch.dict(sys.modules, stubs, clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gguf_extra_llm_reuse_test",
)
model_name = next(
name
for name in module.GGUF_VL_CATALOG["models"]
if name.startswith(model_filename)
)
loader = module.QwenVLGGUFBase()
with mock.patch.dict(sys.modules, stubs, clear=False), mock.patch.object(
module,
"_download_single_file",
side_effect=AssertionError("download should not be called for shared local GGUF"),
), mock.patch.object(
module,
"_pick_device",
return_value="cpu",
), mock.patch.object(
module,
"read_gguf_architecture",
return_value="qwen35",
):
loader._load_model(
model_name=model_name,
device="cpu",
ctx=128,
n_batch=16,
n_ubatch=None,
gpu_layers=0,
image_max_tokens=64,
top_k=0,
pool_size=1024,
n_threads=None,
n_threads_batch=None,
flash_attn=False,
offload_kqv=False,
ctx_checkpoints=None,
enable_thinking=True,
unique_id=None,
)
self.assertEqual(calls[0][0], "chat")
self.assertTrue(str(calls[0][1]).endswith(mmproj_filename))
self.assertEqual(calls[1][0], "llama")
self.assertTrue(str(calls[1][1]).endswith(model_filename))
def test_gguf_loader_reuses_google_gemma_alias_from_extra_llm_path_without_download(self):
calls = []
class DummyChatHandler:
def __init__(self, **kwargs):
calls.append(("chat", kwargs.get("clip_model_path")))
class DummyLlama:
def __init__(self, **kwargs):
calls.append(("llama", kwargs.get("model_path")))
chat_format_module = build_stub_module(
"llama_cpp.llama_chat_format",
Gemma4ChatHandler=DummyChatHandler,
)
installer_module = build_stub_module(
"AILab_LlamaCppInstaller",
ensure_llama_cpp_backend=lambda *args, **kwargs: DummyLlama,
format_llama_cpp_backend_info=lambda info=None: "stub backend",
get_last_llama_cpp_backend_info=lambda: {
"gpu_offload": True,
"vision_handlers": ["Gemma4ChatHandler"],
},
relax_windows_dll_directory_for_long_paths=contextlib.nullcontext,
)
with tempfile.TemporaryDirectory() as temp_dir:
temp_root = Path(temp_dir)
dev_models = temp_root / "dev" / "models"
shared_llm = temp_root / "shared" / "models" / "LLM"
shared_model_dir = shared_llm / "GGUF" / "unsloth" / "gemma-4-E4B-it-GGUF"
shared_model_dir.mkdir(parents=True)
local_model_filename = "gemma-4-E4B-it-Q8_0.gguf"
local_mmproj_filename = "mmproj-F16.gguf"
(shared_model_dir / local_model_filename).write_bytes(b"GGUF")
(shared_model_dir / local_mmproj_filename).write_bytes(b"mmproj")
stubs = build_loader_test_stubs()
stubs.update(
{
"folder_paths": build_stub_module(
"folder_paths",
models_dir=str(dev_models),
folder_names_and_paths={"LLM": ([str(shared_llm)], set())},
get_folder_paths=lambda name: [str(shared_llm)] if name == "LLM" else [],
),
"AILab_LlamaCppInstaller": installer_module,
"llama_cpp": build_stub_module("llama_cpp", llama_chat_format=chat_format_module),
"llama_cpp.llama_chat_format": chat_format_module,
}
)
with mock.patch.dict(sys.modules, stubs, clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gguf_google_gemma_alias_reuse_test",
)
model_name = "google_gemma-4-E4B-it-Q8_0.gguf [~8GB]"
loader = module.QwenVLGGUFBase()
with mock.patch.dict(sys.modules, stubs, clear=False), mock.patch.object(
module,
"_download_single_file",
side_effect=AssertionError("download should not be called for shared local GGUF alias"),
), mock.patch.object(
module,
"_pick_device",
return_value="cpu",
), mock.patch.object(
module,
"read_gguf_architecture",
return_value="gemma4",
):
loader._load_model(
model_name=model_name,
device="cpu",
ctx=128,
n_batch=16,
n_ubatch=None,
gpu_layers=0,
image_max_tokens=64,
top_k=0,
pool_size=1024,
n_threads=None,
n_threads_batch=None,
flash_attn=False,
offload_kqv=False,
ctx_checkpoints=None,
enable_thinking=True,
unique_id=None,
)
self.assertEqual(calls[0][0], "chat")
self.assertTrue(str(calls[0][1]).endswith((local_mmproj_filename, "mmproj-google_gemma-4-E4B-it-f16.gguf")))
self.assertEqual(calls[1][0], "llama")
self.assertTrue(str(calls[1][1]).endswith((local_model_filename, "google_gemma-4-E4B-it-Q8_0.gguf")))
def test_gguf_prompt_enhancer_reuses_google_gemma_alias_from_extra_llm_path_without_download(self):
calls = []
class DummyLlama:
def __init__(self, **kwargs):
calls.append(kwargs.get("model_path"))
installer_module = build_stub_module(
"AILab_LlamaCppInstaller",
ensure_llama_cpp_backend=lambda *args, **kwargs: DummyLlama,
format_llama_cpp_backend_info=lambda info=None: "stub backend",
get_last_llama_cpp_backend_info=lambda: {"gpu_offload": True, "vision_handlers": []},
relax_windows_dll_directory_for_long_paths=contextlib.nullcontext,
)
with tempfile.TemporaryDirectory() as temp_dir:
temp_root = Path(temp_dir)
dev_models = temp_root / "dev" / "models"
shared_llm = temp_root / "shared" / "models" / "LLM"
shared_model_dir = shared_llm / "GGUF" / "unsloth" / "gemma-4-E4B-it-GGUF"
shared_model_dir.mkdir(parents=True)
local_model_filename = "gemma-4-E4B-it-Q8_0.gguf"
(shared_model_dir / local_model_filename).write_bytes(b"GGUF")
stubs = build_loader_test_stubs()
stubs.update(
{
"folder_paths": build_stub_module(
"folder_paths",
models_dir=str(dev_models),
folder_names_and_paths={"LLM": ([str(shared_llm)], set())},
get_folder_paths=lambda name: [str(shared_llm)] if name == "LLM" else [],
),
"AILab_LlamaCppInstaller": installer_module,
}
)
with mock.patch.dict(sys.modules, stubs, clear=False):
vision_module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gguf_prompt_google_gemma_vision_dependency_test",
)
with mock.patch.dict(sys.modules, {"AILab_QwenVL_GGUF": vision_module}, clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF_PromptEnhancer.py",
"thinkingllm_gguf_prompt_google_gemma_alias_reuse_test",
)
model_name = "google_gemma-4-E4B-it-Q8_0.gguf [~8GB]"
loader = module.ThinkingLLM_QwenVL_GGUF_PromptEnhancer()
with mock.patch.dict(sys.modules, stubs, clear=False), mock.patch.object(
module,
"download_hf_file_to_path",
side_effect=AssertionError("download should not be called for shared local GGUF alias"),
), mock.patch.object(
module,
"read_gguf_architecture",
return_value="gemma4",
):
loader._load_model(
model_name=model_name,
device="cpu",
enable_thinking=True,
unique_id=None,
context_length_override=128,
quiet=True,
)
self.assertEqual(len(calls), 1)
self.assertTrue(str(calls[0]).endswith(local_model_filename))
def test_gguf_runtime_reuses_windows_dll_relaxation_for_handler_and_llama(self):
enter_events = []
@contextlib.contextmanager
def tracking_relaxation():
enter_events.append("enter")
try:
yield
finally:
enter_events.append("exit")
class Qwen3VLChatHandler:
def __init__(self, **kwargs):
enter_events.append(("chat", sorted(kwargs)))
class DummyLlama:
def __init__(self, **kwargs):
enter_events.append(("llama", sorted(kwargs)))
chat_format_module = build_stub_module(
"llama_cpp.llama_chat_format",
Qwen3VLChatHandler=Qwen3VLChatHandler,
)
installer_module = build_stub_module(
"AILab_LlamaCppInstaller",
ensure_llama_cpp_backend=lambda *args, **kwargs: DummyLlama,
format_llama_cpp_backend_info=lambda info=None: "stub backend",
get_last_llama_cpp_backend_info=lambda: {"gpu_offload": True, "vision_handlers": ["Qwen3VLChatHandler"]},
relax_windows_dll_directory_for_long_paths=tracking_relaxation,
)
stub_modules = build_loader_test_stubs()
stub_modules.update(
{
"AILab_LlamaCppInstaller": installer_module,
"llama_cpp": build_stub_module("llama_cpp", llama_chat_format=chat_format_module),
"llama_cpp.llama_chat_format": chat_format_module,
}
)
with tempfile.TemporaryDirectory() as temp_dir:
model_path = Path(temp_dir) / "model.gguf"
mmproj_path = Path(temp_dir) / "mmproj.gguf"
model_path.write_bytes(b"GGUF")
mmproj_path.write_bytes(b"mmproj")
with mock.patch.dict(sys.modules, stub_modules, clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gguf_runtime_dll_test",
)
loader = module.QwenVLGGUFBase()
resolved = module.GGUFVLResolved(
display_name="stub",
repo_id=None,
alt_repo_ids=[],
author=None,
repo_dirname="stub",
model_filename=str(model_path),
mmproj_filename=str(mmproj_path),
context_length=4096,
image_max_tokens=1024,
n_batch=64,
gpu_layers=0,
top_k=20,
pool_size=1024,
)
with mock.patch.object(module, "_resolve_model_entry", return_value=resolved):
with mock.patch.object(module, "_pick_device", return_value="cpu"):
with mock.patch.object(module, "read_gguf_architecture", return_value="qwen3"):
loader._load_model(
model_name="stub",
device="cpu",
ctx=None,
n_batch=None,
n_ubatch=None,
gpu_layers=None,
image_max_tokens=None,
top_k=None,
pool_size=None,
n_threads=None,
n_threads_batch=None,
flash_attn=False,
offload_kqv=False,
ctx_checkpoints=None,
enable_thinking=True,
unique_id=None,
)
self.assertEqual(enter_events[0:3], [
"enter",
("chat", ["clip_model_path", "force_reasoning", "image_max_tokens", "image_min_tokens", "verbose"]),
"exit",
])
self.assertEqual(enter_events[3], "enter")
self.assertEqual(enter_events[5], "exit")
self.assertEqual(enter_events[4][0], "llama")
self.assertTrue(
{
"chat_handler",
"model_path",
"n_ctx",
"image_max_tokens",
"image_min_tokens",
}.issubset(set(enter_events[4][1]))
)
def test_gemma4_chat_handler_does_not_receive_qwen_force_reasoning_kwarg(self):
captured_chat_kwargs = []
Gemma4ChatHandler = type(
"Gemma4ChatHandler",
(),
{"__init__": lambda self, **kwargs: captured_chat_kwargs.append(dict(kwargs))},
)
class DummyLlama:
def __init__(self, **kwargs):
pass
chat_format_module = build_stub_module(
"llama_cpp.llama_chat_format",
Gemma4ChatHandler=Gemma4ChatHandler,
)
installer_module = build_stub_module(
"AILab_LlamaCppInstaller",
ensure_llama_cpp_backend=lambda *args, **kwargs: DummyLlama,
format_llama_cpp_backend_info=lambda info=None: "stub backend",
get_last_llama_cpp_backend_info=lambda: {"gpu_offload": True, "vision_handlers": ["Gemma4ChatHandler"]},
relax_windows_dll_directory_for_long_paths=contextlib.nullcontext,
)
stub_modules = build_loader_test_stubs()
stub_modules.update(
{
"AILab_LlamaCppInstaller": installer_module,
"llama_cpp": build_stub_module("llama_cpp", llama_chat_format=chat_format_module),
"llama_cpp.llama_chat_format": chat_format_module,
}
)
with tempfile.TemporaryDirectory() as temp_dir:
model_path = Path(temp_dir) / "gemma.gguf"
mmproj_path = Path(temp_dir) / "mmproj.gguf"
model_path.write_bytes(b"GGUF")
mmproj_path.write_bytes(b"mmproj")
with mock.patch.dict(sys.modules, stub_modules, clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gemma4_handler_kwargs_test",
)
loader = module.QwenVLGGUFBase()
resolved = module.GGUFVLResolved(
display_name="gemma",
repo_id=None,
alt_repo_ids=[],
author=None,
repo_dirname="gemma",
model_filename=str(model_path),
mmproj_filename=str(mmproj_path),
context_length=4096,
image_max_tokens=1024,
n_batch=64,
gpu_layers=0,
top_k=20,
pool_size=1024,
)
with mock.patch.object(module, "_resolve_model_entry", return_value=resolved):
with mock.patch.object(module, "_pick_device", return_value="cpu"):
with mock.patch.object(module, "read_gguf_architecture", return_value="gemma4"):
loader._load_model(
model_name="google_gemma-4-E4B-it-Q8_0.gguf",
device="cpu",
ctx=None,
n_batch=None,
n_ubatch=None,
gpu_layers=None,
image_max_tokens=None,
top_k=None,
pool_size=None,
n_threads=None,
n_threads_batch=None,
flash_attn=False,
offload_kqv=False,
ctx_checkpoints=None,
enable_thinking=True,
unique_id=None,
)
self.assertEqual(len(captured_chat_kwargs), 1)
self.assertNotIn("force_reasoning", captured_chat_kwargs[0])
self.assertIn("clip_model_path", captured_chat_kwargs[0])
def test_gguf_loader_passes_current_cuda_device_as_llama_main_gpu(self):
captured_llama_kwargs = []
class DummyLlama:
def __init__(self, **kwargs):
captured_llama_kwargs.append(dict(kwargs))
installer_module = build_stub_module(
"AILab_LlamaCppInstaller",
ensure_llama_cpp_backend=lambda *args, **kwargs: DummyLlama,
format_llama_cpp_backend_info=lambda info=None: "stub backend",
get_last_llama_cpp_backend_info=lambda: {"gpu_offload": True, "vision_handlers": []},
relax_windows_dll_directory_for_long_paths=contextlib.nullcontext,
)
stub_modules = build_loader_test_stubs()
stub_modules["AILab_LlamaCppInstaller"] = installer_module
stub_modules["comfy.model_management"].get_torch_device = lambda: types.SimpleNamespace(type="cuda", index=0)
with tempfile.TemporaryDirectory() as temp_dir:
model_path = Path(temp_dir) / "model.gguf"
model_path.write_bytes(b"GGUF")
with mock.patch.dict(sys.modules, stub_modules, clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gguf_main_gpu_test",
)
loader = module.QwenVLGGUFBase()
resolved = module.GGUFVLResolved(
display_name="model",
repo_id=None,
alt_repo_ids=[],
author=None,
repo_dirname="model",
model_filename=str(model_path),
mmproj_filename=None,
context_length=4096,
image_max_tokens=1024,
n_batch=64,
gpu_layers=-1,
top_k=20,
pool_size=1024,
)
with mock.patch.object(module, "_resolve_model_entry", return_value=resolved):
with mock.patch.object(module, "_pick_device", return_value="cuda"):
with mock.patch.object(module, "read_gguf_architecture", return_value="qwen"):
with mock.patch.object(module.torch.cuda, "is_available", return_value=True):
with mock.patch.object(module.torch.cuda, "current_device", return_value=1, create=True):
with mock.patch.object(module.torch.cuda, "synchronize", return_value=None, create=True):
with mock.patch.object(module.torch.cuda, "empty_cache", return_value=None, create=True):
with mock.patch.object(module.torch.cuda, "ipc_collect", return_value=None, create=True):
loader._load_model(
model_name="model.gguf",
device="cuda",
ctx=None,
n_batch=None,
n_ubatch=None,
gpu_layers=-1,
image_max_tokens=None,
top_k=None,
pool_size=None,
n_threads=None,
n_threads_batch=None,
flash_attn=False,
offload_kqv=False,
ctx_checkpoints=None,
enable_thinking=True,
unique_id=None,
)
self.assertEqual(captured_llama_kwargs[0]["main_gpu"], 0)
self.assertIn("main_gpu=0", loader.last_backend_trace)
def test_gemma4_multimodal_load_uses_compatible_image_ubatch_settings(self):
captured_chat_kwargs = []
captured_llama_kwargs = []
Gemma4ChatHandler = type(
"Gemma4ChatHandler",
(),
{"__init__": lambda self, **kwargs: captured_chat_kwargs.append(dict(kwargs))},
)
class DummyLlama:
def __init__(self, **kwargs):
captured_llama_kwargs.append(dict(kwargs))
chat_format_module = build_stub_module(
"llama_cpp.llama_chat_format",
Gemma4ChatHandler=Gemma4ChatHandler,
)
installer_module = build_stub_module(
"AILab_LlamaCppInstaller",
ensure_llama_cpp_backend=lambda *args, **kwargs: DummyLlama,
format_llama_cpp_backend_info=lambda info=None: "stub backend",
get_last_llama_cpp_backend_info=lambda: {"gpu_offload": False, "vision_handlers": ["Gemma4ChatHandler"]},
relax_windows_dll_directory_for_long_paths=contextlib.nullcontext,
)
stub_modules = build_loader_test_stubs()
stub_modules.update(
{
"AILab_LlamaCppInstaller": installer_module,
"llama_cpp": build_stub_module("llama_cpp", llama_chat_format=chat_format_module),
"llama_cpp.llama_chat_format": chat_format_module,
}
)
with tempfile.TemporaryDirectory() as temp_dir:
model_path = Path(temp_dir) / "google_gemma-4-E4B-it-Q8_0.gguf"
mmproj_path = Path(temp_dir) / "mmproj-F16.gguf"
model_path.write_bytes(b"GGUF")
mmproj_path.write_bytes(b"mmproj")
with mock.patch.dict(sys.modules, stub_modules, clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gemma4_cpu_load_downgrade_test",
)
loader = module.QwenVLGGUFBase()
resolved = module.GGUFVLResolved(
display_name="gemma",
repo_id=None,
alt_repo_ids=[],
author=None,
repo_dirname="gemma",
model_filename=str(model_path),
mmproj_filename=str(mmproj_path),
context_length=131072,
image_max_tokens=4096,
n_batch=512,
gpu_layers=-1,
top_k=64,
pool_size=4194304,
)
with mock.patch.object(module, "_resolve_model_entry", return_value=resolved):
with mock.patch.object(module, "_pick_device", return_value="cuda"):
with mock.patch.object(module, "read_gguf_architecture", return_value="gemma4"):
loader._load_model(
model_name="google_gemma-4-E4B-it-Q8_0.gguf",
device="cuda",
ctx=None,
n_batch=None,
n_ubatch=None,
gpu_layers=-1,
image_max_tokens=None,
top_k=None,
pool_size=None,
n_threads=None,
n_threads_batch=None,
flash_attn=False,
offload_kqv=False,
ctx_checkpoints=None,
enable_thinking=True,
unique_id=None,
)
self.assertEqual(captured_chat_kwargs[0]["image_max_tokens"], 1120)
self.assertEqual(captured_llama_kwargs[0]["n_ctx"], 32768)
self.assertEqual(captured_llama_kwargs[0]["n_batch"], 1120)
self.assertEqual(captured_llama_kwargs[0]["n_ubatch"], 1120)
self.assertEqual(captured_llama_kwargs[0]["image_max_tokens"], 1120)
self.assertEqual(loader.current_context_length, 32768)
self.assertEqual(loader.current_batch_size, 1120)
self.assertEqual(loader.current_image_token_budget, 1120)
self.assertIn("applied Gemma 4 multimodal compatibility settings", loader.last_backend_trace)
def test_gemma4_gpu_multimodal_load_clamps_crash_log_settings(self):
captured_chat_kwargs = []
captured_llama_kwargs = []
Gemma4ChatHandler = type(
"Gemma4ChatHandler",
(),
{"__init__": lambda self, **kwargs: captured_chat_kwargs.append(dict(kwargs))},
)
class DummyLlama:
def __init__(self, **kwargs):
captured_llama_kwargs.append(dict(kwargs))
chat_format_module = build_stub_module(
"llama_cpp.llama_chat_format",
Gemma4ChatHandler=Gemma4ChatHandler,
)
installer_module = build_stub_module(
"AILab_LlamaCppInstaller",
ensure_llama_cpp_backend=lambda *args, **kwargs: DummyLlama,
format_llama_cpp_backend_info=lambda info=None: "stub backend",
get_last_llama_cpp_backend_info=lambda: {"gpu_offload": True, "vision_handlers": ["Gemma4ChatHandler"]},
relax_windows_dll_directory_for_long_paths=contextlib.nullcontext,
)
stub_modules = build_loader_test_stubs()
stub_modules.update(
{
"AILab_LlamaCppInstaller": installer_module,
"llama_cpp": build_stub_module("llama_cpp", llama_chat_format=chat_format_module),
"llama_cpp.llama_chat_format": chat_format_module,
}
)
with tempfile.TemporaryDirectory() as temp_dir:
model_path = Path(temp_dir) / "google_gemma-4-E4B-it-Q8_0.gguf"
mmproj_path = Path(temp_dir) / "mmproj-F16.gguf"
model_path.write_bytes(b"GGUF")
mmproj_path.write_bytes(b"mmproj")
with mock.patch.dict(sys.modules, stub_modules, clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gemma4_gpu_load_downgrade_test",
)
loader = module.QwenVLGGUFBase()
resolved = module.GGUFVLResolved(
display_name="gemma",
repo_id=None,
alt_repo_ids=[],
author=None,
repo_dirname="gemma",
model_filename=str(model_path),
mmproj_filename=str(mmproj_path),
context_length=131072,
image_max_tokens=4096,
n_batch=512,
gpu_layers=-1,
top_k=64,
pool_size=4194304,
)
with mock.patch.object(module, "_resolve_model_entry", return_value=resolved):
with mock.patch.object(module, "_pick_device", return_value="cuda"):
with mock.patch.object(module, "read_gguf_architecture", return_value="gemma4"):
loader._load_model(
model_name="google_gemma-4-E4B-it-Q8_0.gguf",
device="cuda",
ctx=None,
n_batch=None,
n_ubatch=None,
gpu_layers=-1,
image_max_tokens=None,
top_k=None,
pool_size=None,
n_threads=None,
n_threads_batch=None,
flash_attn=True,
offload_kqv=True,
ctx_checkpoints=0,
enable_thinking=True,
unique_id=None,
)
self.assertEqual(captured_chat_kwargs[0]["image_max_tokens"], 1120)
self.assertEqual(captured_llama_kwargs[0]["n_ctx"], 32768)
self.assertEqual(captured_llama_kwargs[0]["n_batch"], 1120)
self.assertEqual(captured_llama_kwargs[0]["n_ubatch"], 1120)
self.assertEqual(captured_llama_kwargs[0]["image_max_tokens"], 1120)
def test_gemma4_cpu_only_multimodal_guard_blocks_native_abort_risk(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gemma4_cpu_abort_guard_test",
)
loader = module.QwenVLGGUFBase()
loader.chat_handler = object()
loader.gguf_arch = "gemma4"
loader.current_backend_gpu_offload = False
loader.current_context_length = 131072
loader.current_batch_size = 512
loader.current_image_token_budget = 4096
with self.assertRaisesRegex(RuntimeError, "Gemma 4 GGUF multimodal generation was blocked"):
loader._invoke(
system_prompt="You are helpful.",
user_prompt="Describe this image.",
images_b64=["BASE64PNG"],
audio_b64=[],
max_tokens=256,
temperature=0.0,
top_p=1.0,
repetition_penalty=1.0,
seed=1,
model_name="google_gemma-4-E4B-it-Q8_0.gguf",
stream_to_terminal=False,
enable_thinking=False,
auto_finalization_retry=False,
top_k=64,
)
def test_gemma4_audio_guard_blocks_unsupported_native_audio_path(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gemma4_audio_abort_guard_test",
)
loader = module.QwenVLGGUFBase()
loader.chat_handler = object()
loader.gguf_arch = "gemma4"
loader.current_backend_gpu_offload = True
loader.current_context_length = 32768
loader.current_batch_size = 1120
loader.current_ubatch_size = 1120
loader.current_image_token_budget = 1120
loader.current_backend_info = {"version": "0.3.35"}
with self.assertRaisesRegex(RuntimeError, "requires llama-cpp-python >=0.3.36"):
loader._invoke(
system_prompt="You are helpful.",
user_prompt="Transcribe this audio.",
images_b64=[],
audio_b64=["BASE64WAV"],
max_tokens=256,
temperature=0.0,
top_p=1.0,
repetition_penalty=1.0,
seed=1,
model_name="google_gemma-4-E4B-it-Q8_0.gguf",
stream_to_terminal=False,
enable_thinking=False,
auto_finalization_retry=False,
top_k=64,
)
def test_gemma4_audio_runs_when_backend_and_mmproj_report_support(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gemma4_audio_supported_backend_test",
)
captured_messages = []
class DummyChatHandler:
def __init__(self):
self.is_support_audio = None
self.initialized_with = None
def _init_mtmd_context(self, llama):
self.initialized_with = llama
self.is_support_audio = True
class DummyLlama:
def reset(self):
return None
def create_chat_completion(self, **kwargs):
captured_messages.append(kwargs["messages"])
return iter([{"choices": [{"delta": {"content": "audio understood"}}]}])
loader = module.QwenVLGGUFBase()
loader.llm = DummyLlama()
loader.chat_handler = DummyChatHandler()
loader.gguf_arch = "gemma4"
loader.current_backend_info = {"version": "0.3.36+cu130"}
loader.current_context_length = 32768
loader.current_batch_size = 1120
loader.current_ubatch_size = 1120
loader.current_image_token_budget = 1120
text, _raw_trace = loader._invoke(
system_prompt="You are helpful.",
user_prompt="Transcribe this audio.",
images_b64=[],
audio_b64=["BASE64WAV"],
max_tokens=256,
temperature=0.0,
top_p=1.0,
repetition_penalty=1.0,
seed=1,
model_name="google_gemma-4-E4B-it-Q8_0.gguf",
stream_to_terminal=False,
enable_thinking=False,
auto_finalization_retry=False,
top_k=64,
)
self.assertEqual(text, "audio understood")
self.assertIs(loader.chat_handler.initialized_with, loader.llm)
audio_item = captured_messages[0][1]["content"][1]
self.assertEqual(audio_item["type"], "input_audio")
self.assertEqual(audio_item["input_audio"], {"data": "BASE64WAV", "format": "wav"})
def test_gemma4_audio_guard_blocks_mmproj_without_audio_support(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gemma4_audio_unsupported_mmproj_test",
)
class DummyChatHandler:
is_support_audio = None
def _init_mtmd_context(self, _llama):
self.is_support_audio = False
loader = module.QwenVLGGUFBase()
loader.llm = object()
loader.chat_handler = DummyChatHandler()
loader.gguf_arch = "gemma4"
loader.current_backend_info = {"version": "0.3.36"}
with self.assertRaisesRegex(RuntimeError, "does not report audio support"):
loader._raise_if_unsafe_native_multimodal_path([], ["BASE64WAV"])
def test_gguf_thinking_toggle_updates_without_reload_signature(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gguf_thinking_toggle_test",
)
source = read_source("AILab_QwenVL_GGUF.py")
self.assertNotIn("ctx_checkpoints_val,\n bool(enable_thinking),", source)
captured_kwargs = []
class DummyLlama:
def __init__(self):
self.chat_template_kwargs = {"enable_thinking": True}
def create_chat_completion(self, **kwargs):
captured_kwargs.append(kwargs)
return {"choices": [{"message": {"content": "ok"}}]}
loader = module.QwenVLGGUFBase()
loader.llm = DummyLlama()
loader.uses_qwen_template_thinking = True
loader._create_chat_completion(
enable_thinking=False,
messages=[],
max_tokens=1,
temperature=0.0,
top_p=1.0,
stream=False,
)
self.assertEqual(loader.llm.chat_template_kwargs["enable_thinking"], False)
self.assertEqual(captured_kwargs[0]["chat_template_kwargs"], {"enable_thinking": False})
def test_create_chat_completion_retries_without_unexpected_kwarg(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gguf_unexpected_kwarg_retry_test",
)
captured_kwargs = []
class DummyLlama:
def create_chat_completion(self, **kwargs):
captured_kwargs.append(dict(kwargs))
if "repeat_last_n" in kwargs:
raise TypeError("Llama.create_chat_completion() got an unexpected keyword argument 'repeat_last_n'")
return {"choices": [{"message": {"content": "ok"}}]}
loader = module.QwenVLGGUFBase()
loader.llm = DummyLlama()
loader.uses_qwen_template_thinking = False
result = loader._create_chat_completion(
enable_thinking=False,
messages=[],
max_tokens=1,
temperature=0.0,
top_p=1.0,
stream=True,
repeat_last_n=48,
)
self.assertEqual(result["choices"][0]["message"]["content"], "ok")
self.assertIn("repeat_last_n", captured_kwargs[0])
self.assertNotIn("repeat_last_n", captured_kwargs[1])
def test_gguf_invoke_resets_before_each_completion_call(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gguf_reset_before_completion_test",
)
class DummyLlama:
def __init__(self):
self.events = []
self.reset_count = 0
def reset(self):
self.events.append("reset")
self.reset_count += 1
def create_chat_completion(self, **kwargs):
self.events.append("completion")
return iter([
{"choices": [{"delta": {"content": "ok"}}]},
])
loader = module.QwenVLGGUFBase()
loader.llm = DummyLlama()
text, raw_trace = loader._invoke(
system_prompt="system",
user_prompt="user",
images_b64=[],
audio_b64=[],
max_tokens=1,
temperature=0.0,
top_p=1.0,
repetition_penalty=1.0,
seed=1,
stream_to_terminal=False,
enable_thinking=False,
)
self.assertEqual(text, "ok")
self.assertIn("ok", raw_trace)
self.assertEqual(loader.llm.reset_count, 1)
self.assertEqual(loader.llm.events, ["reset", "completion"])
def test_gguf_invoke_returns_content_without_reasoning_drafts(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gguf_separate_reasoning_content_test",
)
cleaner = load_module_from_file(
"AILab_OutputCleaner.py",
"thinkingllm_output_cleaner_invoke_test",
)
stream_helpers = load_module_from_file(
"AILab_StreamDisplay.py",
"thinkingllm_stream_helpers_invoke_test",
)
module.OutputCleanConfig = cleaner.OutputCleanConfig
module.clean_model_output = cleaner.clean_model_output
module.extract_stream_token = stream_helpers.extract_stream_token
module.compose_streamed_model_output = stream_helpers.compose_streamed_model_output
final_prompt = (
"integrated_multimodal_description: [Shot 1] The actual shot.\n\n"
"overall_soundscape: N/A\n\n"
"non_diegetic_music: N/A"
)
class DummyLlama:
def reset(self):
pass
def create_chat_completion(self, **kwargs):
return iter([
{"choices": [{"delta": {"reasoning_content": (
"integrated_multimodal_description: obsolete internal draft. "
"Wait, I need to reconsider it."
)}}]},
{"choices": [{"delta": {"content": final_prompt}}]},
])
loader = module.QwenVLGGUFBase()
loader.llm = DummyLlama()
text, raw_trace = loader._invoke(
system_prompt="system",
user_prompt="user",
images_b64=[],
audio_b64=[],
max_tokens=256,
temperature=0.0,
top_p=1.0,
repetition_penalty=1.0,
seed=1,
stream_to_terminal=False,
enable_thinking=True,
)
self.assertEqual(text, final_prompt)
self.assertNotIn("obsolete internal draft", text)
self.assertIn("obsolete internal draft", raw_trace)
self.assertIn(final_prompt, raw_trace)
def test_gguf_video_prompt_enhancer_retries_unfinished_thinking_with_schema_prefill(self):
stubs = build_loader_test_stubs()
stubs["AILab_QwenVL_GGUF"] = build_stub_module(
"AILab_QwenVL_GGUF",
construct_llama_safely=lambda *args, **kwargs: None,
read_gguf_architecture=lambda *args, **kwargs: "qwen35",
register_active_gguf_loader=lambda *args, **kwargs: None,
release_other_gguf_loaders=lambda *args, **kwargs: None,
_filter_kwargs_for_callable=lambda func, kwargs: kwargs,
_find_existing_local_file_in_dirs=lambda *args, **kwargs: None,
_gguf_search_dirs=lambda *args, **kwargs: [],
)
with mock.patch.dict(sys.modules, stubs, clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF_PromptEnhancer.py",
"thinkingllm_gguf_video_direct_retry_test",
)
cleaner = load_module_from_file(
"AILab_OutputCleaner.py",
"thinkingllm_output_cleaner_video_retry_test",
)
stream_helpers = load_module_from_file(
"AILab_StreamDisplay.py",
"thinkingllm_stream_helpers_video_retry_test",
)
module.OutputCleanConfig = cleaner.OutputCleanConfig
module.clean_model_output = cleaner.clean_model_output
module.extract_stream_token = stream_helpers.extract_stream_token
module.compose_streamed_model_output = stream_helpers.compose_streamed_model_output
module.resolve_video_duration = lambda preset, duration: {
"provider": "minimax_h3",
"video_mode": "t2va",
} if "MiniMax" in preset else None
module.apply_qwen_soft_thinking_directive = lambda text, *args, **kwargs: text
class DummyLlama:
def __init__(self):
self.calls = []
def reset(self):
pass
def create_chat_completion(self, **kwargs):
self.calls.append(kwargs)
if len(self.calls) == 1:
return iter([{
"choices": [{"delta": {"content": (
"<think>Thinking Process:\n\n"
"integrated_multimodal_description: obsolete draft\n\n"
"I need to continue"
)}}]
}])
return iter([{
"choices": [{"delta": {"content": (
"<think>\n</think>\n\n"
"integrated_multimodal_description: [Shot 1] The actual shot.\n\n"
"overall_soundscape: N/A\n\n"
"non_diegetic_music: N/A"
)}}]
}])
loader = module.ThinkingLLM_QwenVL_GGUF_PromptEnhancer()
loader.llm = DummyLlama()
loader.supports_qwen_soft_think = True
text, raw_trace = loader._invoke_llama(
system_prompt="system",
user_prompt="original request",
max_tokens=1024,
temperature=0.5,
top_p=0.9,
repetition_penalty=1.1,
seed=1,
stream_to_terminal=False,
enable_thinking=False,
auto_finalization_retry=False,
preset_name="🎬 MiniMax H3 Text-to-Video",
)
self.assertEqual(len(loader.llm.calls), 2)
self.assertTrue(text.startswith("integrated_multimodal_description: [Shot 1]"))
self.assertEqual(text.count("integrated_multimodal_description:"), 1)
self.assertNotIn("obsolete draft", text)
self.assertIn("FINALIZATION ATTEMPT 2/3", raw_trace)
self.assertEqual(loader.llm.calls[1]["messages"][-1], {
"role": "assistant",
"content": "integrated_multimodal_description:",
})
def test_gguf_video_prompt_acceptance_is_not_a_schema_gate(self):
stubs = build_loader_test_stubs()
stubs["AILab_QwenVL_GGUF"] = build_stub_module(
"AILab_QwenVL_GGUF",
construct_llama_safely=lambda *args, **kwargs: None,
read_gguf_architecture=lambda *args, **kwargs: "qwen35",
register_active_gguf_loader=lambda *args, **kwargs: None,
release_other_gguf_loaders=lambda *args, **kwargs: None,
_filter_kwargs_for_callable=lambda func, kwargs: kwargs,
_find_existing_local_file_in_dirs=lambda *args, **kwargs: None,
_gguf_search_dirs=lambda *args, **kwargs: [],
)
with mock.patch.dict(sys.modules, stubs, clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF_PromptEnhancer.py",
"thinkingllm_gguf_video_no_schema_gate_test",
)
module.resolve_video_duration = lambda preset, duration: {
"provider": "minimax_h3",
"video_mode": "t2va",
}
self.assertTrue(module._preset_output_is_usable(
"integrated_multimodal_description: A usable prompt with a minor missing optional section.",
"🎬 MiniMax H3 Text-to-Video",
))
self.assertFalse(module._preset_output_is_usable(
"",
"🎬 MiniMax H3 Text-to-Video",
))
def test_gguf_invoke_does_not_auto_finalize_by_default(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gguf_no_default_finalization_test",
)
class DummyLlama:
def __init__(self):
self.events = []
self.calls = 0
def reset(self):
self.events.append(("reset", self.calls))
def create_chat_completion(self, **kwargs):
self.calls += 1
self.events.append(("completion", self.calls))
return iter([
{"choices": [{"delta": {"reasoning_content": "hidden reasoning only"}}]},
])
loader = module.QwenVLGGUFBase()
loader.llm = DummyLlama()
text, raw_trace = loader._invoke(
system_prompt="system",
user_prompt="user",
images_b64=[],
audio_b64=[],
max_tokens=8,
temperature=0.0,
top_p=1.0,
repetition_penalty=1.0,
seed=1,
stream_to_terminal=False,
enable_thinking=True,
)
self.assertEqual(text, "")
self.assertIn("[STREAMING]", raw_trace)
self.assertNotIn("FINALIZATION ATTEMPT", raw_trace)
self.assertEqual(loader.llm.events, [("reset", 0), ("completion", 1)])
def test_gguf_invoke_resets_before_finalization_retry_completion(self):
with mock.patch.dict(sys.modules, build_loader_test_stubs(), clear=False):
module = load_module_from_file(
"AILab_QwenVL_GGUF.py",
"thinkingllm_gguf_reset_finalization_test",
)
class DummyLlama:
def __init__(self):
self.events = []
self.calls = 0
def reset(self):
self.events.append(("reset", self.calls))
def create_chat_completion(self, **kwargs):
self.calls += 1
self.events.append(("completion", self.calls))
if self.calls == 1:
return iter([
{"choices": [{"delta": {"reasoning_content": "hidden reasoning only"}}]},
])
return iter([
{"choices": [{"delta": {"content": "final answer"}}]},
])
loader = module.QwenVLGGUFBase()
loader.llm = DummyLlama()
text, raw_trace = loader._invoke(
system_prompt="system",
user_prompt="user",
images_b64=[],
audio_b64=[],
max_tokens=8,
temperature=0.0,
top_p=1.0,
repetition_penalty=1.0,
seed=1,
stream_to_terminal=False,
enable_thinking=True,
auto_finalization_retry=True,
)
self.assertEqual(text, "final answer")
self.assertIn("[STREAMING]", raw_trace)
self.assertIn("[FINALIZATION ATTEMPT 2/3]", raw_trace)
self.assertEqual(
loader.llm.events,
[("reset", 0), ("completion", 1), ("reset", 1), ("completion", 2)],
)
class _GlobalCheck(ast.NodeVisitor):
def __init__(self):
self.found = False
def visit_Global(self, node):
if "LAST_SAVED_PROMPT" in node.names:
self.found = True
if __name__ == "__main__":
unittest.main(verbosity=2)
Reference in New Issue View Git Blame Copy Permalink