# SPDX-License-Identifier: Apache-2.0 """Contract test: a mock Dynamo-style handler wraps FastVideo's public API without touching any private module. The Dynamo backend package (``components/src/dynamo/fastvideo/`` in the Dynamo repo) imports only these symbols: from fastvideo import VideoGenerator from fastvideo.api import ( ContinuationState, GenerationRequest, InputConfig, OutputConfig, SamplingConfig, ) If a FastVideo refactor breaks the adapter shape this test fails at FastVideo CI — before the Dynamo-side integration knows. The plan (PR 7.10) requires the backend to be expressible without flat legacy LTX-2 kwargs or FastVideo-internal imports; this file asserts the subset that exists today and is stable. """ from __future__ import annotations import asyncio import base64 import importlib import sys import time from dataclasses import dataclass from typing import Any, AsyncIterator import pytest # Public imports only — mirror what the Dynamo backend package uses. from fastvideo.api import ( ContinuationState, GenerationRequest, InputConfig, OutputConfig, SamplingConfig, ) # ---------------------------------------------------------------------- # Fake Dynamo-side types (the real ones live in dynamo.llm and dynamo.*) # We mimic their dict shape so the adapter code under test is realistic. # ---------------------------------------------------------------------- @dataclass class _FakeContext: """Stand-in for Dynamo's async RPC context.""" request_id: str = "test-req" def _nv_create_video_request(**overrides: Any) -> dict[str, Any]: """Build a Dynamo ``NvCreateVideoRequest``-shaped dict.""" base: dict[str, Any] = { "prompt": "a fox running through snow", "size": "1024x1536", "seconds": 5, "response_format": "b64_json", "model": "Lightricks/LTX-Video", "nvext": { "fps": 24, "num_inference_steps": 8, "guidance_scale": 1.0, "seed": 42, "negative_prompt": "blurry", }, } base.update(overrides) if "nvext" in overrides: base["nvext"] = {**base["nvext"], **overrides["nvext"]} return base # ---------------------------------------------------------------------- # The adapter under test — written as if it lived in # components/src/dynamo/fastvideo/request_handlers/video_generation/. # ---------------------------------------------------------------------- def _parse_size(size: str | None) -> tuple[int, int]: if not size or "x" not in size: return 1024, 1536 w, h = size.split("x", 1) return int(w), int(h) def nv_request_to_generation_request(request: dict[str, Any], ) -> GenerationRequest: """Translate Dynamo's request shape into FastVideo's typed request. This function is the template integrators copy into the Dynamo repo. It uses only public FastVideo symbols. """ nvext = request.get("nvext") or {} width, height = _parse_size(request.get("size")) fps = nvext.get("fps") or 24 num_frames = nvext.get("num_frames") or (request.get("seconds") or 4) * fps state = nvext.get("continuation_state") if isinstance(state, dict): state = ContinuationState(kind=state["kind"], payload=state["payload"]) return GenerationRequest( prompt=request["prompt"], negative_prompt=nvext.get("negative_prompt"), inputs=InputConfig(image_path=request.get("input_reference")), sampling=SamplingConfig( width=width, height=height, num_frames=num_frames, fps=fps, num_inference_steps=nvext.get("num_inference_steps", 50), guidance_scale=nvext.get("guidance_scale", 1.0), seed=nvext.get("seed", 1024), true_cfg_scale=nvext.get("true_cfg_scale"), ), output=OutputConfig(save_video=False, return_frames=False), state=state, ) class _MockFastVideoHandler: """Mock ``VideoGenerationWorkerHandler`` — the Dynamo backend's entry point. Shape verified: ``async def generate(dict, ctx) -> AsyncGenerator[dict, None]``, the signature Dynamo's ``endpoint.serve_endpoint(...)`` expects. """ def __init__(self, fake_result: dict[str, Any]) -> None: self._fake_result = fake_result self._lock = asyncio.Lock() async def generate( self, request: dict[str, Any], context: _FakeContext, ) -> AsyncIterator[dict[str, Any]]: req = nv_request_to_generation_request(request) assert isinstance(req, GenerationRequest) # The real handler would call: # result = await asyncio.to_thread(self.generator.generate, req) # For contract purposes we just assert the typed request reaches # the adapter boundary and yield a synthetic final event. t0 = time.perf_counter() async with self._lock: await asyncio.sleep(0) # model call stub elapsed = time.perf_counter() - t0 yield { "data": [{ "b64_json": self._fake_result["b64_json"] }], "inference_time_s": elapsed, "model": request.get("model"), "nvext": { "continuation_state": (None if req.state is None else { "kind": req.state.kind, "payload": req.state.payload, }), }, } # ---------------------------------------------------------------------- # Tests # ---------------------------------------------------------------------- class TestDynamoAdapterShape: def test_nv_request_translates_to_typed_request(self): req = nv_request_to_generation_request(_nv_create_video_request()) assert isinstance(req, GenerationRequest) assert req.prompt == "a fox running through snow" assert req.sampling.width == 1024 assert req.sampling.height == 1536 assert req.sampling.num_frames == 120 # 5 seconds * 24 fps assert req.sampling.fps == 24 assert req.sampling.num_inference_steps == 8 assert req.sampling.guidance_scale == 1.0 assert req.sampling.seed == 42 # negative_prompt lives at the request level, not inside sampling assert req.negative_prompt == "blurry" def test_nvext_num_frames_overrides_seconds_times_fps(self): req = nv_request_to_generation_request(_nv_create_video_request(nvext={"num_frames": 33})) assert req.sampling.num_frames == 33 def test_input_reference_maps_to_image_path(self): req = nv_request_to_generation_request(_nv_create_video_request(input_reference="/tmp/init.png")) assert req.inputs.image_path == "/tmp/init.png" def test_continuation_state_round_trips_through_adapter(self): state = { "kind": "ltx2.v1", "payload": { "schema_version": 1, "segment_index": 2, }, } req = nv_request_to_generation_request(_nv_create_video_request(nvext={"continuation_state": state})) assert isinstance(req.state, ContinuationState) assert req.state.kind == "ltx2.v1" assert req.state.payload["segment_index"] == 2 class TestDynamoHandlerContract: def test_handler_generate_is_async_generator(self): handler = _MockFastVideoHandler({"b64_json": "xyz"}) async def run(): out = [] async for chunk in handler.generate(_nv_create_video_request(), _FakeContext()): out.append(chunk) return out chunks = asyncio.run(run()) assert len(chunks) == 1 final = chunks[0] assert final["data"][0]["b64_json"] == "xyz" assert "inference_time_s" in final assert final["nvext"]["continuation_state"] is None def test_handler_serializes_state_back_to_nvext(self): """When the request carries state, the handler should be able to include a matching serialized state on the response. (Dynamo's NvVideosResponse has nvext.continuation_state reserved for this in the pending disaggregation path.)""" handler = _MockFastVideoHandler({"b64_json": "abc"}) state_in = { "kind": "ltx2.v1", "payload": { "schema_version": 1, "segment_index": 4 }, } async def run(): async for chunk in handler.generate(_nv_create_video_request(nvext={"continuation_state": state_in}), _FakeContext()): return chunk raise AssertionError("no chunk emitted") chunk = asyncio.run(run()) assert chunk["nvext"]["continuation_state"]["kind"] == "ltx2.v1" assert (chunk["nvext"]["continuation_state"]["payload"]["segment_index"] == 4) class TestNoInternalImports: """The adapter template in this file imports only the public surface. Any change to FastVideo that requires the Dynamo adapter to reach into a private module would make this test fail at review time. """ _PUBLIC_IMPORTS = frozenset({ "fastvideo", "fastvideo.api", }) _BANNED_PREFIXES = ( "fastvideo.pipelines.", "fastvideo.configs.", "fastvideo.fastvideo_args", "fastvideo.api.compat", "fastvideo.api.parser", "fastvideo.api.overrides", "fastvideo.api.errors", "fastvideo.api.request_metadata", "fastvideo.api.sampling_param", ) def test_adapter_template_does_not_import_private_modules(self): # Scan only the adapter function's source — not the whole test # file — so the banned-prefix list itself doesn't trip the guard. import inspect adapter_source = inspect.getsource(nv_request_to_generation_request) for banned in self._BANNED_PREFIXES: assert banned not in adapter_source, (f"Dynamo adapter template must not depend on {banned}*; " "found a reference in the adapter function body.") def test_public_types_are_stable_imports(self): # These lines match the exact import snippet in docs/design/ # server_contracts/dynamo.md; any rename breaks CI. from fastvideo import VideoGenerator # noqa: F401 from fastvideo.api import ( # noqa: F401 ContinuationState, GenerationRequest, InputConfig, OutputConfig, SamplingConfig, ) @pytest.mark.skipif( "fastvideo.api" not in sys.modules, reason="fastvideo.api not importable in this environment", ) def test_public_api_re_exports_match_doc(self): import fastvideo.api as api required = { "ContinuationState", "GenerationRequest", "InputConfig", "OutputConfig", "SamplingConfig", } missing = required - set(api.__all__) assert not missing, (f"fastvideo.api.__all__ missing Dynamo-contract symbols: {missing}") def _read_own_source() -> str: import pathlib return pathlib.Path(__file__).read_text() # Convenience export for third parties writing their own contract tests. __all__ = [ "nv_request_to_generation_request", ]