Skip to content

test_dynamo_shape

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.

Classes

fastvideo.tests.contract.test_dynamo_shape.TestDynamoHandlerContract

Functions

fastvideo.tests.contract.test_dynamo_shape.TestDynamoHandlerContract.test_handler_serializes_state_back_to_nvext 
test_handler_serializes_state_back_to_nvext()

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.)

Source code in fastvideo/tests/contract/test_dynamo_shape.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
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)

fastvideo.tests.contract.test_dynamo_shape.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.

Functions

fastvideo.tests.contract.test_dynamo_shape.nv_request_to_generation_request 

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.

Source code in fastvideo/tests/contract/test_dynamo_shape.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
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,
    )