diff --git a/litellm/llms/black_forest_labs/videos/__init__.py b/litellm/llms/black_forest_labs/videos/__init__.py new file mode 100644 index 00000000000..97380ade709 --- /dev/null +++ b/litellm/llms/black_forest_labs/videos/__init__.py @@ -0,0 +1,3 @@ +from .transformation import BlackForestLabsVideoConfig + +__all__ = ["BlackForestLabsVideoConfig"] diff --git a/litellm/llms/black_forest_labs/videos/transformation.py b/litellm/llms/black_forest_labs/videos/transformation.py new file mode 100644 index 00000000000..d715a3f6e7e --- /dev/null +++ b/litellm/llms/black_forest_labs/videos/transformation.py @@ -0,0 +1,462 @@ +""" +Black Forest Labs FLUX 3 Video Configuration + +Handles transformation between OpenAI-compatible video params and the Black +Forest Labs FLUX 3 video API. + +API Reference: https://docs.bfl.ai/api-reference/utility/generate-a-video-with-flux-3 +""" + +import time +from typing import TYPE_CHECKING, Any, Final # noqa: TID251 # BaseVideoConfig types its payloads dict[str, Any] + +import httpx + +import litellm +from litellm.llms.base_llm.videos.transformation import BaseVideoConfig +from litellm.llms.custom_httpx.http_handler import ( + AsyncHTTPHandler, + HTTPHandler, + _get_httpx_client, + get_async_httpx_client, +) +from litellm.secret_managers.main import get_secret_str +from litellm.types.router import GenericLiteLLMParams +from litellm.types.videos.main import VideoObject +from litellm.types.videos.utils import ( + encode_video_id_with_provider, + extract_original_video_id, +) + +from ..common_utils import ( + DEFAULT_API_BASE, + BlackForestLabsError, + assert_bfl_polling_url, +) + +if TYPE_CHECKING: + from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj + + LiteLLMLoggingObj = _LiteLLMLoggingObj +else: + LiteLLMLoggingObj = Any + +VIDEO_MODELS: Final[dict[str, str]] = {"flux-3-video": "/v1/flux-3-video"} + +RESOLUTIONS: Final = ("hd", "fhd") +ASPECT_RATIOS: Final = ("21:9", "2:1", "16:9", "4:3", "1:1", "3:4", "9:16", "auto") +MIN_DURATION: Final = 5 +MAX_DURATION: Final = 20 + +# BFL reports one of these on GET /v1/get_result. Anything outside the terminal +# set is still running. +_TERMINAL_STATUSES: Final[dict[str, str]] = { + "Ready": "completed", + "Error": "failed", + "Content Moderated": "failed", + "Request Moderated": "failed", + "Task not found": "failed", +} +_IN_PROGRESS_STATUSES: Final[dict[str, str]] = { + "Pending": "queued", + "Queued": "queued", + "Reasoning": "in_progress", + "Generating": "in_progress", + "Uploading": "in_progress", +} + + +class BlackForestLabsVideoConfig(BaseVideoConfig): + """ + Configuration for Black Forest Labs FLUX 3 video generation. + + FLUX 3 is a single endpoint with a ``mode`` discriminator: ``t2v`` from a + prompt alone, ``i2v`` from keyframe images, ``v2v`` to continue an existing + clip, and ``draft_enhance`` to re-render a draft at full quality. + + Submission returns a job id plus a regional ``polling_url``. That URL has to + be reused verbatim, because the global host answers 404 for a job dispatched + to a region. + """ + + def get_supported_openai_params(self, model: str) -> list: + return [ + "model", + "seconds", + "size", + "input_reference", + "user", + "extra_headers", + ] + + def map_openai_params( + self, + video_create_optional_params: dict, + model: str, + drop_params: bool, + ) -> dict: + """ + Map OpenAI video params onto FLUX 3 params. + + - ``seconds`` -> ``duration`` (whole seconds, 5 to 20) + - ``size`` -> ``resolution`` tier, by the shorter side + - ``input_reference`` -> a single ``keyframes`` entry, which selects i2v + """ + mapped: dict[str, Any] = {} + + seconds = video_create_optional_params.get("seconds") + if seconds is not None: + duration = self._map_duration(seconds) + if duration is not None: + mapped["duration"] = duration + + size = video_create_optional_params.get("size") + if size is not None: + resolution = self._map_size_to_resolution(size) + if resolution is not None: + mapped["resolution"] = resolution + + input_reference = video_create_optional_params.get("input_reference") + if input_reference is not None: + mapped["keyframes"] = [input_reference] + + supported: Final = self.get_supported_openai_params(model) + mapped.update( + { + key: value + for key, value in video_create_optional_params.items() + if key not in supported and key not in ("seconds", "size", "input_reference") + } + ) + + return mapped + + def _map_duration(self, seconds: object) -> int | None: + if not isinstance(seconds, (int, float, str)): + return None + try: + duration = int(float(seconds)) + except (TypeError, ValueError): + return None + return max(MIN_DURATION, min(MAX_DURATION, duration)) + + def _map_size_to_resolution(self, size: object) -> str | None: + """ + FLUX 3 takes a named tier, not pixel dimensions, so map by the shorter + side: at most 720 is ``hd`` and anything larger is ``fhd``. + """ + if not isinstance(size, str): + return None + if size in RESOLUTIONS: + return size + if "x" not in size.lower(): + return None + try: + width, height = (int(part) for part in size.lower().split("x", 1)) + except ValueError: + return None + return "hd" if min(width, height) <= 720 else "fhd" + + def validate_environment( + self, + headers: dict, + model: str, + api_key: str | None = None, + litellm_params: GenericLiteLLMParams | None = None, + ) -> dict: + if litellm_params and litellm_params.api_key: + api_key = api_key or litellm_params.api_key + + final_api_key: Final = ( + api_key or litellm.api_key or get_secret_str("BFL_API_KEY") or get_secret_str("BLACK_FOREST_LABS_API_KEY") + ) + + if not final_api_key: + raise BlackForestLabsError( + status_code=401, + message="BFL_API_KEY is not set. Please set it via environment variable or pass api_key parameter.", + ) + + headers.update( + { + "x-key": final_api_key, + "Content-Type": "application/json", + "Accept": "application/json", + } + ) + return headers + + def get_complete_url( + self, + model: str, + api_base: str | None, + litellm_params: dict, + ) -> str: + base_url: str = api_base or get_secret_str("BFL_API_BASE") or DEFAULT_API_BASE + return base_url.rstrip("/") + + def _get_model_endpoint(self, model: str) -> str: + model_name = model.lower().split("/")[-1] + if model_name in VIDEO_MODELS: + return VIDEO_MODELS[model_name] + raise ValueError(f"Unknown BFL video model: {model_name}. Supported models: {list(VIDEO_MODELS.keys())}") + + def transform_video_create_request( + self, + model: str, + prompt: str, + api_base: str, + video_create_optional_request_params: dict, + litellm_params: GenericLiteLLMParams, + headers: dict, + ) -> tuple[dict, list, str]: + request_data: dict[str, Any] = {"prompt": prompt} + request_data.update(video_create_optional_request_params) + request_data["mode"] = self._infer_mode(request_data) + + if request_data["mode"] == "draft_enhance": + request_data.pop("prompt", None) + + url: Final = f"{api_base}{self._get_model_endpoint(model)}" + return request_data, [], url + + def _infer_mode(self, request_data: dict) -> str: + """FLUX 3 discriminates on ``mode``; derive it from the inputs given.""" + if request_data.get("mode"): + return str(request_data["mode"]) + if request_data.get("draft_cache"): + return "draft_enhance" + if request_data.get("start_video"): + return "v2v" + if request_data.get("keyframes"): + return "i2v" + return "t2v" + + def transform_video_create_response( + self, + model: str, + raw_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + custom_llm_provider: str | None = None, + request_data: dict | None = None, + ) -> VideoObject: + """ + Submission answers with the job handle, not the finished video: + ``{"id": ..., "polling_url": ..., "cost": null}``. + + The regional polling URL is kept on the video id so status and content + calls hit the same region. + """ + response_data: Final = self._parse_json(raw_response) + + job_id: Final = response_data.get("id") + if not job_id: + raise BlackForestLabsError( + status_code=raw_response.status_code, + message=f"No job id in BFL response: {response_data}", + ) + + polling_url: Final = response_data.get("polling_url") + if polling_url: + assert_bfl_polling_url(polling_url) + + video_obj: Final = VideoObject( + id=job_id, + object="video", + status="queued", + created_at=int(time.time()), + model=model, + ) + + if request_data: + if request_data.get("duration") not in (None, "auto"): + video_obj.seconds = str(request_data["duration"]) + if request_data.get("resolution"): + video_obj.size = str(request_data["resolution"]) + + video_obj._hidden_params = {"polling_url": polling_url} + + if custom_llm_provider: + video_obj.id = encode_video_id_with_provider(job_id, custom_llm_provider, model) + + return video_obj + + def transform_video_status_retrieve_request( + self, + video_id: str, + api_base: str, + litellm_params: GenericLiteLLMParams, + headers: dict, + ) -> tuple[str, dict]: + return self._get_result_url(video_id, api_base), {} + + def transform_video_status_retrieve_response( + self, + raw_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + custom_llm_provider: str | None = None, + ) -> VideoObject: + response_data: Final = self._parse_json(raw_response) + bfl_status: Final = response_data.get("status", "Pending") + + video_obj: Final = VideoObject( + id=response_data.get("id", ""), + object="video", + status=self._map_status(bfl_status), + progress=self._map_progress(response_data.get("progress")), + ) + + if bfl_status in _TERMINAL_STATUSES and _TERMINAL_STATUSES[bfl_status] == "failed": + video_obj.error = { + "code": bfl_status, + "message": str(response_data.get("details") or bfl_status), + } + + cost: Final = response_data.get("cost") + if cost is not None: + video_obj.usage = {"credits": cost} + + if custom_llm_provider and video_obj.id: + video_obj.id = encode_video_id_with_provider(video_obj.id, custom_llm_provider, None) + + return video_obj + + def _map_status(self, bfl_status: str) -> str: + if bfl_status in _TERMINAL_STATUSES: + return _TERMINAL_STATUSES[bfl_status] + return _IN_PROGRESS_STATUSES.get(bfl_status, "in_progress") + + def _map_progress(self, progress: object) -> int | None: + if not isinstance(progress, (int, float, str)): + return None + try: + value = float(progress) + except (TypeError, ValueError): + return None + # BFL reports a 0..1 fraction; VideoObject.progress is a percentage. + return round(value * 100) if value <= 1 else round(value) + + def transform_video_content_request( + self, + video_id: str, + api_base: str, + litellm_params: GenericLiteLLMParams, + headers: dict, + variant: str | None = None, + ) -> tuple[str, dict]: + return self._get_result_url(video_id, api_base), {} + + def transform_video_content_response( + self, + raw_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + ) -> bytes: + video_url: Final = self._extract_video_url(self._parse_json(raw_response)) + httpx_client: Final[HTTPHandler] = _get_httpx_client() + video_response: Final = httpx_client.get(video_url) + video_response.raise_for_status() + return video_response.content + + async def async_transform_video_content_response( + self, + raw_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + ) -> bytes: + video_url: Final = self._extract_video_url(self._parse_json(raw_response)) + async_client: Final[AsyncHTTPHandler] = get_async_httpx_client( + llm_provider=litellm.LlmProviders.BLACK_FOREST_LABS, + ) + video_response: Final = await async_client.get(video_url) + video_response.raise_for_status() + return video_response.content + + def _extract_video_url(self, response_data: dict) -> str: + status: Final = response_data.get("status", "Pending") + result: Final = response_data.get("result") or {} + video_url: Final = result.get("sample") + + if video_url: + return video_url + + if status in _TERMINAL_STATUSES: + raise BlackForestLabsError( + status_code=500, + message=f"Video generation did not produce a video (status: {status}).", + ) + raise BlackForestLabsError( + status_code=409, + message=f"Video is still processing (status: {status}). Please wait and try again.", + ) + + def _get_result_url(self, video_id: str, api_base: str) -> str: + original_video_id: Final = extract_original_video_id(video_id) + return f"{api_base.rstrip('/')}/v1/get_result?id={original_video_id}" + + def _parse_json(self, raw_response: httpx.Response) -> dict: + try: + return raw_response.json() + except Exception as e: + raise BlackForestLabsError( + status_code=raw_response.status_code, + message=f"Error parsing BFL response: {e}", + ) + + def transform_video_remix_request( + self, + video_id: str, + prompt: str, + api_base: str, + litellm_params: GenericLiteLLMParams, + headers: dict, + extra_body: dict[str, Any] | None = None, + ) -> tuple[str, dict]: + raise NotImplementedError("video remix is not supported by the FLUX 3 video API") + + def transform_video_remix_response( + self, + raw_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + custom_llm_provider: str | None = None, + ) -> VideoObject: + raise NotImplementedError("video remix is not supported by the FLUX 3 video API") + + def transform_video_list_request( + self, + api_base: str, + litellm_params: GenericLiteLLMParams, + headers: dict, + after: str | None = None, + limit: int | None = None, + order: str | None = None, + extra_query: dict[str, Any] | None = None, + ) -> tuple[str, dict]: + raise NotImplementedError("video listing is not supported by the FLUX 3 video API") + + def transform_video_list_response( + self, + raw_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + custom_llm_provider: str | None = None, + ) -> dict[str, str]: + raise NotImplementedError("video listing is not supported by the FLUX 3 video API") + + def transform_video_delete_request( + self, + video_id: str, + api_base: str, + litellm_params: GenericLiteLLMParams, + headers: dict, + ) -> tuple[str, dict]: + raise NotImplementedError("video delete is not supported by the FLUX 3 video API") + + def transform_video_delete_response( + self, + raw_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + ) -> VideoObject: + raise NotImplementedError("video delete is not supported by the FLUX 3 video API") + + def get_error_class( + self, error_message: str, status_code: int, headers: dict | httpx.Headers + ) -> BlackForestLabsError: + return BlackForestLabsError(status_code=status_code, message=error_message) diff --git a/litellm/utils.py b/litellm/utils.py index d91d3092624..ae56747ee7c 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -8885,6 +8885,12 @@ class ProviderConfigManager: from litellm.llms.runwayml.videos.transformation import RunwayMLVideoConfig return RunwayMLVideoConfig() + elif LlmProviders.BLACK_FOREST_LABS == provider: + from litellm.llms.black_forest_labs.videos.transformation import ( + BlackForestLabsVideoConfig, + ) + + return BlackForestLabsVideoConfig() return None @staticmethod diff --git a/model_prices_and_context_window.json b/model_prices_and_context_window.json index e6c6cab0631..665bb60a6be 100644 --- a/model_prices_and_context_window.json +++ b/model_prices_and_context_window.json @@ -11872,6 +11872,23 @@ "/v1/images/generations" ] }, + "black_forest_labs/flux-3-video": { + "litellm_provider": "black_forest_labs", + "mode": "video_generation", + "output_cost_per_video_per_second": 0.17, + "source": "https://docs.bfl.ai/quick_start/pricing", + "supported_modalities": [ + "text", + "image", + "video" + ], + "supported_output_modalities": [ + "video" + ], + "metadata": { + "comment": "Per-second pricing varies by mode and resolution: t2v and i2v are $0.17/s hd and $0.29/s fhd, v2v is $0.43/s hd and $0.54/s fhd, drafts are $0.06/s (t2v, i2v) and $0.12/s (v2v). The t2v hd rate is used here." + } + }, "cerebras/llama-3.3-70b": { "input_cost_per_token": 8.5e-07, "litellm_provider": "cerebras", diff --git a/tests/test_litellm/llms/black_forest_labs/videos/__init__.py b/tests/test_litellm/llms/black_forest_labs/videos/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/test_litellm/llms/black_forest_labs/videos/test_bfl_video_transformation.py b/tests/test_litellm/llms/black_forest_labs/videos/test_bfl_video_transformation.py new file mode 100644 index 00000000000..8e6a4457952 --- /dev/null +++ b/tests/test_litellm/llms/black_forest_labs/videos/test_bfl_video_transformation.py @@ -0,0 +1,310 @@ +""" +Tests for Black Forest Labs FLUX 3 video generation transformation. + +Payload and response shapes are taken from a live FLUX 3 video generation +against https://api.bfl.ai/v1/flux-3-video and its regional polling URL. +""" + +from unittest.mock import Mock + +import httpx +import pytest + +from litellm.llms.black_forest_labs.common_utils import BlackForestLabsError +from litellm.llms.black_forest_labs.videos.transformation import ( + BlackForestLabsVideoConfig, +) +from litellm.types.router import GenericLiteLLMParams +from litellm.types.videos.utils import extract_original_video_id + +API_BASE = "https://api.bfl.ai" +JOB_ID = "90307d3a-deec-47cb-bdb9-bf1c5bae1f04" +POLLING_URL = f"https://api.us7.bfl.ai/v1/get_result?id={JOB_ID}" +SAMPLE_URL = "https://delivery.us7.bfl.ai/durable/2026081720/video.mp4?se=2026-08-17T21%3A26%3A38Z&sig=abc" + + +def _response(payload: dict, status_code: int = 200) -> httpx.Response: + return httpx.Response( + status_code=status_code, + json=payload, + request=httpx.Request("POST", f"{API_BASE}/v1/flux-3-video"), + ) + + +class TestBlackForestLabsVideoTransformation: + def setup_method(self): + self.config = BlackForestLabsVideoConfig() + self.mock_logging_obj = Mock() + + def test_text_to_video_request_sets_t2v_mode(self): + data, files, url = self.config.transform_video_create_request( + model="flux-3-video", + prompt="A white kitten chases a butterfly across a sunlit garden.", + api_base=API_BASE, + video_create_optional_request_params={ + "duration": 8, + "resolution": "fhd", + "aspect_ratio": "16:9", + }, + litellm_params=GenericLiteLLMParams(), + headers={}, + ) + + assert url == "https://api.bfl.ai/v1/flux-3-video" + assert data["mode"] == "t2v" + assert data["prompt"] == "A white kitten chases a butterfly across a sunlit garden." + assert data["duration"] == 8 + assert data["resolution"] == "fhd" + assert files == [] + + @pytest.mark.parametrize( + "params, expected_mode", + [ + ({}, "t2v"), + ({"keyframes": ["https://example.com/first.png"]}, "i2v"), + ({"start_video": "https://example.com/clip.mp4"}, "v2v"), + ({"draft_cache": "https://example.com/bundle.bin"}, "draft_enhance"), + ], + ) + def test_mode_is_inferred_from_inputs(self, params, expected_mode): + data, _, _ = self.config.transform_video_create_request( + model="flux-3-video", + prompt="a prompt", + api_base=API_BASE, + video_create_optional_request_params=dict(params), + litellm_params=GenericLiteLLMParams(), + headers={}, + ) + + assert data["mode"] == expected_mode + + def test_draft_enhance_drops_the_prompt(self): + """FLUX 3 rejects a prompt in draft_enhance mode; the bundle carries it.""" + data, _, _ = self.config.transform_video_create_request( + model="flux-3-video", + prompt="a prompt the API would reject here", + api_base=API_BASE, + video_create_optional_request_params={"draft_cache": "bundle"}, + litellm_params=GenericLiteLLMParams(), + headers={}, + ) + + assert data["mode"] == "draft_enhance" + assert "prompt" not in data + + def test_explicit_mode_is_not_overridden(self): + data, _, _ = self.config.transform_video_create_request( + model="flux-3-video", + prompt="a prompt", + api_base=API_BASE, + video_create_optional_request_params={ + "mode": "i2v", + "keyframes": ["https://example.com/first.png"], + }, + litellm_params=GenericLiteLLMParams(), + headers={}, + ) + + assert data["mode"] == "i2v" + + def test_unknown_model_is_rejected(self): + with pytest.raises(ValueError, match="Unknown BFL video model"): + self.config.transform_video_create_request( + model="flux-9-video", + prompt="a prompt", + api_base=API_BASE, + video_create_optional_request_params={}, + litellm_params=GenericLiteLLMParams(), + headers={}, + ) + + @pytest.mark.parametrize( + "size, expected_resolution", + [ + ("1280x720", "hd"), + ("1920x1080", "fhd"), + ("1080x1920", "fhd"), + ("720x1280", "hd"), + ("fhd", "fhd"), + ], + ) + def test_size_maps_to_a_resolution_tier(self, size, expected_resolution): + mapped = self.config.map_openai_params( + video_create_optional_params={"size": size}, + model="flux-3-video", + drop_params=False, + ) + + assert mapped["resolution"] == expected_resolution + + @pytest.mark.parametrize( + "seconds, expected_duration", + [("8", 8), (8, 8), ("2", 5), ("60", 20), ("8.6", 8)], + ) + def test_seconds_maps_into_the_supported_duration_range(self, seconds, expected_duration): + mapped = self.config.map_openai_params( + video_create_optional_params={"seconds": seconds}, + model="flux-3-video", + drop_params=False, + ) + + assert mapped["duration"] == expected_duration + + def test_input_reference_becomes_a_keyframe(self): + mapped = self.config.map_openai_params( + video_create_optional_params={"input_reference": "https://example.com/first.png"}, + model="flux-3-video", + drop_params=False, + ) + + assert mapped["keyframes"] == ["https://example.com/first.png"] + + def test_create_response_returns_the_job_handle_and_keeps_the_polling_url(self): + video = self.config.transform_video_create_response( + model="flux-3-video", + raw_response=_response({"id": JOB_ID, "polling_url": POLLING_URL, "cost": None}), + logging_obj=self.mock_logging_obj, + custom_llm_provider="black_forest_labs", + request_data={"duration": 8, "resolution": "fhd"}, + ) + + assert extract_original_video_id(video.id) == JOB_ID + assert video.status == "queued" + assert video.seconds == "8" + assert video.size == "fhd" + assert video._hidden_params["polling_url"] == POLLING_URL + + def test_create_response_without_a_job_id_raises(self): + with pytest.raises(BlackForestLabsError): + self.config.transform_video_create_response( + model="flux-3-video", + raw_response=_response({"detail": "bad request"}, status_code=422), + logging_obj=self.mock_logging_obj, + ) + + def test_create_response_rejects_a_polling_url_outside_bfl(self): + with pytest.raises(BlackForestLabsError, match="not within the bfl.ai domain"): + self.config.transform_video_create_response( + model="flux-3-video", + raw_response=_response( + {"id": JOB_ID, "polling_url": "https://attacker.example.com/v1/get_result"} + ), + logging_obj=self.mock_logging_obj, + ) + + @pytest.mark.parametrize( + "bfl_status, expected_status", + [ + ("Pending", "queued"), + ("Queued", "queued"), + ("Reasoning", "in_progress"), + ("Generating", "in_progress"), + ("Ready", "completed"), + ("Error", "failed"), + ("Content Moderated", "failed"), + ("Task not found", "failed"), + ], + ) + def test_bfl_status_maps_to_openai_status(self, bfl_status, expected_status): + video = self.config.transform_video_status_retrieve_response( + raw_response=_response({"id": JOB_ID, "status": bfl_status, "result": {}}), + logging_obj=self.mock_logging_obj, + ) + + assert video.status == expected_status + + def test_failed_status_carries_the_bfl_detail(self): + video = self.config.transform_video_status_retrieve_response( + raw_response=_response( + {"id": JOB_ID, "status": "Content Moderated", "details": "flagged by moderation"} + ), + logging_obj=self.mock_logging_obj, + ) + + assert video.status == "failed" + assert video.error["code"] == "Content Moderated" + assert video.error["message"] == "flagged by moderation" + + def test_completed_status_has_no_error(self): + video = self.config.transform_video_status_retrieve_response( + raw_response=_response({"id": JOB_ID, "status": "Ready", "result": {"sample": SAMPLE_URL}}), + logging_obj=self.mock_logging_obj, + ) + + assert video.error is None + + def test_credit_cost_is_reported_as_usage(self): + video = self.config.transform_video_status_retrieve_response( + raw_response=_response({"id": JOB_ID, "status": "Ready", "cost": 30.0}), + logging_obj=self.mock_logging_obj, + ) + + assert video.usage == {"credits": 30.0} + + def test_fractional_progress_is_reported_as_a_percentage(self): + video = self.config.transform_video_status_retrieve_response( + raw_response=_response({"id": JOB_ID, "status": "Generating", "progress": 0.42}), + logging_obj=self.mock_logging_obj, + ) + + assert video.progress == 42 + + def test_status_request_targets_get_result_with_the_original_job_id(self): + encoded_id = self.config.transform_video_create_response( + model="flux-3-video", + raw_response=_response({"id": JOB_ID, "polling_url": POLLING_URL}), + logging_obj=self.mock_logging_obj, + custom_llm_provider="black_forest_labs", + ).id + + url, params = self.config.transform_video_status_retrieve_request( + video_id=encoded_id, + api_base="https://api.us7.bfl.ai", + litellm_params=GenericLiteLLMParams(), + headers={}, + ) + + assert url == POLLING_URL + assert params == {} + + def test_content_request_targets_get_result(self): + url, _ = self.config.transform_video_content_request( + video_id=JOB_ID, + api_base="https://api.us7.bfl.ai", + litellm_params=GenericLiteLLMParams(), + headers={}, + ) + + assert url == POLLING_URL + + def test_content_response_raises_while_still_generating(self): + with pytest.raises(BlackForestLabsError, match="still processing"): + self.config.transform_video_content_response( + raw_response=_response({"id": JOB_ID, "status": "Generating", "result": {}}), + logging_obj=self.mock_logging_obj, + ) + + def test_content_response_raises_when_a_terminal_job_has_no_video(self): + with pytest.raises(BlackForestLabsError, match="did not produce a video"): + self.config.transform_video_content_response( + raw_response=_response({"id": JOB_ID, "status": "Error", "result": {}}), + logging_obj=self.mock_logging_obj, + ) + + def test_validate_environment_sets_the_x_key_header(self): + headers = self.config.validate_environment( + headers={}, + model="flux-3-video", + api_key="test-key", + ) + + assert headers["x-key"] == "test-key" + assert headers["Content-Type"] == "application/json" + + def test_validate_environment_without_a_key_raises(self, monkeypatch): + monkeypatch.delenv("BFL_API_KEY", raising=False) + monkeypatch.delenv("BLACK_FOREST_LABS_API_KEY", raising=False) + monkeypatch.setattr("litellm.api_key", None) + + with pytest.raises(BlackForestLabsError, match="BFL_API_KEY is not set"): + self.config.validate_environment(headers={}, model="flux-3-video")