873 lines
33 KiB
Python
873 lines
33 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from fastvideo.api.errors import ConfigValidationError
|
|
from fastvideo.api.presets import (
|
|
InferencePreset,
|
|
PresetStageSpec,
|
|
get_all_preset_names,
|
|
get_preset,
|
|
get_presets_for_family,
|
|
register_preset,
|
|
validate_preset_selection,
|
|
validate_stage_names,
|
|
validate_stage_overrides,
|
|
)
|
|
|
|
# -------------------------------------------------------------------
|
|
# Fixtures
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture()
|
|
def _isolated_registry():
|
|
"""Run each test with an empty preset registry, restoring after."""
|
|
from fastvideo.api.presets import _PRESET_REGISTRY
|
|
saved = dict(_PRESET_REGISTRY)
|
|
_PRESET_REGISTRY.clear()
|
|
yield
|
|
_PRESET_REGISTRY.clear()
|
|
_PRESET_REGISTRY.update(saved)
|
|
|
|
|
|
_SIMPLE_STAGE = PresetStageSpec(
|
|
name="denoise",
|
|
kind="denoising",
|
|
allowed_overrides=frozenset({"num_inference_steps", "guidance_scale"}),
|
|
)
|
|
|
|
_SR_STAGE = PresetStageSpec(
|
|
name="sr",
|
|
kind="super_resolution",
|
|
allowed_overrides=frozenset({"height_sr", "width_sr"}),
|
|
)
|
|
|
|
_NO_OVERRIDES_STAGE = PresetStageSpec(
|
|
name="encode",
|
|
kind="text_encoding",
|
|
)
|
|
|
|
|
|
def _make_preset(
|
|
name: str = "test_preset",
|
|
version: int = 1,
|
|
model_family: str = "test",
|
|
stage_schemas: tuple[PresetStageSpec, ...] = (_SIMPLE_STAGE, ),
|
|
**kwargs,
|
|
) -> InferencePreset:
|
|
return InferencePreset(
|
|
name=name,
|
|
version=version,
|
|
model_family=model_family,
|
|
stage_schemas=stage_schemas,
|
|
**kwargs,
|
|
)
|
|
|
|
|
|
def _write_minimal_model_index(model_dir: Path, class_name: str) -> None:
|
|
"""Minimal on-disk diffusers repo so registry resolution can read ``_class_name``."""
|
|
model_dir.mkdir(parents=True, exist_ok=True)
|
|
(model_dir / "transformer").mkdir(exist_ok=True)
|
|
(model_dir / "model_index.json").write_text(
|
|
json.dumps({
|
|
"_class_name": class_name,
|
|
"_diffusers_version": "0.39.0"
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _record_registry_warnings(monkeypatch: pytest.MonkeyPatch) -> list[tuple]:
|
|
"""Capture ``fastvideo.registry`` warnings.
|
|
|
|
The ``fastvideo`` logger sets ``propagate = False``, so ``caplog`` never
|
|
sees these records.
|
|
"""
|
|
from fastvideo import registry
|
|
|
|
warnings: list[tuple] = []
|
|
monkeypatch.setattr(registry.logger, "warning", lambda *args, **kwargs: warnings.append(args))
|
|
return warnings
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Registration and lookup
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
class TestRegistration:
|
|
|
|
@pytest.mark.usefixtures("_isolated_registry")
|
|
def test_register_and_get(self) -> None:
|
|
p = _make_preset()
|
|
register_preset(p)
|
|
assert get_preset("test_preset", "test") is p
|
|
|
|
@pytest.mark.usefixtures("_isolated_registry")
|
|
def test_get_with_explicit_version(self) -> None:
|
|
p = _make_preset(version=2)
|
|
register_preset(p)
|
|
assert get_preset("test_preset", "test", version=2) is p
|
|
|
|
@pytest.mark.usefixtures("_isolated_registry")
|
|
def test_get_latest_version(self) -> None:
|
|
p1 = _make_preset(version=1)
|
|
p2 = _make_preset(version=2)
|
|
register_preset(p1)
|
|
register_preset(p2)
|
|
assert get_preset("test_preset", "test") is p2
|
|
|
|
@pytest.mark.usefixtures("_isolated_registry")
|
|
def test_get_missing_raises(self) -> None:
|
|
with pytest.raises(ConfigValidationError, match="unknown preset"):
|
|
get_preset("nope", "test")
|
|
|
|
@pytest.mark.usefixtures("_isolated_registry")
|
|
def test_get_wrong_version_raises(self) -> None:
|
|
register_preset(_make_preset(version=1))
|
|
with pytest.raises(ConfigValidationError, match="version"):
|
|
get_preset("test_preset", "test", version=99)
|
|
|
|
@pytest.mark.usefixtures("_isolated_registry")
|
|
def test_duplicate_raises(self) -> None:
|
|
register_preset(_make_preset())
|
|
with pytest.raises(ValueError, match="Duplicate"):
|
|
register_preset(_make_preset())
|
|
|
|
@pytest.mark.usefixtures("_isolated_registry")
|
|
def test_get_presets_for_family(self) -> None:
|
|
register_preset(_make_preset(name="a"))
|
|
register_preset(_make_preset(name="b"))
|
|
register_preset(_make_preset(name="c", model_family="other"))
|
|
result = get_presets_for_family("test")
|
|
assert {p.name for p in result} == {"a", "b"}
|
|
|
|
@pytest.mark.usefixtures("_isolated_registry")
|
|
def test_get_all_preset_names(self) -> None:
|
|
register_preset(_make_preset(name="beta"))
|
|
register_preset(_make_preset(name="alpha"))
|
|
assert get_all_preset_names() == ["alpha", "beta"]
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Stage-name validation
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
class TestStageNameValidation:
|
|
|
|
def test_valid_stage_name_passes(self) -> None:
|
|
preset = _make_preset(stage_schemas=(_SIMPLE_STAGE, _SR_STAGE))
|
|
validate_stage_names(preset, {"denoise": {}, "sr": {}})
|
|
|
|
def test_unknown_stage_name_raises(self) -> None:
|
|
preset = _make_preset(stage_schemas=(_SIMPLE_STAGE, ))
|
|
with pytest.raises(ConfigValidationError, match="stage_overrides.bogus"):
|
|
validate_stage_names(preset, {"bogus": {}})
|
|
|
|
def test_empty_overrides_passes(self) -> None:
|
|
preset = _make_preset(stage_schemas=(_SIMPLE_STAGE, ))
|
|
validate_stage_names(preset, {})
|
|
|
|
def test_error_lists_valid_stages(self) -> None:
|
|
preset = _make_preset(stage_schemas=(_SIMPLE_STAGE, _SR_STAGE))
|
|
with pytest.raises(ConfigValidationError, match="'denoise'"):
|
|
validate_stage_names(preset, {"nope": {}})
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Stage-override validation
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
class TestStageOverrideValidation:
|
|
|
|
def test_allowed_override_passes(self) -> None:
|
|
preset = _make_preset(stage_schemas=(_SIMPLE_STAGE, ))
|
|
validate_stage_overrides(
|
|
preset,
|
|
{"denoise": {
|
|
"num_inference_steps": 25
|
|
}},
|
|
)
|
|
|
|
def test_disallowed_override_raises(self) -> None:
|
|
preset = _make_preset(stage_schemas=(_SIMPLE_STAGE, ))
|
|
with pytest.raises(
|
|
ConfigValidationError,
|
|
match="stage_overrides.denoise.height",
|
|
):
|
|
validate_stage_overrides(
|
|
preset,
|
|
{"denoise": {
|
|
"height": 720
|
|
}},
|
|
)
|
|
|
|
def test_override_on_stage_with_no_allowed_raises(self) -> None:
|
|
preset = _make_preset(stage_schemas=(_NO_OVERRIDES_STAGE, ))
|
|
with pytest.raises(
|
|
ConfigValidationError,
|
|
match="does not accept overrides",
|
|
):
|
|
validate_stage_overrides(
|
|
preset,
|
|
{"encode": {
|
|
"some_key": 1
|
|
}},
|
|
)
|
|
|
|
def test_empty_override_on_no_allowed_passes(self) -> None:
|
|
preset = _make_preset(stage_schemas=(_NO_OVERRIDES_STAGE, ))
|
|
validate_stage_overrides(preset, {"encode": {}})
|
|
|
|
def test_non_mapping_override_raises(self) -> None:
|
|
preset = _make_preset(stage_schemas=(_SIMPLE_STAGE, ))
|
|
with pytest.raises(ConfigValidationError, match="mapping"):
|
|
validate_stage_overrides(preset, {"denoise": "not a dict"})
|
|
|
|
def test_unknown_stage_still_caught(self) -> None:
|
|
preset = _make_preset(stage_schemas=(_SIMPLE_STAGE, ))
|
|
with pytest.raises(ConfigValidationError, match="unknown"):
|
|
validate_stage_overrides(preset, {"missing_stage": {"a": 1}})
|
|
|
|
def test_error_lists_allowed_overrides(self) -> None:
|
|
preset = _make_preset(stage_schemas=(_SIMPLE_STAGE, ))
|
|
with pytest.raises(ConfigValidationError, match="guidance_scale"):
|
|
validate_stage_overrides(
|
|
preset,
|
|
{"denoise": {
|
|
"bad_key": 1
|
|
}},
|
|
)
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# validate_preset_selection end-to-end
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
class TestValidatePresetSelection:
|
|
|
|
@pytest.mark.usefixtures("_isolated_registry")
|
|
def test_none_preset_returns_none(self) -> None:
|
|
assert validate_preset_selection(None, "test") is None
|
|
|
|
@pytest.mark.usefixtures("_isolated_registry")
|
|
def test_valid_preset_resolves(self) -> None:
|
|
p = _make_preset()
|
|
register_preset(p)
|
|
result = validate_preset_selection("test_preset", "test")
|
|
assert result is p
|
|
|
|
@pytest.mark.usefixtures("_isolated_registry")
|
|
def test_valid_preset_with_overrides(self) -> None:
|
|
p = _make_preset()
|
|
register_preset(p)
|
|
result = validate_preset_selection(
|
|
"test_preset",
|
|
"test",
|
|
stage_overrides={"denoise": {
|
|
"guidance_scale": 2.0
|
|
}},
|
|
)
|
|
assert result is p
|
|
|
|
@pytest.mark.usefixtures("_isolated_registry")
|
|
def test_invalid_preset_raises(self) -> None:
|
|
with pytest.raises(ConfigValidationError, match="unknown"):
|
|
validate_preset_selection("nope", "test")
|
|
|
|
@pytest.mark.usefixtures("_isolated_registry")
|
|
def test_bad_stage_override_raises(self) -> None:
|
|
register_preset(_make_preset())
|
|
with pytest.raises(ConfigValidationError):
|
|
validate_preset_selection(
|
|
"test_preset",
|
|
"test",
|
|
stage_overrides={"denoise": {
|
|
"bad": 1
|
|
}},
|
|
)
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Wan preset integration (uses real registry)
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
class TestWanPresets:
|
|
"""Verify the Wan presets registered from registry.py."""
|
|
|
|
def test_wan_presets_are_registered(self) -> None:
|
|
# Force registration by importing registry.
|
|
import fastvideo.registry # noqa: F401
|
|
presets = get_presets_for_family("wan")
|
|
names = {p.name for p in presets}
|
|
assert "wan_t2v_1_3b" in names
|
|
assert "wan_t2v_14b" in names
|
|
assert "wan_i2v_14b_480p" in names
|
|
assert "wan_2_2_t2v_a14b" in names
|
|
|
|
def test_wan_t2v_1_3b_lookup(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
preset = get_preset("wan_t2v_1_3b", "wan")
|
|
assert preset.model_family == "wan"
|
|
assert preset.workload_type == "t2v"
|
|
assert len(preset.stage_schemas) == 1
|
|
assert preset.stage_schemas[0].name == "denoise"
|
|
assert preset.defaults["height"] == 480
|
|
assert preset.defaults["width"] == 832
|
|
|
|
def test_wan_2_2_allows_dual_guidance(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
preset = get_preset("wan_2_2_t2v_a14b", "wan")
|
|
stage = preset.stage_schemas[0]
|
|
assert "guidance_scale_2" in stage.allowed_overrides
|
|
assert "boundary_ratio" in stage.allowed_overrides
|
|
|
|
def test_wan_stage_override_validation(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
preset = get_preset("wan_t2v_14b", "wan")
|
|
# Valid override.
|
|
validate_stage_overrides(
|
|
preset,
|
|
{"denoise": {
|
|
"num_inference_steps": 25
|
|
}},
|
|
)
|
|
# Invalid override key.
|
|
with pytest.raises(ConfigValidationError):
|
|
validate_stage_overrides(
|
|
preset,
|
|
{"denoise": {
|
|
"height": 1080
|
|
}},
|
|
)
|
|
|
|
def test_wan_model_family_in_registry(self) -> None:
|
|
from fastvideo.registry import get_model_family
|
|
family = get_model_family("Wan-AI/Wan2.1-T2V-1.3B-Diffusers")
|
|
assert family == "wan"
|
|
|
|
def test_more_specific_later_detector_match_keeps_first_and_warns(self, tmp_path, monkeypatch) -> None:
|
|
"""A later detector match that shares more of the path never overrides
|
|
registration order (test_wan_definitions pins that resolution), so the
|
|
choice stays ambiguous and still warns."""
|
|
import fastvideo.registry # noqa: F401
|
|
from fastvideo.registry import get_preset_selection
|
|
warnings = _record_registry_warnings(monkeypatch)
|
|
model_dir = tmp_path / "lucy-edit-custom"
|
|
_write_minimal_model_index(model_dir, "WanPipeline")
|
|
assert get_preset_selection(str(model_dir)) == ("wan_t2v_1_3b", "wan")
|
|
assert len(warnings) == 1
|
|
assert "Multiple models matched" in warnings[0][0]
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# LTX2 preset integration
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
class TestLtx2Presets:
|
|
|
|
def test_ltx2_presets_registered(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
presets = get_presets_for_family("ltx2")
|
|
names = {p.name for p in presets}
|
|
assert names == {
|
|
"ltx2_base",
|
|
"ltx2_3_base",
|
|
"ltx2_distilled",
|
|
"ltx2_two_stage",
|
|
"ltx2_5_dev",
|
|
"ltx2_5_distilled",
|
|
"ltx2_5_distilled_two_stage",
|
|
}
|
|
|
|
def test_ltx2_base_lookup(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
p = get_preset("ltx2_base", "ltx2")
|
|
assert p.workload_type == "t2v"
|
|
assert p.defaults["height"] == 512
|
|
assert p.defaults["width"] == 768
|
|
|
|
def test_ltx2_distilled_fewer_steps(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
p = get_preset("ltx2_distilled", "ltx2")
|
|
assert p.defaults["num_inference_steps"] == 8
|
|
assert p.defaults["guidance_scale"] == 1.0
|
|
|
|
def test_ltx2_two_stage_is_two_stage(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
p = get_preset("ltx2_two_stage", "ltx2")
|
|
assert len(p.stage_schemas) == 2
|
|
assert p.stage_schemas[0].name == "denoise"
|
|
assert p.stage_schemas[1].name == "refine"
|
|
assert p.stage_schemas[1].kind == "refinement"
|
|
|
|
def test_ltx2_two_stage_stage_defaults(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
p = get_preset("ltx2_two_stage", "ltx2")
|
|
refine = p.stage_defaults["refine"]
|
|
# stage-2 refine only supports 2 or 3 denoising steps; preset
|
|
# defaults to 2 (matches gpu_pool.py load_kwargs).
|
|
assert refine["num_inference_steps"] == 2
|
|
assert refine["guidance_scale"] == 1.0
|
|
|
|
def test_ltx2_two_stage_refine_overrides_valid(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
p = get_preset("ltx2_two_stage", "ltx2")
|
|
validate_stage_overrides(p, {"refine": {"num_inference_steps": 3}})
|
|
validate_stage_overrides(p, {"refine": {"guidance_scale": 1.0}})
|
|
validate_stage_overrides(p, {"refine": {"image_crf": 18}})
|
|
validate_stage_overrides(p, {"refine": {"video_position_offset_sec": 2.5}})
|
|
|
|
def test_ltx2_two_stage_rejects_unknown_refine_override(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
from fastvideo.api.errors import ConfigValidationError
|
|
p = get_preset("ltx2_two_stage", "ltx2")
|
|
with pytest.raises(ConfigValidationError):
|
|
validate_stage_overrides(p, {"refine": {"bogus_field": 1}})
|
|
|
|
@pytest.mark.parametrize(
|
|
"directory_name",
|
|
("ltx-2.3-distilled", "ltx2.3-distilled", "LTX-2.3-Distilled-Diffusers-ckpt"),
|
|
)
|
|
def test_ltx23_distilled_local_dir_resolves_to_distilled(
|
|
self,
|
|
tmp_path,
|
|
monkeypatch,
|
|
directory_name: str,
|
|
) -> None:
|
|
"""Truncated/re-hyphenated local dirs must not be shadowed by the base
|
|
entry, whose detector fires on the shared pipeline class name."""
|
|
import fastvideo.registry # noqa: F401
|
|
from fastvideo.registry import get_preset_selection
|
|
warnings = _record_registry_warnings(monkeypatch)
|
|
model_dir = tmp_path / directory_name
|
|
_write_minimal_model_index(model_dir, "LTX2Pipeline")
|
|
assert get_preset_selection(str(model_dir)) == ("ltx2_distilled", "ltx2")
|
|
assert warnings == []
|
|
|
|
def test_ltx2_equally_specific_local_dir_still_warns(self, tmp_path, monkeypatch) -> None:
|
|
"""Equally specific matches stay ambiguous: warn, keep registration order."""
|
|
import fastvideo.registry # noqa: F401
|
|
from fastvideo.registry import get_preset_selection
|
|
warnings = _record_registry_warnings(monkeypatch)
|
|
model_dir = tmp_path / "ltx2_distilled_diffusers"
|
|
_write_minimal_model_index(model_dir, "LTX2Pipeline")
|
|
assert get_preset_selection(str(model_dir)) == ("ltx2_distilled", "ltx2")
|
|
assert len(warnings) == 1
|
|
assert "Multiple models matched" in warnings[0][0]
|
|
assert str(model_dir) in warnings[0][1]
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Hunyuan preset integration
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
class TestHunyuanPresets:
|
|
|
|
def test_hunyuan_presets_registered(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
presets = get_presets_for_family("hunyuan")
|
|
names = {p.name for p in presets}
|
|
assert names == {"hunyuan_t2v", "fast_hunyuan_t2v"}
|
|
|
|
def test_fast_hunyuan_fewer_steps(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
p = get_preset("fast_hunyuan_t2v", "hunyuan")
|
|
assert p.defaults["num_inference_steps"] == 6
|
|
|
|
def test_removed_gamecraft_checkpoint_is_not_detected_as_hunyuan(self, tmp_path) -> None:
|
|
from fastvideo.registry import get_model_family
|
|
|
|
def make_checkpoint(name: str, class_name: str) -> str:
|
|
model_dir = tmp_path / name
|
|
model_dir.mkdir()
|
|
(model_dir / "model_index.json").write_text(
|
|
json.dumps({
|
|
"_class_name": class_name,
|
|
"_diffusers_version": "0.33.1"
|
|
}))
|
|
return str(model_dir)
|
|
|
|
assert get_model_family(make_checkpoint("HunyuanVideo-Custom", "HunyuanVideoPipeline")) == "hunyuan"
|
|
assert get_model_family(make_checkpoint("Hunyuan-GameCraft-Diffusers", "HunyuanGameCraftPipeline")) is None
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Hunyuan15 preset integration (includes two-stage SR)
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
class TestHunyuan15Presets:
|
|
|
|
def test_hunyuan15_presets_registered(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
presets = get_presets_for_family("hunyuan15")
|
|
assert len(presets) == 5
|
|
|
|
def test_hunyuan15_sr_is_two_stage(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
p = get_preset("hunyuan15_sr_1080p", "hunyuan15")
|
|
assert len(p.stage_schemas) == 2
|
|
assert p.stage_schemas[0].name == "denoise"
|
|
assert p.stage_schemas[1].name == "sr"
|
|
assert p.stage_schemas[1].kind == "super_resolution"
|
|
|
|
def test_hunyuan15_sr_stage_defaults(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
p = get_preset("hunyuan15_sr_1080p", "hunyuan15")
|
|
sr = p.stage_defaults["sr"]
|
|
assert sr["height_sr"] == 1072
|
|
assert sr["width_sr"] == 1920
|
|
|
|
def test_hunyuan15_sr_stage_override_validation(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
p = get_preset("hunyuan15_sr_1080p", "hunyuan15")
|
|
# Valid: override sr num_inference_steps.
|
|
validate_stage_overrides(p, {"sr": {"num_inference_steps": 12}})
|
|
# Invalid: height not in sr allowed_overrides.
|
|
with pytest.raises(ConfigValidationError):
|
|
validate_stage_overrides(p, {"sr": {"height": 1080}})
|
|
|
|
def test_hunyuan15_local_dir_resolves_to_15(self, tmp_path) -> None:
|
|
"""A truncated local dir must not fall back to the generic Hunyuan entry.
|
|
|
|
Both entries share the same amount of text with this name, so the tie
|
|
is broken by registration order, which keeps the dedicated entry.
|
|
"""
|
|
import fastvideo.registry # noqa: F401
|
|
from fastvideo.configs.pipelines.hunyuan15 import Hunyuan15T2V480PConfig
|
|
from fastvideo.registry import get_pipeline_config_cls_from_name, get_preset_selection
|
|
model_dir = tmp_path / "hunyuanvideo15-480p_t2v"
|
|
_write_minimal_model_index(model_dir, "HunyuanVideoPipeline")
|
|
assert get_pipeline_config_cls_from_name(str(model_dir)) is Hunyuan15T2V480PConfig
|
|
assert get_preset_selection(str(model_dir)) == ("hunyuan15_t2v_480p", "hunyuan15")
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Cosmos / Cosmos25 preset integration
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
class TestCosmosPresets:
|
|
|
|
def test_cosmos_preset_registered(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
presets = get_presets_for_family("cosmos")
|
|
assert len(presets) == 1
|
|
assert presets[0].name == "cosmos_predict2_2b"
|
|
|
|
def test_cosmos25_separate_family(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
presets = get_presets_for_family("cosmos25")
|
|
assert {preset.name for preset in presets} == {
|
|
"cosmos25_predict2_2b",
|
|
"cosmos25_dfd_v2w_2b",
|
|
}
|
|
|
|
def test_cosmos25_2b_path_resolves_to_preset(self) -> None:
|
|
"""Regression: the Cosmos-Predict2.5 2B model path must select the
|
|
cosmos25 preset (guidance_scale=7.0), not fall back to the generic
|
|
default (guidance_scale=1.0) which produced blurry output.
|
|
"""
|
|
import fastvideo.registry # noqa: F401
|
|
from fastvideo.api.sampling_param import SamplingParam
|
|
sp = SamplingParam.from_pretrained("KyleShao/Cosmos-Predict2.5-2B-Diffusers")
|
|
assert sp.guidance_scale == 7.0
|
|
assert sp.num_inference_steps == 35
|
|
|
|
def test_cosmos_and_cosmos25_different_fps(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
c = get_preset("cosmos_predict2_2b", "cosmos")
|
|
c25 = get_preset("cosmos25_predict2_2b", "cosmos25")
|
|
assert c.defaults["fps"] == 16
|
|
assert c25.defaults["fps"] == 24
|
|
|
|
def test_cosmos25_dfd_path_resolves_to_four_step_i2v_preset(self, tmp_path) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
from fastvideo.api.sampling_param import SamplingParam
|
|
|
|
model_dir = tmp_path / "Cosmos-Predict2.5-2B-DFD"
|
|
model_dir.mkdir()
|
|
(model_dir / "model_index.json").write_text(
|
|
'{"_class_name": "Cosmos2_5Pipeline", "_diffusers_version": "0.37.0"}'
|
|
)
|
|
sp = SamplingParam.from_pretrained(str(model_dir))
|
|
assert sp.guidance_scale == 1.0
|
|
assert sp.num_inference_steps == 4
|
|
assert sp.num_frames == 81
|
|
assert sp.fps == 24
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# SD35 preset integration
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
class TestSD35Presets:
|
|
|
|
def test_sd35_preset_registered(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
p = get_preset("sd35_medium", "sd35")
|
|
assert p.workload_type == "t2i"
|
|
assert p.defaults["height"] == 512
|
|
assert p.defaults["num_frames"] == 1
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# LingBot-Video preset integration (Dense and MoE/refiner)
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
class TestLingBotVideoPresets:
|
|
|
|
@pytest.mark.parametrize(
|
|
("directory_name", "pipeline_class", "expected_preset"),
|
|
(
|
|
("arbitrary-layout-a", "LingBotVideoDensePipeline", "lingbot_video_dense_t2v"),
|
|
("arbitrary-layout-b", "LingBotVideoMoePipeline", "lingbot_video_moe_refiner_t2v"),
|
|
),
|
|
)
|
|
def test_local_layouts_select_one_variant(
|
|
self,
|
|
tmp_path,
|
|
caplog,
|
|
directory_name: str,
|
|
pipeline_class: str,
|
|
expected_preset: str,
|
|
) -> None:
|
|
"""Resolve canonical local Dense and MoE paths without detector overlap."""
|
|
from fastvideo.registry import get_preset_selection
|
|
model_dir = tmp_path / directory_name
|
|
model_dir.mkdir()
|
|
(model_dir / "transformer").mkdir()
|
|
(model_dir / "model_index.json").write_text(
|
|
json.dumps({
|
|
"_class_name": pipeline_class,
|
|
"_diffusers_version": "0.39.0"
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
assert get_preset_selection(str(model_dir)) == (expected_preset, "lingbot_video")
|
|
assert "Multiple models matched" not in caplog.text
|
|
|
|
def test_lingbot_video_presets_registered(self) -> None:
|
|
"""Register both released LingBot-Video inference layouts."""
|
|
import fastvideo.registry # noqa: F401
|
|
from fastvideo.registry import get_preset_selection
|
|
|
|
presets = get_presets_for_family("lingbot_video")
|
|
assert {preset.name
|
|
for preset in presets} == {
|
|
"lingbot_video_dense_t2v",
|
|
"lingbot_video_moe_refiner_t2v",
|
|
}
|
|
assert get_preset_selection("FastVideo/LingBot-Video-Dense-1.3B-Diffusers") == (
|
|
"lingbot_video_dense_t2v",
|
|
"lingbot_video",
|
|
)
|
|
assert get_preset_selection("FastVideo/LingBot-Video-MoE-30B-A3B-Diffusers") == (
|
|
"lingbot_video_moe_refiner_t2v",
|
|
"lingbot_video",
|
|
)
|
|
|
|
def test_moe_refiner_defaults(self) -> None:
|
|
"""Expose the official base and 1080p refiner sampling defaults."""
|
|
import fastvideo.registry # noqa: F401
|
|
preset = get_preset("lingbot_video_moe_refiner_t2v", "lingbot_video")
|
|
expected = {
|
|
"height": 480,
|
|
"width": 832,
|
|
"height_sr": 1088,
|
|
"width_sr": 1920,
|
|
"num_frames": 121,
|
|
"num_inference_steps": 40,
|
|
"num_inference_steps_sr": 8,
|
|
"guidance_scale": 3.0,
|
|
"guidance_scale_2": 3.0,
|
|
"batch_cfg": True,
|
|
"t_thresh": 0.85,
|
|
"seed": 42,
|
|
}
|
|
assert {key: preset.defaults[key] for key in expected} == expected
|
|
assert preset.stage_defaults["refine"] == {
|
|
key: expected[key]
|
|
for key in ("height_sr", "width_sr", "num_inference_steps_sr", "guidance_scale_2", "t_thresh")
|
|
}
|
|
|
|
def test_moe_refiner_stage_schema_uses_sampling_fields(self) -> None:
|
|
"""Accept only declared ``SamplingParam`` fields as stage overrides."""
|
|
import fastvideo.registry # noqa: F401
|
|
from fastvideo.api.sampling_param import SamplingParam
|
|
preset = get_preset("lingbot_video_moe_refiner_t2v", "lingbot_video")
|
|
sampling_fields = set(SamplingParam.__dataclass_fields__)
|
|
assert [(stage.name, stage.kind) for stage in preset.stage_schemas] == [
|
|
("denoise", "denoising"),
|
|
("refine", "refinement"),
|
|
]
|
|
assert set(preset.defaults) <= sampling_fields
|
|
assert all(stage.allowed_overrides <= sampling_fields for stage in preset.stage_schemas)
|
|
validate_stage_overrides(
|
|
preset,
|
|
{
|
|
"denoise": {
|
|
"num_inference_steps": 32,
|
|
"guidance_scale": 2.5
|
|
},
|
|
"refine": {
|
|
"height_sr": 1088,
|
|
"t_thresh": 0.8
|
|
},
|
|
},
|
|
)
|
|
with pytest.raises(ConfigValidationError):
|
|
validate_stage_overrides(preset, {"refine": {"refiner_steps": 8}})
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# LingBot-World-Fast preset integration
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
class TestLingBotWorldFastPresets:
|
|
|
|
def test_registered_defaults_are_fixed(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
from fastvideo.registry import get_preset_selection
|
|
|
|
model_path = "FastVideo/LingBot-World-Fast-Diffusers"
|
|
assert get_preset_selection(model_path) == ("lingbotworld_fast_i2v", "lingbotworld_fast")
|
|
|
|
preset = get_preset("lingbotworld_fast_i2v", "lingbotworld_fast")
|
|
assert preset.defaults["num_inference_steps"] == 4
|
|
assert preset.defaults["guidance_scale"] == 1.0
|
|
assert preset.stage_schemas[0].allowed_overrides == frozenset()
|
|
with pytest.raises(ConfigValidationError, match="does not accept overrides"):
|
|
validate_stage_overrides(preset, {"denoise": {"num_inference_steps": 8}})
|
|
|
|
def test_local_checkpoint_resolves_fast_pipeline(self, tmp_path) -> None:
|
|
from fastvideo.configs.pipelines.lingbotworld_fast import LingBotWorldFastI2V480PConfig
|
|
from fastvideo.fastvideo_args import WorkloadType
|
|
from fastvideo.pipelines.basic.lingbotworld_fast import LingBotWorldFastPipeline
|
|
from fastvideo.registry import get_model_info
|
|
|
|
model_dir = tmp_path / "lingbot-world-fast"
|
|
(model_dir / "transformer").mkdir(parents=True)
|
|
(model_dir / "model_index.json").write_text(
|
|
json.dumps({
|
|
"_class_name": "LingBotWorldCausalDMDPipeline",
|
|
"_diffusers_version": "0.36.0",
|
|
}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
info = get_model_info(str(model_dir), workload_type=WorkloadType.I2V)
|
|
assert info.pipeline_cls is LingBotWorldFastPipeline
|
|
assert info.pipeline_config_cls is LingBotWorldFastI2V480PConfig
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# LingBotWorld preset integration (dual guidance)
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
class TestLingBotWorldPresets:
|
|
|
|
def test_lingbotworld_dual_guidance(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
p = get_preset("lingbotworld_i2v", "lingbotworld")
|
|
stage = p.stage_schemas[0]
|
|
assert "guidance_scale_2" in stage.allowed_overrides
|
|
assert "boundary_ratio" in stage.allowed_overrides
|
|
|
|
def test_lingbotworld_override_validation(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
p = get_preset("lingbotworld_i2v", "lingbotworld")
|
|
validate_stage_overrides(p, {"denoise": {"boundary_ratio": 0.95}})
|
|
with pytest.raises(ConfigValidationError):
|
|
validate_stage_overrides(p, {"denoise": {"height": 720}})
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Remaining single-preset families
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
class TestSinglePresetFamilies:
|
|
|
|
def test_hyworld_registered(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
p = get_preset("hyworld_t2v", "hyworld")
|
|
assert p.workload_type == "t2v"
|
|
|
|
def test_gen3c_registered(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
p = get_preset("gen3c_cosmos_7b", "gen3c")
|
|
assert p.defaults["num_inference_steps"] == 35
|
|
|
|
def test_matrixgame_registered(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
p2 = get_preset("matrixgame2_i2v", "matrixgame")
|
|
assert p2.defaults["num_inference_steps"] == 3
|
|
assert p2.defaults["fps"] == 25
|
|
p3 = get_preset("matrixgame3_i2v", "matrixgame")
|
|
assert p3.defaults["num_inference_steps"] == 3
|
|
assert p3.defaults["height"] == 720
|
|
|
|
|
|
# -------------------------------------------------------------------
|
|
# Cross-family: total preset count
|
|
# -------------------------------------------------------------------
|
|
|
|
|
|
class TestPresetCountIntegrity:
|
|
|
|
def test_total_preset_count(self) -> None:
|
|
"""At least the baseline 37 presets from 13 families are registered."""
|
|
import fastvideo.registry # noqa: F401
|
|
names = get_all_preset_names()
|
|
assert len(names) >= 37
|
|
|
|
|
|
class TestPresetDefaultTypes:
|
|
"""Preset ``defaults`` values must match the types on
|
|
:class:`SamplingParam`. Assigning ``None`` to a typed-``str`` field
|
|
(e.g. ``negative_prompt``) breaks downstream stages that assert the
|
|
runtime type — see the CFG branch in
|
|
``pipelines/stages/text_encoding.py:81``."""
|
|
|
|
def test_ltx2_cfg_defaults_are_off(self) -> None:
|
|
"""SamplingParam's LTX-2 CFG class defaults must be 1.0 (CFG
|
|
off). ``ForwardBatch.__post_init__`` force-enables
|
|
``do_classifier_free_guidance`` when either
|
|
``ltx2_cfg_scale_video`` or ``ltx2_cfg_scale_audio`` is != 1.0,
|
|
so any non-1.0 default silently forces CFG on for every model
|
|
family that doesn't explicitly override these fields. Guard
|
|
against the regression that surfaced as a Wan-based I2V SSIM
|
|
crash (``text_encoding.py:81`` assertion on
|
|
``negative_prompt``)."""
|
|
from fastvideo.api.sampling_param import SamplingParam
|
|
sp = SamplingParam()
|
|
assert sp.ltx2_cfg_scale_video == 1.0
|
|
assert sp.ltx2_cfg_scale_audio == 1.0
|
|
|
|
def test_no_preset_sets_negative_prompt_to_none(self) -> None:
|
|
import fastvideo.registry # noqa: F401
|
|
from fastvideo.api.presets import _PRESET_REGISTRY
|
|
offenders = [
|
|
f"{preset.model_family}/{preset.name}" for preset in _PRESET_REGISTRY.values()
|
|
if preset.defaults.get("negative_prompt", "") is None
|
|
]
|
|
assert not offenders, ("These presets set negative_prompt=None, which violates "
|
|
"SamplingParam.negative_prompt's typed str contract and "
|
|
"crashes the CFG path in text_encoding. Use \"\" instead:\n" +
|
|
"\n".join(f" - {p}" for p in offenders))
|