[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:
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"
|
||||
Reference in New Issue
Block a user