Claude
8e22f17b00
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BzJLJWVu7RZnbRqm3nYAnX
46 lines
2.8 KiB
Python
46 lines
2.8 KiB
Python
"""An LTX project as Dex runs it, over HTTP on the real route table: the direct video-to-video preset (also as saved with
|
|
the old wiring), a project with a video in References and Video input switched on, and every call the app makes at start-up
|
|
and for a render. Each must answer JSON: a page that gets anything else fails on 'r is null'."""
|
|
|
|
import pytest
|
|
|
|
from core.tests.test_routes_pipeline import _request, model_files, registered, server # noqa: F401
|
|
|
|
|
|
def _preset(pid):
|
|
from modules.models.ltx.pipeline import presets
|
|
return next(p["slots"] for p in presets() if p["id"] == pid)
|
|
|
|
|
|
@pytest.mark.parametrize("pid", ["ltx23_video_to_video_direct", "ltx23_image_to_video_references"])
|
|
def test_every_startup_and_render_call_answers_json_on_an_ltx_project(server, pid):
|
|
slots = _preset(pid)
|
|
old = [dict(s, inputs={k: (["source_image", 0] if v == ["source_image", 1] else v) for k, v in s["inputs"].items()}) for s in slots]
|
|
takes_video = any(r.get("video_frames") for x in slots for r in x.get("roles") or [])
|
|
for saved in ((slots, old) if takes_video else (slots,)):
|
|
status, body = _request(server, "POST", "/fun/api/pipeline", {"slots": saved})
|
|
assert status == 200 and "raw" not in body, (status, body)
|
|
users = [(s["id"], k) for s in body["slots"] for k, v in s["inputs"].items() if isinstance(v, list) and v[0] == "source_image"]
|
|
assert all(next(s for s in body["slots"] if s["id"] == i)["inputs"][k][1] == 1 for i, k in users), "the old wiring was not brought up to date"
|
|
status, body = _request(server, "GET", "/fun/api/pipeline")
|
|
assert status == 200 and "slots" in body, body
|
|
|
|
traits = "audio_stream,ltx_av"
|
|
for path in ("/fun/api/story/markers", f"/fun/api/modules?traits={traits}", "/fun/api/modules", "/fun/api/shortcuts",
|
|
"/fun/api/readiness", "/fun/api/system", "/fun/api/addons", "/fun/api/m/render/library", "/fun/api/pipeline/presets"):
|
|
status, body = _request(server, "GET", path)
|
|
assert status < 500 and "raw" not in body, (path, status, body)
|
|
|
|
status, body = _request(server, "GET", f"/fun/api/modules?traits={traits}")
|
|
ids = [m["id"] for m in body.get("modules", [])]
|
|
assert "conditioning_shot_camera" not in ids, "shot camera must be hidden for LTX"
|
|
assert "conditioning_shot_camera" in [m["id"] for m in body.get("incompatible", [])]
|
|
|
|
|
|
def test_a_saved_image_to_video_pipeline_without_the_aspect_source_comes_back_with_it(server):
|
|
slots = [s for s in _preset("ltx23_image_to_video_references") if s["id"] not in ("aspect_source", "aspect")]
|
|
latent = next(s for s in slots if s["id"] == "latent")
|
|
latent["inputs"] = {**latent["inputs"], "width": 768, "height": 512}
|
|
status, body = _request(server, "POST", "/fun/api/pipeline", {"slots": slots})
|
|
assert status == 200 and {"aspect_source", "aspect"} <= {s["id"] for s in body["slots"]}, body
|