450 lines
16 KiB
Python
450 lines
16 KiB
Python
"""Unit tests for the OpenAI-compatible API server helpers (no GPU needed)."""
|
|
|
|
import os
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from fastvideo.api.parser import parse_config
|
|
from fastvideo.api.schema import GenerationRequest
|
|
from fastvideo.entrypoints.openai.protocol import (
|
|
ImageGenerationsRequest,
|
|
ImageResponseData,
|
|
VideoGenerationsRequest,
|
|
VideoListResponse,
|
|
VideoResponse,
|
|
generate_request_id,
|
|
)
|
|
from fastvideo.entrypoints.openai.utils import (
|
|
choose_image_ext,
|
|
merge_image_input_list,
|
|
parse_size,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# parse_size
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestParseSize:
|
|
|
|
def test_valid(self):
|
|
assert parse_size("1024x768") == (1024, 768)
|
|
|
|
def test_valid_uppercase(self):
|
|
assert parse_size("512X512") == (512, 512)
|
|
|
|
def test_valid_with_spaces(self):
|
|
assert parse_size(" 720 x 480 ") == (720, 480)
|
|
|
|
def test_invalid_single_number(self):
|
|
assert parse_size("1024") == (None, None)
|
|
|
|
def test_invalid_non_numeric(self):
|
|
assert parse_size("widexhigh") == (None, None)
|
|
|
|
def test_invalid_empty(self):
|
|
assert parse_size("") == (None, None)
|
|
|
|
def test_invalid_triple(self):
|
|
assert parse_size("1x2x3") == (None, None)
|
|
|
|
def test_zero_dimensions(self):
|
|
assert parse_size("0x0") == (0, 0)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# choose_image_ext
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestChooseImageExt:
|
|
|
|
def test_explicit_png(self):
|
|
assert choose_image_ext("png", None) == "png"
|
|
|
|
def test_explicit_webp(self):
|
|
assert choose_image_ext("webp", None) == "webp"
|
|
|
|
def test_explicit_jpeg_normalises(self):
|
|
assert choose_image_ext("jpeg", None) == "jpg"
|
|
|
|
def test_explicit_jpg(self):
|
|
assert choose_image_ext("jpg", None) == "jpg"
|
|
|
|
def test_transparent_background_defaults_png(self):
|
|
assert choose_image_ext(None, "transparent") == "png"
|
|
|
|
def test_opaque_background_defaults_jpg(self):
|
|
assert choose_image_ext(None, "opaque") == "jpg"
|
|
|
|
def test_no_args_defaults_jpg(self):
|
|
assert choose_image_ext(None, None) == "jpg"
|
|
|
|
def test_format_overrides_background(self):
|
|
assert choose_image_ext("webp", "transparent") == "webp"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# merge_image_input_list
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMergeImageInputList:
|
|
|
|
def test_none_inputs(self):
|
|
assert merge_image_input_list(None, None) == []
|
|
|
|
def test_single_item(self):
|
|
assert merge_image_input_list("a") == ["a"]
|
|
|
|
def test_list_input(self):
|
|
assert merge_image_input_list(["a", "b"]) == ["a", "b"]
|
|
|
|
def test_mixed_inputs(self):
|
|
result = merge_image_input_list(None, ["x", "y"], "z")
|
|
assert result == ["x", "y", "z"]
|
|
|
|
def test_empty_list_ignored(self):
|
|
assert merge_image_input_list([], "a") == ["a"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# image_api._build_generation_kwargs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestImageBuildGenerationKwargs:
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _patch_output_dir(self, tmp_path):
|
|
with patch(
|
|
"fastvideo.entrypoints.openai.image_api.get_output_dir",
|
|
return_value=str(tmp_path),
|
|
):
|
|
yield tmp_path
|
|
|
|
def _build(self, **overrides):
|
|
from fastvideo.entrypoints.openai.image_api import (
|
|
_build_generation_kwargs, )
|
|
|
|
defaults = dict(request_id="req-1", prompt="a cat")
|
|
defaults.update(overrides)
|
|
return _build_generation_kwargs(**defaults)
|
|
|
|
def test_output_path_under_images_subdir(self, tmp_path):
|
|
kw = self._build()
|
|
assert kw["output_path"].startswith(os.path.join(str(tmp_path), "images"))
|
|
|
|
def test_num_frames_always_one(self):
|
|
kw = self._build()
|
|
assert kw["num_frames"] == 1
|
|
|
|
def test_size_parsed(self):
|
|
kw = self._build(size="640x480")
|
|
assert kw["width"] == 640
|
|
assert kw["height"] == 480
|
|
|
|
def test_seed_forwarded(self):
|
|
kw = self._build(seed=42)
|
|
assert kw["seed"] == 42
|
|
|
|
def test_n_clamped(self):
|
|
kw = self._build(n=20)
|
|
assert kw["num_videos_per_prompt"] == 10
|
|
|
|
def test_extension_jpg_default(self):
|
|
kw = self._build()
|
|
assert kw["output_path"].endswith(".jpg")
|
|
|
|
def test_extension_png_for_transparent(self):
|
|
kw = self._build(background="transparent")
|
|
assert kw["output_path"].endswith(".png")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# video_api._build_generation_kwargs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestVideoBuildGenerationKwargs:
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _patch_output_dir(self, tmp_path):
|
|
with patch(
|
|
"fastvideo.entrypoints.openai.video_api.get_output_dir",
|
|
return_value=str(tmp_path),
|
|
):
|
|
yield tmp_path
|
|
|
|
def _build(self, **overrides):
|
|
from fastvideo.entrypoints.openai.video_api import (
|
|
_build_generation_kwargs, )
|
|
|
|
defaults = dict(prompt="a running dog", seconds=4)
|
|
defaults.update(overrides)
|
|
req = VideoGenerationsRequest(**defaults)
|
|
return _build_generation_kwargs("req-v1", req)
|
|
|
|
def test_output_path_under_videos_subdir(self, tmp_path):
|
|
kw = self._build()
|
|
assert kw["output_path"].startswith(os.path.join(str(tmp_path), "videos"))
|
|
assert kw["output_path"].endswith(".mp4")
|
|
|
|
def test_fps_defaults_24(self):
|
|
kw = self._build()
|
|
assert kw["fps"] == 24
|
|
|
|
def test_num_frames_from_seconds(self):
|
|
kw = self._build(seconds=2, fps=30)
|
|
assert kw["num_frames"] == 60
|
|
|
|
def test_explicit_num_frames_overrides_seconds(self):
|
|
kw = self._build(seconds=10, num_frames=5)
|
|
assert kw["num_frames"] == 5
|
|
|
|
def test_seed_forwarded(self):
|
|
kw = self._build(seed=123)
|
|
assert kw["seed"] == 123
|
|
|
|
def test_custom_output_path_overrides_default(self, tmp_path):
|
|
custom = str(tmp_path / "custom")
|
|
kw = self._build(output_path=custom)
|
|
assert kw["output_path"].startswith(custom)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# video_api._build_generation_kwargs with ServeConfig.default_request
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_default_request(raw: dict) -> GenerationRequest:
|
|
"""Parse a raw config dict into a tracked GenerationRequest."""
|
|
return parse_config(GenerationRequest, raw)
|
|
|
|
|
|
class TestVideoDefaultRequestMerge:
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _patch_output_dir(self, tmp_path):
|
|
with patch(
|
|
"fastvideo.entrypoints.openai.video_api.get_output_dir",
|
|
return_value=str(tmp_path),
|
|
):
|
|
yield tmp_path
|
|
|
|
def _build(self, default_raw=None, **body_overrides):
|
|
from fastvideo.entrypoints.openai.video_api import (
|
|
_build_generation_kwargs, )
|
|
|
|
body_defaults = dict(prompt="a running dog", seconds=4)
|
|
body_defaults.update(body_overrides)
|
|
req = VideoGenerationsRequest(**body_defaults)
|
|
default_request = _make_default_request(default_raw) if default_raw else None
|
|
return _build_generation_kwargs("req-v1", req, default_request=default_request)
|
|
|
|
def test_default_seed_flows_through_when_body_omits(self):
|
|
kw = self._build(default_raw={"sampling": {"seed": 42}})
|
|
assert kw["seed"] == 42
|
|
|
|
def test_body_seed_overrides_default(self):
|
|
kw = self._build(
|
|
default_raw={"sampling": {
|
|
"seed": 42
|
|
}},
|
|
seed=7,
|
|
)
|
|
assert kw["seed"] == 7
|
|
|
|
def test_default_fps_used_for_num_frames_from_seconds(self):
|
|
# Default fps=30, body only provides seconds=2 -> num_frames=60.
|
|
kw = self._build(default_raw={"sampling": {"fps": 30}}, seconds=2)
|
|
assert kw["fps"] == 30
|
|
assert kw["num_frames"] == 60
|
|
|
|
def test_default_guidance_scale_preserved_when_body_omits(self):
|
|
kw = self._build(default_raw={"sampling": {"guidance_scale": 5.5}})
|
|
assert kw["guidance_scale"] == 5.5
|
|
|
|
def test_body_guidance_scale_overrides_default(self):
|
|
kw = self._build(
|
|
default_raw={"sampling": {
|
|
"guidance_scale": 5.5
|
|
}},
|
|
guidance_scale=9.0,
|
|
)
|
|
assert kw["guidance_scale"] == 9.0
|
|
|
|
def test_body_size_overrides_default_sampling_dims(self):
|
|
kw = self._build(
|
|
default_raw={"sampling": {
|
|
"width": 640,
|
|
"height": 360
|
|
}},
|
|
size="1024x576",
|
|
)
|
|
assert kw["width"] == 1024
|
|
assert kw["height"] == 576
|
|
|
|
def test_default_width_height_preserved_when_body_omits_size(self):
|
|
kw = self._build(default_raw={"sampling": {"width": 640, "height": 360}})
|
|
assert kw["width"] == 640
|
|
assert kw["height"] == 360
|
|
|
|
def test_default_output_path_used_as_output_dir(self, tmp_path):
|
|
custom = str(tmp_path / "from_default")
|
|
kw = self._build(default_raw={"output": {"output_path": custom}})
|
|
assert kw["output_path"].startswith(custom)
|
|
|
|
def test_body_output_path_overrides_default(self, tmp_path):
|
|
body_dir = str(tmp_path / "body")
|
|
default_dir = str(tmp_path / "default")
|
|
kw = self._build(
|
|
default_raw={"output": {
|
|
"output_path": default_dir
|
|
}},
|
|
output_path=body_dir,
|
|
)
|
|
assert kw["output_path"].startswith(body_dir)
|
|
assert default_dir not in kw["output_path"]
|
|
|
|
def test_default_negative_prompt_flows_through(self):
|
|
kw = self._build(default_raw={"negative_prompt": "low quality, blur"})
|
|
assert kw["negative_prompt"] == "low quality, blur"
|
|
|
|
def test_body_negative_prompt_overrides_default(self):
|
|
kw = self._build(
|
|
default_raw={"negative_prompt": "low quality"},
|
|
negative_prompt="watermark",
|
|
)
|
|
assert kw["negative_prompt"] == "watermark"
|
|
|
|
def test_no_default_request_behaves_like_before(self):
|
|
kw = self._build(seed=123, fps=24)
|
|
assert kw["seed"] == 123
|
|
assert kw["fps"] == 24
|
|
|
|
def test_default_request_not_mutated_by_build(self):
|
|
# Merge should operate on a fresh copy (caller supplies a clone);
|
|
# the helper itself must not mutate the passed-in default.
|
|
default_request = _make_default_request({"sampling": {"seed": 42, "fps": 30}})
|
|
from fastvideo.entrypoints.openai.video_api import (
|
|
_build_generation_kwargs, )
|
|
req = VideoGenerationsRequest(prompt="p", seconds=1)
|
|
_ = _build_generation_kwargs("req-1", req, default_request=default_request)
|
|
assert default_request.sampling.seed == 42
|
|
assert default_request.sampling.fps == 30
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Preset stage-override validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestValidateDefaultRequestAgainstPreset:
|
|
"""Startup-time validation lives in api_server._validate_default_request_against_preset.
|
|
|
|
Called once by ``run_server`` before the FastAPI app is created — the
|
|
default_request is static server config, so per-request re-validation
|
|
would be pure overhead.
|
|
"""
|
|
|
|
def test_empty_stage_overrides_is_noop(self):
|
|
from fastvideo.entrypoints.openai.api_server import (
|
|
_validate_default_request_against_preset, )
|
|
default_request = _make_default_request({"sampling": {"seed": 42}})
|
|
_validate_default_request_against_preset(default_request, "any/model")
|
|
|
|
def test_unknown_model_path_is_noop(self):
|
|
from fastvideo.entrypoints.openai.api_server import (
|
|
_validate_default_request_against_preset, )
|
|
default_request = _make_default_request({"stage_overrides": {"denoise": {"num_inference_steps": 10}}})
|
|
with patch(
|
|
"fastvideo.entrypoints.openai.api_server.get_preset_selection",
|
|
return_value=(None, None),
|
|
):
|
|
_validate_default_request_against_preset(default_request, "unknown/model")
|
|
|
|
def test_unknown_stage_name_raises(self):
|
|
from fastvideo.api.errors import ConfigValidationError
|
|
from fastvideo.entrypoints.openai.api_server import (
|
|
_validate_default_request_against_preset, )
|
|
default_request = _make_default_request({"stage_overrides": {"not_a_real_stage": {"num_inference_steps": 10}}})
|
|
with patch(
|
|
"fastvideo.entrypoints.openai.api_server.get_preset_selection",
|
|
return_value=("wan_t2v_1_3b", "wan"),
|
|
):
|
|
with pytest.raises(ConfigValidationError):
|
|
_validate_default_request_against_preset(default_request, "Wan-AI/Wan2.1-T2V-1.3B-Diffusers")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Server state accessors
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDefaultRequestState:
|
|
|
|
def test_set_and_get_default_request(self):
|
|
from fastvideo.entrypoints.openai import state as state_mod
|
|
saved = state_mod._default_request
|
|
try:
|
|
dr = _make_default_request({"sampling": {"seed": 7}})
|
|
state_mod.set_state.__wrapped__ if False else None # keep lint happy
|
|
state_mod._default_request = dr
|
|
assert state_mod.get_default_request() is dr
|
|
finally:
|
|
state_mod._default_request = saved
|
|
|
|
def test_clear_state_resets_default_request(self):
|
|
from fastvideo.entrypoints.openai import state as state_mod
|
|
saved = state_mod._default_request
|
|
try:
|
|
state_mod._default_request = _make_default_request({"sampling": {"seed": 7}})
|
|
state_mod.clear_state()
|
|
assert state_mod.get_default_request() is None
|
|
finally:
|
|
state_mod._default_request = saved
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Protocol Pydantic models
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestProtocolModels:
|
|
|
|
def test_image_request_required_fields(self):
|
|
req = ImageGenerationsRequest(prompt="test")
|
|
assert req.prompt == "test"
|
|
assert req.n == 1
|
|
assert req.size == "1024x1024"
|
|
|
|
def test_image_request_missing_prompt_raises(self):
|
|
with pytest.raises(Exception):
|
|
ImageGenerationsRequest()
|
|
|
|
def test_video_request_defaults(self):
|
|
req = VideoGenerationsRequest(prompt="hello")
|
|
assert req.seconds == 4
|
|
assert req.seed == 1024
|
|
|
|
def test_video_response_defaults(self):
|
|
resp = VideoResponse(id="v1")
|
|
assert resp.status == "queued"
|
|
assert resp.object == "video"
|
|
|
|
def test_image_response_data_optional_fields(self):
|
|
d = ImageResponseData()
|
|
assert d.b64_json is None
|
|
assert d.url is None
|
|
|
|
def test_generate_request_id_unique(self):
|
|
ids = {generate_request_id() for _ in range(100)}
|
|
assert len(ids) == 100
|
|
|
|
def test_video_list_response(self):
|
|
resp = VideoListResponse(data=[VideoResponse(id="a"), VideoResponse(id="b")])
|
|
assert len(resp.data) == 2
|
|
assert resp.object == "list"
|