[test] streaming: generate_async coverage + refreshed streaming test

* tests/contract/test_generate_async.py: event ordering (Progress ->
  Final); exactly-one-final per request; batch expansion emits one
  Final per sub-result; VideoFinalEvent carries frames / full
  GenerationResult / continuation_state as expected; health-check
  request is minimal and round-trips through normalize_generation_
  request; public fastvideo.api exports the new symbols; a mock
  Dynamo-style async handler wraps generate_async using only the
  public import surface.
* tests/api/test_cli_translation.py: the streaming-dispatch test no
  longer expects NotImplementedError (PR 7.5 landed the live server);
  it now injects a fake run_server and asserts the CLI hands the
  typed streaming ServeConfig through unchanged.

Co-authored-by: Junda (David) Su <90978028+Davids048@users.noreply.github.com>
Co-authored-by: Matthew Noto <99706358+RandNMR73@users.noreply.github.com>
Co-authored-by: XOR-op <17672363+XOR-op@users.noreply.github.com>
Co-authored-by: Zhang Peiyuan <42993249+jzhang38@users.noreply.github.com>
This commit is contained in:
SolitaryThinker
2026-05-05 03:05:53 -07:00
co-authored by Junda Su Matthew Noto XOR-op Zhang Peiyuan
parent 00aa56498e
commit 6ae7a99f18
@@ -0,0 +1,273 @@
# SPDX-License-Identifier: Apache-2.0
"""Contract tests for ``VideoGenerator.generate_async``.
These tests monkey-patch the synchronous ``_generate_request_impl`` so
the suite runs CPU-only -- the async wrapper is the piece under test,
not the pipeline.
"""
from __future__ import annotations
import asyncio
from typing import Any
import numpy as np
import pytest
from fastvideo.api.results import (
GenerationResult,
VideoFinalEvent,
VideoProgressEvent,
)
from fastvideo.api.schema import (
ContinuationState,
GenerationRequest,
OutputConfig,
SamplingConfig,
)
class _FakeExecutor:
def set_log_queue(self, q):
self.log_queue = q
def clear_log_queue(self):
self.log_queue = None
class _FakeVideoGenerator:
"""Stand-in that exposes the same generate_async contract as the
real VideoGenerator without requiring a loaded pipeline. Binds the
real method via ``__func__`` so the implementation under test is
exactly the production code."""
def __init__(self, result: GenerationResult | list[GenerationResult]):
self._result = result
self.executor = _FakeExecutor()
def _generate_request_impl(
self, request: GenerationRequest,
) -> GenerationResult | list[GenerationResult]:
return self._result
@staticmethod
def _wrap_legacy_result(result):
return GenerationResult.from_legacy_result(result)
@classmethod
def bind(cls, result):
from fastvideo.entrypoints.video_generator import VideoGenerator
instance = cls(result)
instance.generate_async = VideoGenerator.generate_async.__get__(
instance, cls)
instance.default_health_check_request = (
VideoGenerator.default_health_check_request)
return instance
def _make_request(**overrides: Any) -> GenerationRequest:
base = GenerationRequest(
prompt="hi",
sampling=SamplingConfig(num_inference_steps=4, height=64, width=64,
num_frames=8),
output=OutputConfig(save_video=False, return_frames=False),
)
for key, value in overrides.items():
setattr(base.sampling, key, value)
return base
def _make_result(state: ContinuationState | None = None) -> GenerationResult:
return GenerationResult(
prompt="hi",
frames=[np.zeros((64, 64, 3), dtype=np.uint8)],
samples=None,
generation_time=0.01,
state=state,
)
class TestEventOrdering:
def test_emits_progress_then_final(self):
gen = _FakeVideoGenerator.bind(_make_result())
async def run():
events = []
async for evt in gen.generate_async(_make_request()):
events.append(evt)
return events
events = asyncio.run(run())
assert len(events) == 2
assert isinstance(events[0], VideoProgressEvent)
assert isinstance(events[1], VideoFinalEvent)
assert events[0].step == 0
assert events[0].total_steps == 4
def test_exactly_one_final_event_per_request(self):
gen = _FakeVideoGenerator.bind(_make_result())
async def run():
events = []
async for evt in gen.generate_async(_make_request()):
events.append(evt)
return events
events = asyncio.run(run())
finals = [e for e in events if isinstance(e, VideoFinalEvent)]
assert len(finals) == 1
def test_batch_expansion_emits_one_final_per_result(self):
batch_result = [_make_result(), _make_result()]
gen = _FakeVideoGenerator.bind(batch_result)
async def run():
events = []
async for evt in gen.generate_async(_make_request()):
events.append(evt)
return events
events = asyncio.run(run())
finals = [e for e in events if isinstance(e, VideoFinalEvent)]
assert len(finals) == 2
class TestFinalEventShape:
def test_carries_frames_and_result(self):
result = _make_result()
gen = _FakeVideoGenerator.bind(result)
async def run() -> VideoFinalEvent:
async for evt in gen.generate_async(_make_request()):
if isinstance(evt, VideoFinalEvent):
return evt
raise AssertionError("no final event")
final = asyncio.run(run())
assert final.frames is result.frames
assert final.result is result
assert final.metadata["generation_time"] == 0.01
def test_carries_continuation_state_when_present(self):
state = ContinuationState(
kind="ltx2.v1", payload={"schema_version": 1, "segment_index": 3})
result = _make_result(state=state)
gen = _FakeVideoGenerator.bind(result)
async def run() -> VideoFinalEvent:
async for evt in gen.generate_async(_make_request()):
if isinstance(evt, VideoFinalEvent):
return evt
raise AssertionError("no final")
final = asyncio.run(run())
assert final.continuation_state is state
def test_no_state_when_result_has_none(self):
gen = _FakeVideoGenerator.bind(_make_result())
async def run() -> VideoFinalEvent:
async for evt in gen.generate_async(_make_request()):
if isinstance(evt, VideoFinalEvent):
return evt
raise AssertionError("no final")
final = asyncio.run(run())
assert final.continuation_state is None
class TestHealthCheckRequest:
def test_returns_minimal_workload(self):
from fastvideo.entrypoints.video_generator import VideoGenerator
req = VideoGenerator.default_health_check_request()
assert isinstance(req, GenerationRequest)
assert req.sampling.num_inference_steps == 1
assert req.sampling.num_frames == 8
assert req.sampling.width == 256
assert req.sampling.height == 256
def test_does_not_persist_output(self):
from fastvideo.entrypoints.video_generator import VideoGenerator
req = VideoGenerator.default_health_check_request()
assert req.output.save_video is False
assert req.output.return_frames is False
def test_round_trips_through_normalization(self):
from fastvideo.api.compat import normalize_generation_request
from fastvideo.entrypoints.video_generator import VideoGenerator
req = VideoGenerator.default_health_check_request()
normalized = normalize_generation_request(req)
assert normalized.prompt == "health check"
assert normalized.sampling.num_inference_steps == 1
class TestPublicExports:
def test_new_event_types_available_via_fastvideo_api(self):
import fastvideo.api as api
for name in (
"VideoEvent",
"VideoFinalEvent",
"VideoPartialEvent",
"VideoProgressEvent",
"VideoResult",
):
assert name in api.__all__
def test_video_result_is_generation_result_alias(self):
from fastvideo.api import GenerationResult as Public
from fastvideo.api import VideoResult
assert VideoResult is Public
class TestDynamoStyleHandlerIntegration:
"""Mirror the shape the Dynamo backend package uses."""
def test_async_handler_yields_typed_events_without_internal_imports(self):
# The import set is exactly what the Dynamo backend uses.
from fastvideo.api import ( # noqa: F401
ContinuationState,
GenerationRequest,
InputConfig,
OutputConfig,
SamplingConfig,
VideoFinalEvent,
VideoProgressEvent,
)
gen = _FakeVideoGenerator.bind(_make_result())
async def handler(request_dict, context):
# Adapter: dict -> typed request.
req = GenerationRequest(
prompt=request_dict["prompt"],
sampling=SamplingConfig(
num_inference_steps=1, num_frames=8,
height=256, width=256),
output=OutputConfig(save_video=False, return_frames=False),
)
async for event in gen.generate_async(req):
if isinstance(event, VideoFinalEvent):
yield {"type": "final"}
elif isinstance(event, VideoProgressEvent):
yield {"type": "progress", "step": event.step}
async def drive():
out = []
async for chunk in handler({"prompt": "x"}, None):
out.append(chunk)
return out
chunks = asyncio.run(drive())
types = [c["type"] for c in chunks]
assert "progress" in types
assert types[-1] == "final"