litellm/tests/unit/videos/test_utils.py

"""
Pure-logic contract tests for litellm/videos/main.py's request utils
(litellm/videos/utils.py: VideoGenerationRequestUtils).

These lock the exact param-shaping behavior so a mutation that drops a filter,
flips a precedence, or stops removing a key fails loudly. The only seam is the
provider config's map_openai_params (a provider boundary); filter_out_litellm_params
runs for real, so the "litellm-internal params get stripped" assertions reflect
production. Every test asserts the exact resulting dict, never "ran without error".
"""

from unittest.mock import MagicMock
from typing import Final

import pytest



import litellm
from litellm.exceptions import UnsupportedParamsError
from litellm.llms.volcengine.videos.transformation import VolcEngineVideoConfig
from litellm.llms.vertex_ai.videos.transformation import VertexAIVideoConfig
from litellm.videos.utils import VideoGenerationRequestUtils

get_requested = (
    VideoGenerationRequestUtils.get_requested_video_generation_optional_param
)


@pytest.mark.parametrize("drop_params", [False, True])
def test_extra_body_video_fields_remain_raw(drop_params: bool) -> None:
    extra: Final = {"seed": 0, "duration": 6, "aspect_ratio": "1:1", "generate_audio": False,
                   "frame_images": [], "input_references": []}
    params: Final = get_requested({"extra_body": extra})
    assert get_optional("veo-3.1-generate-001", VertexAIVideoConfig(), params, drop_params=drop_params) == extra
get_optional = VideoGenerationRequestUtils.get_optional_params_video_generation


# =========================================================================== #
# get_requested_video_generation_optional_param
#
# Receives the caller's full local_vars; must return only the API-bound optional
# params. filter_out_litellm_params strips known internal keys for real; the
# values used below were chosen against the live set: seconds/size/user/foo_param/
# vertex_project/extra/a/b survive, api_key/metadata/litellm_* are stripped.
# =========================================================================== #


def test_requested__drops_none_and_excluded_keys():
    result = get_requested(
        {
            "seconds": "8",
            "size": None,  # None -> dropped
            "prompt": "a sunset",  # excluded
            "model": "sora-2",  # excluded
            "user": "u1",
        }
    )
    assert result == {"seconds": "8", "user": "u1"}


def test_requested__strips_litellm_internal_params():
    result = get_requested(
        {
            "seconds": "8",
            "api_key": "sk-secret",
            "metadata": {"x": 1},
            "litellm_call_id": "id-123",
        }
    )
    assert result == {"seconds": "8"}


def test_requested__timeout_always_removed():
    # timeout is NOT a litellm-internal param, so only the explicit pop removes it.
    result = get_requested({"seconds": "8", "timeout": 30})
    assert result == {"seconds": "8"}


def test_requested__nested_kwargs_merge_and_override_base():
    result = get_requested(
        {"seconds": "8", "kwargs": {"size": "720x1280", "seconds": "override"}}
    )
    # nested kwargs win over the top-level base params on collision.
    assert result == {"seconds": "override", "size": "720x1280"}


def test_requested__non_dict_kwargs_treated_as_empty():
    result = get_requested({"seconds": "8", "kwargs": "not-a-dict"})
    assert result == {"seconds": "8"}


def test_requested__none_input_returns_empty():
    assert get_requested(None) == {}


def test_requested__top_level_extra_body_spread_and_preserved():
    result = get_requested(
        {"seconds": "8", "extra_body": {"vertex_project": "proj", "foo_param": "bar"}}
    )
    # extra_body keys are both spread at top level AND kept under "extra_body".
    assert result == {
        "seconds": "8",
        "vertex_project": "proj",
        "foo_param": "bar",
        "extra_body": {"vertex_project": "proj", "foo_param": "bar"},
    }


def test_requested__extra_body_kwargs_overrides_top_level():
    result = get_requested(
        {
            "extra_body": {"a": "top", "b": "top_b"},
            "kwargs": {"extra_body": {"a": "kw"}},
        }
    )
    # kwargs' extra_body wins over the top-level extra_body on collision; the
    # non-colliding top-level key survives.
    assert result == {
        "a": "kw",
        "b": "top_b",
        "extra_body": {"a": "kw", "b": "top_b"},
    }


def test_requested__extra_body_strips_litellm_internal_params():
    result = get_requested({"extra_body": {"api_key": "sk", "foo_param": "bar"}})
    # api_key filtered out of extra_body; only foo_param remains (and is spread).
    assert result == {"foo_param": "bar", "extra_body": {"foo_param": "bar"}}


def test_requested__empty_extra_body_not_added():
    result = get_requested({"seconds": "8", "extra_body": {}})
    assert result == {"seconds": "8"}
    assert "extra_body" not in result


# =========================================================================== #
# get_optional_params_video_generation
#
# Delegates mapping to the provider config (the seam) then folds extra_body in.
# =========================================================================== #


def _config(map_return):
    config = MagicMock()
    config.map_openai_params.return_value = map_return
    return config


def test_optional__delegates_to_map_openai_params_with_drop_params():
    config = _config({"seconds": "8"})
    optional_params = {"seconds": "8"}

    result = get_optional(
        model="sora-2",
        video_generation_provider_config=config,
        video_generation_optional_params=optional_params,
    )

    assert result == {"seconds": "8"}
    config.map_openai_params.assert_called_once_with(
        video_create_optional_params=optional_params,
        model="sora-2",
        drop_params=litellm.drop_params,
    )


def test_optional__extra_body_overrides_mapped_and_is_removed():
    # mapped output carries a leftover extra_body that must be popped; the input
    # extra_body overrides a colliding mapped key and is spread in.
    config = _config({"seconds": "8", "size": "mapped", "extra_body": {"leftover": 1}})

    result = get_optional(
        model="sora-2",
        video_generation_provider_config=config,
        video_generation_optional_params={
            "extra_body": {"size": "override", "extra": "x"}
        },
    )

    assert result == {"seconds": "8", "size": "override", "extra": "x"}
    assert "extra_body" not in result


def test_optional__non_dict_extra_body_ignored():
    config = _config({"seconds": "8"})

    result = get_optional(
        model="sora-2",
        video_generation_provider_config=config,
        video_generation_optional_params={"seconds": "8", "extra_body": None},
    )

    assert result == {"seconds": "8"}


@pytest.mark.parametrize("extra_body", [
    {"content": [{"type": "image_url", "image_url": {"url": "https://example.com/ref.png"}, "role": "reference_image"}]},
    {"input_reference": "https://example.com/ref.png"},
])
def test_extra_body_overrides_mapped_images_after_dropping_conflicts(extra_body: dict[str, object]) -> None:
    params: Final = get_requested({
        "kwargs": {
            "frame_images": [{"type": "image_url", "image_url": {"url": "https://example.com/first.png"}, "frame_type": "first_frame"}],
            "extra_body": extra_body,
        },
    })
    config: Final = VolcEngineVideoConfig()
    with pytest.raises(UnsupportedParamsError, match="cannot be combined"):
        get_optional("model", config, params)
    assert get_optional("model", config, params, drop_params=True) == {
        "content": [{"type": "image_url", "image_url": {"url": "https://example.com/first.png"}, "role": "first_frame"}],
        **extra_body,
    }