teknium1
d8053f4806
fal's LTX 2.5 fast endpoints accept 6-20s only up to 1080p — "At 1440p and 2160p, all frame rates support up to 10 seconds" — so a 4K request with the family's 20s ceiling was rejected by the vendor. Families can now declare `duration_cap_by_resolution`, applied after the enum snap / range clamp on the resolved resolution enum. An unset duration on a duration_enum family also snapped to enum[0] (6s), silently overriding the endpoint's own "auto" default; None now omits the key for enum families exactly as it already did for range families. test_managed_media_gateways asserts the alibaba/happy-horse/ namespace by prefix rather than the exact v1.1 literal so the next version bump doesn't flip an unrelated gateway test.
842 lines
33 KiB
Python
842 lines
33 KiB
Python
"""Tests for the FAL video gen plugin — family routing, payload shape."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from agent import video_gen_registry
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_registry():
|
|
video_gen_registry._reset_for_tests()
|
|
yield
|
|
video_gen_registry._reset_for_tests()
|
|
|
|
|
|
def test_fal_provider_registers():
|
|
from plugins.video_gen.fal import FALVideoGenProvider, DEFAULT_MODEL
|
|
|
|
provider = FALVideoGenProvider()
|
|
video_gen_registry.register_provider(provider)
|
|
|
|
assert video_gen_registry.get_provider("fal") is provider
|
|
assert provider.display_name == "FAL"
|
|
# DEFAULT_MODEL is the cheap-tier default
|
|
assert provider.default_model() == DEFAULT_MODEL
|
|
assert DEFAULT_MODEL in {"pixverse-v6", "ltx-2.3"}
|
|
|
|
|
|
def test_kling_v3_standard_and_pro_payload_shape():
|
|
"""Kling 3.0 (v3 standard/pro): start_image_url on i2v, aspect_ratio
|
|
dropped on i2v (schema derives it from the image), no seed/resolution
|
|
keys, string duration 3-15, generate_audio + negative_prompt real."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
for fid in ("kling-v3", "kling-v3-pro"):
|
|
meta = FAL_FAMILIES[fid]
|
|
assert meta.get("image_param_key") == "start_image_url"
|
|
|
|
# text-to-video route
|
|
p = _build_payload(
|
|
meta,
|
|
prompt="a mecha lands",
|
|
image_url=None,
|
|
duration=7,
|
|
aspect_ratio="16:9",
|
|
resolution="1080p",
|
|
negative_prompt="blurry",
|
|
audio=True,
|
|
seed=3,
|
|
)
|
|
assert p == {
|
|
"prompt": "a mecha lands",
|
|
"aspect_ratio": "16:9",
|
|
"duration": "7",
|
|
"generate_audio": True,
|
|
"negative_prompt": "blurry",
|
|
}, fid
|
|
|
|
# image-to-video route: start_image_url in, aspect_ratio dropped
|
|
p = _build_payload(
|
|
meta,
|
|
prompt="animate it",
|
|
image_url="https://example.com/i.png",
|
|
duration=20, # clamps to 15
|
|
aspect_ratio="16:9",
|
|
resolution="720p",
|
|
negative_prompt=None,
|
|
audio=False,
|
|
seed=None,
|
|
)
|
|
assert p.get("start_image_url") == "https://example.com/i.png", fid
|
|
assert "image_url" not in p, fid
|
|
assert "aspect_ratio" not in p, fid
|
|
assert p["duration"] == "15", fid
|
|
assert p["generate_audio"] is False, fid
|
|
|
|
|
|
def test_kling_4k_uses_start_image_url():
|
|
"""Kling v3 4K's image-to-video endpoint expects start_image_url,
|
|
not image_url. The family must declare image_param_key='start_image_url'."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
meta = FAL_FAMILIES["kling-v3-4k"]
|
|
assert meta.get("image_param_key") == "start_image_url"
|
|
payload = _build_payload(
|
|
meta,
|
|
prompt="x",
|
|
image_url="https://example.com/i.png",
|
|
duration=5,
|
|
aspect_ratio="16:9",
|
|
resolution="720p",
|
|
negative_prompt=None,
|
|
audio=None,
|
|
seed=None,
|
|
)
|
|
assert payload.get("start_image_url") == "https://example.com/i.png"
|
|
assert "image_url" not in payload
|
|
|
|
|
|
def test_minimax_h3_int_duration_and_resolution_alias():
|
|
"""MiniMax H3 requires duration as a JSON integer and uses the
|
|
768P/2K/4K resolution enum — the tool's 720p/1080p values must map."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
meta = FAL_FAMILIES["minimax-h3"]
|
|
payload = _build_payload(
|
|
meta,
|
|
prompt="x",
|
|
image_url=None,
|
|
duration=7,
|
|
aspect_ratio="16:9",
|
|
resolution="720p",
|
|
negative_prompt=None,
|
|
audio=True,
|
|
seed=None,
|
|
)
|
|
assert payload["duration"] == 7 and isinstance(payload["duration"], int)
|
|
assert payload["resolution"] == "768P"
|
|
assert payload["aspect_ratio"] == "16:9"
|
|
# H3 has no generate_audio key (audio is native/always-on)
|
|
assert "generate_audio" not in payload
|
|
|
|
hi = _build_payload(
|
|
meta, prompt="x", image_url=None, duration=5, aspect_ratio="16:9",
|
|
resolution="1080p", negative_prompt=None, audio=None, seed=None,
|
|
)
|
|
assert hi["resolution"] == "2K"
|
|
|
|
|
|
def test_h3_max_turbo_static_key_and_1080p_alias():
|
|
"""H3 Max Turbo requires prompt_expansion_mode on both endpoints, adds a real
|
|
1080P tier (unlike Max, which caps at 768P), and its i2v drops aspect_ratio."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
meta = FAL_FAMILIES["minimax-h3-max-turbo"]
|
|
t2v = _build_payload(
|
|
meta, prompt="x", image_url=None, duration=7, aspect_ratio="16:9",
|
|
resolution="1080p", negative_prompt=None, audio=None, seed=11,
|
|
)
|
|
assert t2v["prompt_expansion_mode"] == "balanced"
|
|
assert t2v["resolution"] == "1080P"
|
|
assert t2v["duration"] == 7 and isinstance(t2v["duration"], int)
|
|
assert t2v["seed"] == 11
|
|
|
|
i2v = _build_payload(
|
|
meta, prompt="x", image_url="https://example.com/i.png", duration=5,
|
|
aspect_ratio="16:9", resolution="480p", negative_prompt=None, audio=None, seed=None,
|
|
)
|
|
assert i2v["prompt_expansion_mode"] == "balanced"
|
|
assert "aspect_ratio" not in i2v
|
|
assert i2v["image_url"] == "https://example.com/i.png"
|
|
|
|
|
|
def test_image_drop_keys_strips_aspect_ratio_on_i2v():
|
|
"""Seedance 2.5 / MiniMax H3 / Grok 1.5 i2v endpoints derive the
|
|
aspect ratio from the input image; sending the key is rejected."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
for fid in ("seedance-2.5", "minimax-h3", "grok-imagine-1.5"):
|
|
meta = FAL_FAMILIES[fid]
|
|
i2v = _build_payload(
|
|
meta, prompt="x", image_url="https://example.com/i.png",
|
|
duration=5, aspect_ratio="16:9", resolution="480p",
|
|
negative_prompt=None, audio=None, seed=None,
|
|
)
|
|
assert "aspect_ratio" not in i2v, fid
|
|
# ...but text-to-video keeps it
|
|
t2v = _build_payload(
|
|
meta, prompt="x", image_url=None, duration=5,
|
|
aspect_ratio="16:9", resolution="480p",
|
|
negative_prompt=None, audio=None, seed=None,
|
|
)
|
|
assert t2v.get("aspect_ratio") == "16:9", fid
|
|
|
|
|
|
def test_seedance_25_string_duration_up_to_30():
|
|
"""Seedance 2.5 keeps the stringified duration convention and supports
|
|
the full 4-30s range."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
meta = FAL_FAMILIES["seedance-2.5"]
|
|
payload = _build_payload(
|
|
meta, prompt="x", image_url=None, duration=30, aspect_ratio="1:1",
|
|
resolution="480p", negative_prompt=None, audio=True, seed=None,
|
|
)
|
|
assert payload["duration"] == "30"
|
|
assert payload["generate_audio"] is True
|
|
|
|
|
|
def test_wan_30_audio_toggle_uses_family_key_and_start_image_url():
|
|
"""Wan 3.0's schema names the audio toggle `audio` (not `generate_audio`), takes
|
|
`start_image_url` on i2v and an integer duration; veo3.1 keeps `generate_audio`."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
kw = dict(prompt="x", duration=7, aspect_ratio="16:9", resolution="720p", negative_prompt=None, audio=True, seed=None)
|
|
p = _build_payload(FAL_FAMILIES["wan-3.0"], image_url="https://i.png", **kw)
|
|
assert p["audio"] is True and "generate_audio" not in p
|
|
assert p["start_image_url"] == "https://i.png" and p["duration"] == 7
|
|
assert _build_payload(FAL_FAMILIES["veo3.1"], image_url=None, **kw)["generate_audio"] is True
|
|
|
|
|
|
def test_gemini_omni_flash_v11_is_dual_modality():
|
|
"""v1.1 (Aug 2026) added a text-to-video endpoint; both modalities
|
|
must route to the versioned v1.1 endpoints."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES
|
|
|
|
meta = FAL_FAMILIES["gemini-omni-flash"]
|
|
assert meta["text_endpoint"] == "google/gemini-omni-flash/v1.1/text-to-video"
|
|
assert meta["image_endpoint"] == "google/gemini-omni-flash/v1.1/image-to-video"
|
|
|
|
|
|
def test_text_only_job_errors_cleanly_for_i2v_only_family(monkeypatch):
|
|
"""Catalog-shape guard: a family without a text endpoint must error cleanly
|
|
instead of submitting to a None endpoint. Every cataloged family is now
|
|
dual-modality, so the guard is exercised with a synthetic family."""
|
|
from plugins.video_gen import fal as fal_plugin
|
|
from plugins.video_gen.fal import FALVideoGenProvider, _family
|
|
|
|
synthetic = _family("Synthetic i2v", "~1s", "cheap", "test", None, "example/i2v-only/image-to-video", durations=(3, 10), duration_int=True)
|
|
monkeypatch.setattr(fal_plugin, "_fal_video_available", lambda: True)
|
|
monkeypatch.setattr(fal_plugin, "_load_fal_client", lambda: object())
|
|
monkeypatch.setattr(fal_plugin, "_resolve_family", lambda explicit: ("synthetic", synthetic))
|
|
monkeypatch.setattr(fal_plugin, "_submit_fal_video_request", lambda *a, **k: pytest.fail("submitted to a None endpoint"))
|
|
|
|
result = FALVideoGenProvider().generate("a dog running")
|
|
assert result["success"] is False
|
|
assert result["error_type"] == "modality_unsupported"
|
|
|
|
|
|
def test_every_family_has_required_metadata():
|
|
"""Invariant: every family entry carries the picker-facing metadata and
|
|
at least one endpoint."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES
|
|
|
|
for fid, meta in FAL_FAMILIES.items():
|
|
assert meta.get("display"), fid
|
|
assert meta.get("tier") in {"cheap", "premium"}, fid
|
|
assert meta.get("text_endpoint") or meta.get("image_endpoint"), fid
|
|
|
|
|
|
class TestFamilyRouting:
|
|
"""The headline behavior: image_url presence picks the endpoint."""
|
|
|
|
@pytest.fixture
|
|
def with_fake_fal(self, monkeypatch):
|
|
"""Stub fal_client.submit to capture which endpoint we hit."""
|
|
import sys
|
|
import types
|
|
|
|
captured = {"endpoint": None, "arguments": None}
|
|
|
|
class FakeHandle:
|
|
def get(self):
|
|
return {"video": {"url": "https://fake/out.mp4"}}
|
|
|
|
fake = types.ModuleType("fal_client")
|
|
def _submit(endpoint, arguments=None, headers=None):
|
|
captured["endpoint"] = endpoint
|
|
captured["arguments"] = arguments
|
|
return FakeHandle()
|
|
fake.submit = _submit # type: ignore
|
|
monkeypatch.setitem(sys.modules, "fal_client", fake)
|
|
|
|
# Reset the lazy global so it picks up our stub
|
|
from plugins.video_gen import fal as fal_plugin
|
|
fal_plugin._fal_client = None
|
|
# Also reset the managed client cache
|
|
fal_plugin._managed_fal_video_client = None
|
|
fal_plugin._managed_fal_video_client_config = None
|
|
|
|
monkeypatch.setenv("FAL_KEY", "test")
|
|
# Force direct mode — no managed gateway
|
|
monkeypatch.setattr(fal_plugin, "_resolve_managed_fal_video_gateway", lambda: None)
|
|
return captured
|
|
|
|
def test_text_to_video_routes_to_text_endpoint(self, with_fake_fal):
|
|
from plugins.video_gen.fal import FALVideoGenProvider
|
|
|
|
result = FALVideoGenProvider().generate(
|
|
"a dog running",
|
|
model="pixverse-v6",
|
|
)
|
|
assert result["success"] is True
|
|
assert with_fake_fal["endpoint"] == "fal-ai/pixverse/v6/text-to-video"
|
|
assert result["modality"] == "text"
|
|
assert with_fake_fal["arguments"]["prompt"] == "a dog running"
|
|
assert "image_url" not in with_fake_fal["arguments"]
|
|
|
|
def test_image_to_video_routes_to_image_endpoint(self, with_fake_fal):
|
|
from plugins.video_gen.fal import FALVideoGenProvider
|
|
|
|
result = FALVideoGenProvider().generate(
|
|
"animate this dog",
|
|
model="pixverse-v6",
|
|
image_url="https://example.com/dog.png",
|
|
)
|
|
assert result["success"] is True
|
|
assert with_fake_fal["endpoint"] == "fal-ai/pixverse/v6/image-to-video"
|
|
assert result["modality"] == "image"
|
|
assert with_fake_fal["arguments"]["image_url"] == "https://example.com/dog.png"
|
|
|
|
def test_default_family_text_routing(self, with_fake_fal):
|
|
"""No model arg → DEFAULT_MODEL → text-to-video endpoint."""
|
|
from plugins.video_gen.fal import FALVideoGenProvider, FAL_FAMILIES, DEFAULT_MODEL
|
|
|
|
result = FALVideoGenProvider().generate("a dog")
|
|
assert result["success"] is True
|
|
expected_endpoint = FAL_FAMILIES[DEFAULT_MODEL]["text_endpoint"]
|
|
assert with_fake_fal["endpoint"] == expected_endpoint
|
|
|
|
|
|
def test_unknown_family_falls_back_to_default(self, with_fake_fal):
|
|
from plugins.video_gen.fal import FALVideoGenProvider, FAL_FAMILIES, DEFAULT_MODEL
|
|
|
|
result = FALVideoGenProvider().generate(
|
|
"x",
|
|
model="not-a-real-family",
|
|
)
|
|
assert result["success"] is True
|
|
expected_endpoint = FAL_FAMILIES[DEFAULT_MODEL]["text_endpoint"]
|
|
assert with_fake_fal["endpoint"] == expected_endpoint
|
|
|
|
def test_premium_seedance_routing(self, with_fake_fal):
|
|
"""Sanity check the premium-tier seedance routes correctly."""
|
|
from plugins.video_gen.fal import FALVideoGenProvider
|
|
|
|
result = FALVideoGenProvider().generate(
|
|
"a dog",
|
|
model="seedance-2.0",
|
|
image_url="https://example.com/dog.png",
|
|
)
|
|
assert result["success"] is True
|
|
assert with_fake_fal["endpoint"] == "bytedance/seedance-2.0/image-to-video"
|
|
# Seedance uses regular image_url (not start_image_url)
|
|
assert with_fake_fal["arguments"]["image_url"] == "https://example.com/dog.png"
|
|
|
|
|
|
class TestFamilyKeyNormalization:
|
|
def test_full_endpoint_paths_resolve_to_their_own_family(self):
|
|
"""A configured endpoint path must resolve to the family that declares
|
|
it. The segment scan alone reads the "seedance-2.0" in
|
|
".../seedance-2.0/mini/..." and bills the full-price family."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _normalize_family_key
|
|
|
|
for fid, meta in FAL_FAMILIES.items():
|
|
for key in ("text_endpoint", "image_endpoint"):
|
|
endpoint = meta.get(key)
|
|
if endpoint:
|
|
assert _normalize_family_key(endpoint) == fid, endpoint
|
|
|
|
def test_bare_and_prefixed_ids_still_resolve(self):
|
|
from plugins.video_gen.fal import _normalize_family_key
|
|
|
|
assert _normalize_family_key("seedance-2.5") == "seedance-2.5"
|
|
assert _normalize_family_key("bytedance/seedance-2.5") == "seedance-2.5"
|
|
assert _normalize_family_key(" pixverse-v6 ") == "pixverse-v6"
|
|
assert _normalize_family_key("nonsense/thing") is None
|
|
|
|
def test_truncated_endpoint_stems_resolve(self):
|
|
"""Config often stores the FAL app path without the modality leaf."""
|
|
from plugins.video_gen.fal import _normalize_family_key
|
|
|
|
assert _normalize_family_key("bytedance/seedance-2.0/mini") == "seedance-2.0-mini"
|
|
assert _normalize_family_key("bytedance/seedance-2.0") == "seedance-2.0"
|
|
assert _normalize_family_key("minimax/h3") == "minimax-h3"
|
|
assert _normalize_family_key("xai/grok-imagine-video/v1.5") == "grok-imagine-1.5"
|
|
assert _normalize_family_key("google/gemini-omni-flash") == "gemini-omni-flash"
|
|
assert _normalize_family_key("blackforestlabs/flux-3") == "flux-3"
|
|
|
|
def test_capabilities_span_longest_family_duration(self):
|
|
"""capabilities() is active-MODEL-aware (#95681 diet): it reports
|
|
the resolved family's real window, so the schema doesn't overstate
|
|
short families or understate Seedance 2.5. The union fallback
|
|
(resolution failure) must still span the 30s ceiling."""
|
|
from unittest.mock import patch as _patch
|
|
|
|
import plugins.video_gen.fal as _fp
|
|
from plugins.video_gen.fal import FAL_FAMILIES, FALVideoGenProvider
|
|
|
|
# Active model resolved → that family's actual window.
|
|
meta = FAL_FAMILIES["seedance-2.5"]
|
|
with _patch.object(_fp, "_resolve_family",
|
|
return_value=("seedance-2.5", meta)):
|
|
caps = FALVideoGenProvider().capabilities()
|
|
assert caps["max_duration"] >= 30
|
|
# A short family must NOT be inflated to the union ceiling.
|
|
short = FAL_FAMILIES["pixverse-v6"]
|
|
durs = short.get("durations")
|
|
hi = durs[1] if isinstance(durs, tuple) else max(durs)
|
|
with _patch.object(_fp, "_resolve_family",
|
|
return_value=("pixverse-v6", short)):
|
|
caps = FALVideoGenProvider().capabilities()
|
|
assert caps["max_duration"] == hi
|
|
|
|
# Resolution failure → union fallback still spans the ceiling.
|
|
with _patch.object(_fp, "_resolve_family",
|
|
side_effect=RuntimeError("no config")):
|
|
caps = FALVideoGenProvider().capabilities()
|
|
assert caps["max_duration"] >= 30
|
|
assert caps["min_duration"] <= 1
|
|
|
|
|
|
class TestPayloadBuilder:
|
|
def test_drops_unsupported_keys(self):
|
|
"""Veo enum-clamps duration, supports aspect+resolution+audio+neg."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
meta = FAL_FAMILIES["veo3.1"]
|
|
p = _build_payload(
|
|
meta,
|
|
prompt="x",
|
|
image_url=None,
|
|
duration=12, # not in enum (4,6,8) — snap to 8
|
|
aspect_ratio="16:9",
|
|
resolution="720p",
|
|
negative_prompt="ugly",
|
|
audio=True,
|
|
seed=42,
|
|
)
|
|
assert p["prompt"] == "x"
|
|
assert p["duration"] == "8s" # veo3.1 uses "Ns" format per FAL API
|
|
assert p["aspect_ratio"] == "16:9"
|
|
assert p["resolution"] == "720p"
|
|
assert p["generate_audio"] is True
|
|
assert p["negative_prompt"] == "ugly"
|
|
assert p["seed"] == 42
|
|
|
|
def test_pixverse_range_clamps_correctly(self):
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
meta = FAL_FAMILIES["pixverse-v6"]
|
|
p = _build_payload(
|
|
meta,
|
|
prompt="x",
|
|
image_url="https://i.png",
|
|
duration=99, # over max → 15
|
|
aspect_ratio="16:9",
|
|
resolution="540p",
|
|
negative_prompt=None,
|
|
audio=None,
|
|
seed=None,
|
|
)
|
|
assert p["duration"] == "15"
|
|
|
|
@pytest.mark.parametrize(
|
|
"family_id",
|
|
[
|
|
"seedance-2.0",
|
|
"seedance-2.0-mini",
|
|
"seedance-2.5",
|
|
"minimax-h3",
|
|
"flux-3",
|
|
"grok-imagine-1.5",
|
|
"gemini-omni-flash",
|
|
],
|
|
)
|
|
def test_seed_dropped_for_families_without_seed_support(self, family_id):
|
|
"""These FAL endpoints declare no `seed`; the gateway forwards whatever
|
|
we send, so an unknown key would reach the vendor."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
p = _build_payload(
|
|
FAL_FAMILIES[family_id],
|
|
prompt="x",
|
|
image_url="https://i.png",
|
|
duration=None,
|
|
aspect_ratio="16:9",
|
|
resolution="720p",
|
|
negative_prompt=None,
|
|
audio=None,
|
|
seed=42,
|
|
)
|
|
assert "seed" not in p
|
|
|
|
def test_minimax_h3_uses_uppercase_resolution_enum(self):
|
|
"""FAL spells MiniMax H3 resolutions "768P"/"2K"/"4K"; tool-style
|
|
values like "720p" are aliased via resolution_aliases."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
meta = FAL_FAMILIES["minimax-h3"]
|
|
accepted = _build_payload(
|
|
meta, prompt="x", image_url=None, duration=7, aspect_ratio="16:9",
|
|
resolution="2K", negative_prompt=None, audio=None, seed=None,
|
|
)
|
|
assert accepted["resolution"] == "2K"
|
|
assert accepted["duration"] == 7
|
|
|
|
aliased = _build_payload(
|
|
meta, prompt="x", image_url=None, duration=7, aspect_ratio="16:9",
|
|
resolution="720p", negative_prompt=None, audio=None, seed=None,
|
|
)
|
|
assert aliased["resolution"] == "768P"
|
|
|
|
def test_audio_only_sent_for_families_that_declare_it(self):
|
|
"""minimax-h3 and the i2v-only families have no generate_audio field."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
for family_id in ("minimax-h3", "grok-imagine-1.5", "gemini-omni-flash"):
|
|
p = _build_payload(
|
|
FAL_FAMILIES[family_id],
|
|
prompt="x", image_url="https://i.png", duration=None,
|
|
aspect_ratio="16:9", resolution="720p", negative_prompt="ugly",
|
|
audio=True, seed=None,
|
|
)
|
|
assert "generate_audio" not in p, family_id
|
|
assert "negative_prompt" not in p, family_id
|
|
|
|
@pytest.mark.parametrize(
|
|
"family_id,expected",
|
|
[
|
|
("minimax-h3", 7), # FAL types duration as an integer
|
|
("flux-3", 7), # mixed ["auto", 5, 6, ...] literal enum
|
|
("grok-imagine-1.5", 7),
|
|
("gemini-omni-flash", 7),
|
|
("seedance-2.5", "7"), # FAL enum is strings: "auto","4",...
|
|
("seedance-2.0-mini", "7"),
|
|
("pixverse-v6", "7"), # unchanged legacy string form
|
|
("veo3.1", "6s"), # unchanged suffix form (7 snaps to 6)
|
|
],
|
|
)
|
|
def test_duration_is_emitted_in_the_form_fal_declares(self, family_id, expected):
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
p = _build_payload(
|
|
FAL_FAMILIES[family_id],
|
|
prompt="x", image_url=None, duration=7, aspect_ratio="16:9",
|
|
resolution="720p", negative_prompt=None, audio=None, seed=None,
|
|
)
|
|
assert p["duration"] == expected
|
|
assert type(p["duration"]) is type(expected)
|
|
|
|
def test_every_family_declares_both_endpoints(self):
|
|
"""Catalog invariant: since Gemini Omni Flash 1.1 every family is
|
|
dual-modality — both endpoints must be non-empty strings."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES
|
|
|
|
for fid, meta in FAL_FAMILIES.items():
|
|
assert meta.get("text_endpoint"), fid
|
|
assert meta.get("image_endpoint"), fid
|
|
|
|
def test_ltx_omits_duration_aspect_resolution(self):
|
|
"""LTX 2.3 doesn't declare duration/aspect/resolution enums —
|
|
the payload should NOT include those keys (let FAL default)."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
meta = FAL_FAMILIES["ltx-2.3"]
|
|
p = _build_payload(
|
|
meta,
|
|
prompt="x",
|
|
image_url=None,
|
|
duration=8,
|
|
aspect_ratio="16:9",
|
|
resolution="720p",
|
|
negative_prompt="ugly",
|
|
audio=True,
|
|
seed=None,
|
|
)
|
|
assert "duration" not in p
|
|
assert "aspect_ratio" not in p
|
|
assert "resolution" not in p
|
|
# But audio + negative are advertised
|
|
assert p["generate_audio"] is True
|
|
assert p["negative_prompt"] == "ugly"
|
|
|
|
def test_range_families_omit_duration_when_unspecified(self):
|
|
"""Range-based families must omit `duration` when the caller doesn't
|
|
specify one so FAL applies its endpoint default, not the minimum."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
for family_id in ("pixverse-v6", "seedance-2.0", "kling-v3-4k"):
|
|
meta = FAL_FAMILIES[family_id]
|
|
p = _build_payload(
|
|
meta,
|
|
prompt="x",
|
|
image_url=None,
|
|
duration=None,
|
|
aspect_ratio="16:9",
|
|
resolution="720p",
|
|
negative_prompt=None,
|
|
audio=None,
|
|
seed=None,
|
|
)
|
|
assert "duration" not in p, (
|
|
f"{family_id}: duration=None should omit the field, "
|
|
f"got {p.get('duration')!r}"
|
|
)
|
|
|
|
def test_happy_horse_minimal_payload(self):
|
|
"""Happy Horse 1.1 has a full published schema — verified fields only."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
meta = FAL_FAMILIES["happy-horse"]
|
|
p = _build_payload(
|
|
meta,
|
|
prompt="a horse galloping",
|
|
image_url=None,
|
|
duration=8,
|
|
aspect_ratio="16:9",
|
|
resolution="720p",
|
|
negative_prompt="watermark",
|
|
audio=True,
|
|
seed=None,
|
|
)
|
|
# v1.1 declares aspect_ratio/resolution/duration (integer). Audio is
|
|
# native (no generate_audio key) and negative_prompt is unsupported.
|
|
assert p == {
|
|
"prompt": "a horse galloping",
|
|
"aspect_ratio": "16:9",
|
|
"resolution": "720p",
|
|
"duration": 8,
|
|
}
|
|
assert isinstance(p["duration"], int)
|
|
|
|
def test_happy_horse_i2v_drops_aspect_ratio(self):
|
|
"""Happy Horse 1.1 i2v derives aspect ratio from the input image."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
meta = FAL_FAMILIES["happy-horse"]
|
|
p = _build_payload(
|
|
meta,
|
|
prompt="animate this",
|
|
image_url="https://example.com/frame.png",
|
|
duration=5,
|
|
aspect_ratio="16:9",
|
|
resolution="1080p",
|
|
negative_prompt=None,
|
|
audio=None,
|
|
seed=123,
|
|
)
|
|
assert "aspect_ratio" not in p
|
|
assert p["image_url"] == "https://example.com/frame.png"
|
|
assert p["seed"] == 123
|
|
|
|
def test_ltx_25_payload(self):
|
|
"""LTX 2.5: integer duration enum, 4K alias, no seed key."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
meta = FAL_FAMILIES["ltx-2.5"]
|
|
p = _build_payload(
|
|
meta,
|
|
prompt="a drone shot",
|
|
image_url=None,
|
|
duration=7,
|
|
aspect_ratio="16:9",
|
|
resolution="4k",
|
|
negative_prompt="blurry",
|
|
audio=True,
|
|
seed=42,
|
|
)
|
|
# duration snaps to the nearest enum value as a JSON integer; the
|
|
# tool's "4k" maps to the endpoint's "2160p"; seed/negative dropped.
|
|
assert p == {
|
|
"prompt": "a drone shot",
|
|
"aspect_ratio": "16:9",
|
|
"resolution": "2160p",
|
|
"duration": 6,
|
|
"generate_audio": True,
|
|
}
|
|
assert isinstance(p["duration"], int)
|
|
|
|
# i2v: the fast i2v endpoint takes the same even enum (19 snaps to 18 — a tie picks the lower entry, like veo 7→6),
|
|
# no seed key.
|
|
p = _build_payload(meta, prompt="animate", image_url="https://example.com/f.png", duration=19, aspect_ratio="9:16",
|
|
resolution="720p", negative_prompt=None, audio=None, seed=7)
|
|
assert p == {"prompt": "animate", "image_url": "https://example.com/f.png", "aspect_ratio": "9:16", "resolution": "720p", "duration": 18}
|
|
|
|
# fal caps 1440p/2160p at 10s regardless of frame rate: 18 at 720p stays 18, at 4k it is capped to 10; and an
|
|
# unspecified duration is omitted so the endpoint's own default ("auto") applies instead of the enum minimum.
|
|
kw = dict(prompt="x", image_url=None, aspect_ratio="16:9", negative_prompt=None, audio=None, seed=None)
|
|
assert _build_payload(meta, duration=18, resolution="4k", **kw)["duration"] == 10
|
|
assert _build_payload(meta, duration=18, resolution="2k", **kw)["duration"] == 10
|
|
assert _build_payload(meta, duration=18, resolution="1080p", **kw)["duration"] == 18
|
|
assert "duration" not in _build_payload(meta, duration=None, resolution="4k", **kw)
|
|
|
|
def test_kling_o3_payload(self):
|
|
"""Kling O3: string duration, i2v drops aspect_ratio, no seed."""
|
|
from plugins.video_gen.fal import FAL_FAMILIES, _build_payload
|
|
|
|
meta = FAL_FAMILIES["kling-o3"]
|
|
p = _build_payload(
|
|
meta,
|
|
prompt="a mecha lands",
|
|
image_url=None,
|
|
duration=5,
|
|
aspect_ratio="16:9",
|
|
resolution="1080p",
|
|
negative_prompt=None,
|
|
audio=True,
|
|
seed=3,
|
|
)
|
|
# No resolution/seed keys in the O3 schema; duration is the usual
|
|
# stringified queue-API form.
|
|
assert p == {
|
|
"prompt": "a mecha lands",
|
|
"aspect_ratio": "16:9",
|
|
"duration": "5",
|
|
"generate_audio": True,
|
|
}
|
|
|
|
p = _build_payload(
|
|
meta,
|
|
prompt="animate",
|
|
image_url="https://example.com/f.png",
|
|
duration=2,
|
|
aspect_ratio="16:9",
|
|
resolution="1080p",
|
|
negative_prompt=None,
|
|
audio=False,
|
|
seed=None,
|
|
)
|
|
assert "aspect_ratio" not in p
|
|
assert p["duration"] == "3" # clamped to the 3-15 range
|
|
assert p["generate_audio"] is False
|
|
|
|
|
|
class TestUpscalePass:
|
|
"""Opt-in SeedVR2 upscale chain after generation."""
|
|
|
|
@pytest.fixture
|
|
def with_fake_fal(self, monkeypatch):
|
|
"""Stub fal_client.submit, capturing every endpoint hit in order."""
|
|
import sys
|
|
import types
|
|
|
|
captured = {"calls": []}
|
|
|
|
class FakeHandle:
|
|
def __init__(self, endpoint):
|
|
self._endpoint = endpoint
|
|
|
|
def get(self):
|
|
if self._endpoint.endswith("upscale/video"):
|
|
return {"video": {"url": "https://fake/upscaled.mp4"}}
|
|
return {"video": {"url": "https://fake/native.mp4"}}
|
|
|
|
fake = types.ModuleType("fal_client")
|
|
def _submit(endpoint, arguments=None, headers=None):
|
|
captured["calls"].append((endpoint, arguments))
|
|
return FakeHandle(endpoint)
|
|
fake.submit = _submit # type: ignore
|
|
monkeypatch.setitem(sys.modules, "fal_client", fake)
|
|
|
|
from plugins.video_gen import fal as fal_plugin
|
|
fal_plugin._fal_client = None
|
|
fal_plugin._managed_fal_video_client = None
|
|
fal_plugin._managed_fal_video_client_config = None
|
|
|
|
monkeypatch.setenv("FAL_KEY", "test")
|
|
monkeypatch.setattr(fal_plugin, "_resolve_managed_fal_video_gateway", lambda: None)
|
|
return captured
|
|
|
|
def test_upscale_chains_seedvr(self, with_fake_fal):
|
|
from plugins.video_gen.fal import FALVideoGenProvider, UPSCALER_ENDPOINT
|
|
|
|
result = FALVideoGenProvider().generate(
|
|
"a dog", model="pixverse-v6", upscale=True,
|
|
)
|
|
assert result["success"] is True
|
|
assert result["video"] == "https://fake/upscaled.mp4"
|
|
assert result["upscaled"] is True
|
|
assert result["upscale_factor"] == 2
|
|
endpoints = [c[0] for c in with_fake_fal["calls"]]
|
|
assert endpoints == ["fal-ai/pixverse/v6/text-to-video", UPSCALER_ENDPOINT]
|
|
# Upscale request carries the native URL + factor mode.
|
|
upscale_args = with_fake_fal["calls"][1][1]
|
|
assert upscale_args["video_url"] == "https://fake/native.mp4"
|
|
assert upscale_args["upscale_mode"] == "factor"
|
|
|
|
def test_no_upscale_by_default(self, with_fake_fal):
|
|
from plugins.video_gen.fal import FALVideoGenProvider
|
|
|
|
result = FALVideoGenProvider().generate("a dog", model="pixverse-v6")
|
|
assert result["success"] is True
|
|
assert result["video"] == "https://fake/native.mp4"
|
|
assert result["upscaled"] is False
|
|
assert len(with_fake_fal["calls"]) == 1
|
|
|
|
def test_upscale_failure_falls_back_to_native(self, with_fake_fal, monkeypatch):
|
|
from plugins.video_gen import fal as fal_plugin
|
|
from plugins.video_gen.fal import FALVideoGenProvider
|
|
|
|
monkeypatch.setattr(
|
|
fal_plugin,
|
|
"_upscale_video",
|
|
lambda url, source_request_id=None: None,
|
|
)
|
|
result = FALVideoGenProvider().generate(
|
|
"a dog", model="pixverse-v6", upscale=True,
|
|
)
|
|
assert result["success"] is True
|
|
assert result["video"] == "https://fake/native.mp4"
|
|
assert result["upscaled"] is False
|
|
|
|
def test_managed_upscale_binds_the_source_request(self, monkeypatch):
|
|
from plugins.video_gen import fal as fal_plugin
|
|
|
|
captured = {}
|
|
|
|
class FakeHandle:
|
|
def get(self):
|
|
return {"video": {"url": "https://fake/upscaled.mp4"}}
|
|
|
|
monkeypatch.setattr(
|
|
fal_plugin,
|
|
"_resolve_managed_fal_video_gateway",
|
|
lambda: object(),
|
|
)
|
|
monkeypatch.setattr(
|
|
fal_plugin,
|
|
"_submit_fal_video_request",
|
|
lambda endpoint, arguments: (
|
|
captured.update(endpoint=endpoint, arguments=arguments)
|
|
or FakeHandle()
|
|
),
|
|
)
|
|
|
|
assert (
|
|
fal_plugin._upscale_video(
|
|
"https://fake/native.mp4",
|
|
"source-request-1",
|
|
)
|
|
== "https://fake/upscaled.mp4"
|
|
)
|
|
assert captured["arguments"]["source_request_id"] == "source-request-1"
|
|
|
|
def test_managed_upscale_without_source_request_falls_back(self, monkeypatch):
|
|
from plugins.video_gen import fal as fal_plugin
|
|
|
|
submit = Mock()
|
|
monkeypatch.setattr(
|
|
fal_plugin,
|
|
"_resolve_managed_fal_video_gateway",
|
|
lambda: object(),
|
|
)
|
|
monkeypatch.setattr(fal_plugin, "_submit_fal_video_request", submit)
|
|
|
|
assert fal_plugin._upscale_video("https://fake/native.mp4") is None
|
|
submit.assert_not_called()
|