test(backend): cover the six zero-coverage modules — ratchet moves up

apps/modal-backend/pyproject.toml

@@ -49,12 +49,12 @@ markers = [
     "paid: spends fal/openrouter; skipped unless its *_BENCH_RUN flag is set",
 ]
 
-# The RATCHET: floor pinned just under the measured baseline (2026-07-18:
-# 84.68% over providers/ + generate.py). CI's pytest runs with --cov so a
+# The RATCHET: floor pinned just under the measured baseline (2026-07-19:
+# 85.72% over providers/ + generate.py). CI's pytest runs with --cov so a
 # PR that drops the total below this FAILS; a PR that raises coverage bumps
 # the floor. Never lower it to make a PR pass — that defeats the ratchet.
 [tool.coverage.report]
-fail_under = 83
+fail_under = 84
 
 [tool.mypy]
 python_version = "3.12"

apps/modal-backend/tests/conftest.py

@@ -120,6 +120,14 @@
     "EDIT_LOOP_RETRY_BUDGET_S",
     "WORLD_BENCH_JUDGE_MODEL",
     "CONTINUITY_BENCH_JUDGE_MODEL",
+    # Video tier pins (providers/video.py) — same hermeticity reasoning.
+    "FAL_ANIMATE_MODEL",
+    "FAL_VIDEO_TIER",
+    "FAL_VIDEO_TIER_FAST",
+    "FAL_VIDEO_TIER_BALANCED",
+    "FAL_VIDEO_TIER_PRO",
+    "LTX_PRO_RESOLUTION",
+    "WAN_RESOLUTION",
     # Map-pan expand (outpaint the world outward).
     "EXPAND_MAP_PAN",
     "FAL_EXPAND_MODEL",

apps/modal-backend/tests/test_deploy_safety.py

@@ -133,6 +133,14 @@ def boom() -> Any:
     assert await moderation.flagged("anything") == (False, "")
 
 
+async def test_moderation_block_without_reason_gets_default(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("MODERATE_PROMPTS", "1")
+    monkeypatch.setattr(llm_mod, "_client", lambda: _fake_client({"allowed": False}))
+    assert await moderation.flagged("bad thing") == (True, "blocked by moderation")
+
+
 async def _collect(agen: Any) -> list[dict[str, Any]]:
     events: list[dict[str, Any]] = []
     async for chunk in agen:

apps/modal-backend/tests/test_local_server.py

@@ -0,0 +1,35 @@
+"""local_server.py import smoke — the module wires up generate's FastAPI app
+without starting uvicorn (the run call sits behind __main__)."""
+
+from __future__ import annotations
+
+import importlib
+import sys
+from unittest.mock import MagicMock
+
+import pytest
+
+
+def test_import_exposes_the_generate_app(monkeypatch: pytest.MonkeyPatch) -> None:
+    # generate.py needs modal at import time; stub it like test_deploy_safety.
+    sys.modules.setdefault("modal", MagicMock())
+    # local_server loads .env files at import — stub so host config stays out.
+    import dotenv
+
+    monkeypatch.setattr(dotenv, "load_dotenv", lambda *a, **k: False)
+
+    sys.modules.pop("local_server", None)
+    local_server = importlib.import_module("local_server")
+
+    from generate import fastapi_app
+
+    assert local_server.fastapi_app is fastapi_app
+    paths = {route.path for route in fastapi_app.routes}
+    assert {
+        "/sse/generate",
+        "/animate",
+        "/resolve-click",
+        "/health",
+        "/status",
+        "/models",
+    } <= paths

apps/modal-backend/tests/test_ltxf.py

@@ -0,0 +1,88 @@
+"""ltxf.py — the LTXF WebSocket envelope and the fMP4 init/media splitter,
+exercised on tiny synthetic byte sequences (pure logic, no I/O)."""
+
+from __future__ import annotations
+
+import json
+import struct
+
+import pytest
+
+import ltxf
+
+# ── encode / decode ─────────────────────────────────────────────────────────
+
+
+def test_encode_decode_round_trip() -> None:
+    header = {"media_type": "video/mp4", "sequence": 3, "is_init_segment": True}
+    payload = b"\x00\x01\x02fmp4-bytes"
+    buf = ltxf.encode(header, payload)
+
+    # wire layout: magic + uint32 BE header length + compact JSON + payload
+    assert buf[:4] == b"LTXF"
+    (header_len,) = struct.unpack(">I", buf[4:8])
+    assert buf[8 : 8 + header_len] == json.dumps(header, separators=(",", ":")).encode()
+    assert buf[8 + header_len :] == payload
+
+    assert ltxf.decode(buf) == ltxf.LTXFPacket(header=header, payload=payload)
+
+
+def test_decode_empty_payload() -> None:
+    assert ltxf.decode(ltxf.encode({"final": True}, b"")).payload == b""
+
+
+@pytest.mark.parametrize(
+    "buf",
+    [b"", b"LTX", b"NOPE" + struct.pack(">I", 0)],
+    ids=["empty", "short", "wrong-magic"],
+)
+def test_decode_rejects_short_or_wrong_magic(buf: bytes) -> None:
+    with pytest.raises(ValueError, match="not an LTXF packet"):
+        ltxf.decode(buf)
+
+
+def test_decode_rejects_truncated_header() -> None:
+    buf = b"LTXF" + struct.pack(">I", 99) + b'{"a":1}'
+    with pytest.raises(ValueError, match="header length exceeds"):
+        ltxf.decode(buf)
+
+
+# ── split_fmp4 ──────────────────────────────────────────────────────────────
+
+
+def _box(box_type: bytes, payload: bytes = b"") -> bytes:
+    return struct.pack(">I", 8 + len(payload)) + box_type + payload
+
+
+def test_split_init_from_media_segments() -> None:
+    ftyp = _box(b"ftyp", b"isom")
+    moov = _box(b"moov", b"\x00" * 12)
+    moof = _box(b"moof", b"\x01" * 4)
+    mdat = _box(b"mdat", b"\x02" * 9)
+    init, media = ltxf.split_fmp4(ftyp + moov + moof + mdat)
+    assert init == ftyp + moov
+    assert media == moof + mdat
+
+
+def test_split_handles_64bit_box_size() -> None:
+    # size field 1 -> real size lives in the 8-byte largesize after the type
+    ftyp = _box(b"ftyp")
+    payload = b"\x00" * 5
+    large_moov = struct.pack(">I", 1) + b"moov" + struct.pack(">Q", 16 + len(payload)) + payload
+    moof = _box(b"moof")
+    init, media = ltxf.split_fmp4(ftyp + large_moov + moof)
+    assert init == ftyp + large_moov
+    assert media == moof
+
+
+def test_split_size_zero_extends_to_end() -> None:
+    ftyp = _box(b"ftyp")
+    open_moov = struct.pack(">I", 0) + b"moov" + b"rest of file"
+    init, media = ltxf.split_fmp4(ftyp + open_moov)
+    assert init == ftyp + open_moov
+    assert media == b""
+
+
+def test_split_without_moov_raises() -> None:
+    with pytest.raises(ValueError, match="no moov"):
+        ltxf.split_fmp4(_box(b"ftyp") + _box(b"free"))

apps/modal-backend/tests/test_provider_fallback.py

@@ -43,6 +43,31 @@ def test_breaker_opens_after_threshold_and_recovers() -> None:
     assert breaker.available(slug)
 
 
+def test_breaker_cooldown_expiry_half_open_probe(monkeypatch: pytest.MonkeyPatch) -> None:
+    """After COOLDOWN_S the circuit half-opens: one probe is allowed, a probe
+    failure re-opens immediately (count stays >= threshold), a probe success
+    closes fully."""
+    slug = "fal-ai/nano-banana-pro"
+    clock = [1000.0]
+    monkeypatch.setattr(breaker.time, "monotonic", lambda: clock[0])
+
+    for _ in range(breaker.FAILURE_THRESHOLD):
+        breaker.record_failure(slug)
+    clock[0] += breaker.COOLDOWN_S - 1
+    assert not breaker.available(slug)  # still cooling
+    clock[0] += 2
+    assert breaker.available(slug)  # cooldown over -> probe allowed
+
+    breaker.record_failure(slug)  # failed probe re-opens on the spot
+    assert not breaker.available(slug)
+
+    clock[0] += breaker.COOLDOWN_S + 1
+    assert breaker.available(slug)
+    breaker.record_success(slug)  # good probe closes and clears the count
+    breaker.record_failure(slug)  # so one fresh failure stays sub-threshold
+    assert breaker.available(slug)
+
+
 def _img(model: str) -> GeneratedImage:
     return GeneratedImage(b"jpeg", "image/jpeg", model, None)
 

apps/modal-backend/tests/test_video.py

@@ -0,0 +1,143 @@
+"""providers/video.py — tier→model resolution and the animate_image
+orchestration, with the fal boundary (`to_fal_url` / `_fal_subscribe`)
+mocked. No network, no spend."""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+
+from providers import video
+
+# ── tier / model resolution ─────────────────────────────────────────────────
+
+
+def test_resolve_tier_default_env_and_garbage(monkeypatch: pytest.MonkeyPatch) -> None:
+    assert video._resolve_video_tier(None) == video.DEFAULT_VIDEO_TIER
+    assert video._resolve_video_tier("PRO") == "pro"  # case-insensitive
+    assert video._resolve_video_tier("bogus") == video.DEFAULT_VIDEO_TIER
+    monkeypatch.setenv("FAL_VIDEO_TIER", "balanced")
+    assert video._resolve_video_tier(None) == "balanced"
+    assert video._resolve_video_tier("pro") == "pro"  # explicit arg beats env
+
+
+def test_animate_model_precedence(monkeypatch: pytest.MonkeyPatch) -> None:
+    # tier slot defaults
+    assert video._animate_model() == video.DEFAULT_ANIMATE_MODEL
+    assert video._animate_model("balanced") == video.TIER_VIDEO_MODELS["balanced"]
+    assert video._animate_model("pro") == video.PRO_ANIMATE_MODEL
+    # per-tier env slot beats the built-in table
+    monkeypatch.setenv("FAL_VIDEO_TIER_BALANCED", "fal-ai/hunyuan-video-i2v")
+    assert video._animate_model("balanced") == "fal-ai/hunyuan-video-i2v"
+    # global override beats everything
+    monkeypatch.setenv("FAL_ANIMATE_MODEL", "fal-ai/custom-i2v")
+    assert video._animate_model("balanced") == "fal-ai/custom-i2v"
+
+
+# ── animate_image orchestration ─────────────────────────────────────────────
+
+
+def _fal_result(**video_over: Any) -> dict[str, Any]:
+    payload: dict[str, Any] = {
+        "url": "https://fal.media/out.mp4",
+        "content_type": "video/mp4",
+        "duration": 4.2,
+    }
+    payload.update(video_over)
+    return {"video": payload}
+
+
+async def _animate(
+    monkeypatch: pytest.MonkeyPatch,
+    *,
+    result: dict[str, Any] | None = None,
+    **kwargs: Any,
+) -> tuple[video.AnimatedClip, AsyncMock]:
+    monkeypatch.setenv("FAL_KEY", "test-key")
+    monkeypatch.setattr(
+        video, "to_fal_url", AsyncMock(return_value="https://fal.media/in.png")
+    )
+    subscribe = AsyncMock(return_value=_fal_result() if result is None else result)
+    monkeypatch.setattr(video, "_fal_subscribe", subscribe)
+    clip = await video.animate_image(
+        image_data_url="data:image/png;base64,AAAA", prompt="gentle pan", **kwargs
+    )
+    return clip, subscribe
+
+
+async def test_requires_fal_key(monkeypatch: pytest.MonkeyPatch) -> None:
+    # conftest scrubs FAL_KEY; the guard must fire before any provider call.
+    subscribe = AsyncMock()
+    monkeypatch.setattr(video, "_fal_subscribe", subscribe)
+    with pytest.raises(RuntimeError, match="FAL_KEY"):
+        await video.animate_image(image_data_url="data:x", prompt="pan")
+    subscribe.assert_not_awaited()
+
+
+async def test_fast_default_sends_plain_arguments(monkeypatch: pytest.MonkeyPatch) -> None:
+    clip, subscribe = await _animate(monkeypatch)
+    model, arguments = subscribe.await_args.args
+    assert model == video.DEFAULT_ANIMATE_MODEL
+    # fast path adds no duration/resolution knobs at all
+    assert arguments == {"image_url": "https://fal.media/in.png", "prompt": "gentle pan"}
+    assert clip == video.AnimatedClip(
+        video_url="https://fal.media/out.mp4",
+        content_type="video/mp4",
+        model=video.DEFAULT_ANIMATE_MODEL,
+        duration_seconds=4.2,
+    )
+
+
+@pytest.mark.parametrize(
+    ("duration", "snapped"),
+    [(3, "6"), (6, "6"), (7, "8"), (8, "8"), (9, "10"), (30, "10")],
+)
+async def test_pro_snaps_duration_to_string_enum(
+    monkeypatch: pytest.MonkeyPatch, duration: int, snapped: str
+) -> None:
+    # LTX-2 wants duration/resolution as STRING enums; ints make fal 502.
+    _, subscribe = await _animate(monkeypatch, tier="pro", duration=duration)
+    model, arguments = subscribe.await_args.args
+    assert model == video.PRO_ANIMATE_MODEL
+    assert arguments["duration"] == snapped
+    assert arguments["resolution"] == "1080p"
+
+
+@pytest.mark.parametrize(("duration", "num_frames"), [(1, 16), (5, 80), (30, 96)])
+async def test_wan_clamps_duration_into_num_frames(
+    monkeypatch: pytest.MonkeyPatch, duration: int, num_frames: int
+) -> None:
+    _, subscribe = await _animate(monkeypatch, tier="balanced", duration=duration)
+    model, arguments = subscribe.await_args.args
+    assert model == "fal-ai/wan-i2v"
+    assert arguments["num_frames"] == num_frames
+    assert arguments["resolution"] == "720p"
+    assert "duration" not in arguments
+
+
+async def test_no_video_payload_raises(monkeypatch: pytest.MonkeyPatch) -> None:
+    with pytest.raises(RuntimeError, match="no video payload"):
+        await _animate(monkeypatch, result={"images": []})
+
+
+async def test_video_without_url_raises(monkeypatch: pytest.MonkeyPatch) -> None:
+    with pytest.raises(RuntimeError, match="without url"):
+        await _animate(monkeypatch, result={"video": {"content_type": "video/mp4"}})
+
+
+async def test_content_type_and_duration_fall_back(monkeypatch: pytest.MonkeyPatch) -> None:
+    clip, _ = await _animate(
+        monkeypatch, result={"video": {"url": "https://fal.media/x.mp4"}}, duration=7
+    )
+    assert clip.content_type == "video/mp4"
+    assert clip.duration_seconds == 7.0  # requested duration when fal omits it
+
+
+# ── data_url_from_bytes ─────────────────────────────────────────────────────
+
+
+def test_data_url_from_bytes() -> None:
+    assert video.data_url_from_bytes(b"abc") == "data:image/jpeg;base64,YWJj"
+    assert video.data_url_from_bytes(b"abc", "image/png") == "data:image/png;base64,YWJj"