mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-24 00:52:24 +00:00
fix(fal_ai): reuse the status client and headers on the video result probe (#42511)
* fix(fal_ai): reuse the status client and headers on the video result probe The Fal result GET issued after a COMPLETED status poll built its own default client and only carried Authorization and Content-Type, so an injected client, a request-level ssl_verify and extra_headers were honored on the status GET but not on the result GET, and a transport failure on that probe escaped as a 500. The handler now hands the selected sync or async client to the provider status transform, Fal reuses it with the full validated header set, and a probe transport error stays non-terminal like the existing 429 and 5xx handling Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(fal_ai): mark the result probe header dict as mutable-ok for the type discipline gate Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(fal_ai): move the result probe repro to tests/integration Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(e2e): restore the shared e2e helpers to the merge base Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: kerry <kerry@berri.ai> Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot]
GitHub
parent
5d3b31fb02
commit
153e5ed185
12 changed files with 391 additions and 36 deletions
|
|
@ -12,6 +12,7 @@ from litellm.types.videos.main import VideoCreateOptionalRequestParams
|
|||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
|
||||
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
|
||||
from litellm.types.videos.main import CharacterObject as _CharacterObject
|
||||
from litellm.types.videos.main import VideoObject as _VideoObject
|
||||
|
||||
|
|
@ -269,6 +270,7 @@ class BaseVideoConfig(ABC):
|
|||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
custom_llm_provider: str | None = None,
|
||||
client: "HTTPHandler | None" = None,
|
||||
) -> VideoObject:
|
||||
pass
|
||||
|
||||
|
|
@ -277,6 +279,7 @@ class BaseVideoConfig(ABC):
|
|||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
custom_llm_provider: str | None = None,
|
||||
client: "AsyncHTTPHandler | None" = None,
|
||||
) -> VideoObject:
|
||||
"""Async transform video status retrieve response."""
|
||||
return self.transform_video_status_retrieve_response(
|
||||
|
|
|
|||
|
|
@ -8812,6 +8812,7 @@ class BaseLLMHTTPHandler:
|
|||
raw_response=response,
|
||||
logging_obj=logging_obj,
|
||||
custom_llm_provider=custom_llm_provider,
|
||||
client=sync_httpx_client,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
|
|
@ -8901,6 +8902,7 @@ class BaseLLMHTTPHandler:
|
|||
raw_response=response,
|
||||
logging_obj=logging_obj,
|
||||
custom_llm_provider=custom_llm_provider,
|
||||
client=async_httpx_client,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ from ..common_utils import EdenAIException, authorized_headers, endpoint_url, re
|
|||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
||||
from litellm.llms.custom_httpx.http_handler import HTTPHandler
|
||||
|
||||
|
||||
def _usage_with_reported_cost(
|
||||
|
|
@ -110,6 +111,7 @@ class EdenAIVideoConfig(OpenAIVideoConfig):
|
|||
raw_response: httpx.Response,
|
||||
logging_obj: "LiteLLMLoggingObj",
|
||||
custom_llm_provider: str | None = None,
|
||||
client: "HTTPHandler | None" = None,
|
||||
) -> VideoObject:
|
||||
raw_response.raise_for_status() # the shared GET helpers return error bodies instead of raising
|
||||
video: Final = super().transform_video_status_retrieve_response(
|
||||
|
|
|
|||
|
|
@ -230,6 +230,9 @@ def _response_string(response_data: Mapping[str, object], key: str, default: str
|
|||
return value if isinstance(value, str) else default
|
||||
|
||||
|
||||
_RESULT_HEADERS_NOT_FORWARDED: Final[frozenset[str]] = frozenset({"host", "content-length", "transfer-encoding"})
|
||||
|
||||
|
||||
def _result_request(
|
||||
raw_response: httpx.Response,
|
||||
response_data: Mapping[str, object],
|
||||
|
|
@ -237,14 +240,12 @@ def _result_request(
|
|||
if _response_string(response_data, "status", "IN_QUEUE") != "COMPLETED":
|
||||
return None
|
||||
result_url: Final[str] = str(raw_response.request.url).removesuffix("/status")
|
||||
encoding: Final[str] = raw_response.request.headers.encoding
|
||||
result_headers: Final[Mapping[str, str]] = MappingProxyType(
|
||||
{
|
||||
key: value
|
||||
for key, value in (
|
||||
("Authorization", raw_response.request.headers.get("Authorization")),
|
||||
("Content-Type", raw_response.request.headers.get("Content-Type")),
|
||||
)
|
||||
if value is not None
|
||||
key.decode(encoding): value.decode(encoding)
|
||||
for key, value in raw_response.request.headers.raw
|
||||
if key.decode(encoding).lower() not in _RESULT_HEADERS_NOT_FORWARDED
|
||||
}
|
||||
)
|
||||
return result_url, result_headers
|
||||
|
|
@ -463,9 +464,10 @@ class FalAIVideoConfig(BaseVideoConfig):
|
|||
raw_response: httpx.Response,
|
||||
logging_obj: object,
|
||||
custom_llm_provider: str | None = None,
|
||||
client: HTTPHandler | None = None,
|
||||
) -> VideoObject:
|
||||
response_data: Final[Mapping[str, object]] = _response_data(raw_response)
|
||||
result_error: Final[str | None] = self._fetch_result_error(raw_response, response_data)
|
||||
result_error: Final[str | None] = self._fetch_result_error(raw_response, response_data, client)
|
||||
return _status_video_object(
|
||||
response_data=response_data,
|
||||
raw_response=raw_response,
|
||||
|
|
@ -477,15 +479,20 @@ class FalAIVideoConfig(BaseVideoConfig):
|
|||
self,
|
||||
raw_response: httpx.Response,
|
||||
response_data: Mapping[str, object],
|
||||
client: HTTPHandler | None,
|
||||
) -> str | None:
|
||||
result_request: Final[tuple[str, Mapping[str, str]] | None] = _result_request(raw_response, response_data)
|
||||
if result_request is None:
|
||||
return None
|
||||
result_url, result_headers = result_request
|
||||
result_response: Final[httpx.Response] = self._sync_client_factory().get(
|
||||
url=result_url,
|
||||
headers=result_headers,
|
||||
)
|
||||
result_client: Final[HTTPHandler] = client if client is not None else self._sync_client_factory()
|
||||
try:
|
||||
result_response: Final[httpx.Response] = result_client.get(
|
||||
url=result_url,
|
||||
headers=dict(result_headers), # mutable-ok: HTTPHandler.get only accepts a dict
|
||||
)
|
||||
except httpx.TransportError:
|
||||
return None
|
||||
return _terminal_result_error(result_response)
|
||||
|
||||
async def async_transform_video_status_retrieve_response(
|
||||
|
|
@ -493,9 +500,10 @@ class FalAIVideoConfig(BaseVideoConfig):
|
|||
raw_response: httpx.Response,
|
||||
logging_obj: object,
|
||||
custom_llm_provider: str | None = None,
|
||||
client: AsyncHTTPHandler | None = None,
|
||||
) -> VideoObject:
|
||||
response_data: Final[Mapping[str, object]] = _response_data(raw_response)
|
||||
result_error: Final[str | None] = await self._fetch_result_error_async(raw_response, response_data)
|
||||
result_error: Final[str | None] = await self._fetch_result_error_async(raw_response, response_data, client)
|
||||
return _status_video_object(
|
||||
response_data=response_data,
|
||||
raw_response=raw_response,
|
||||
|
|
@ -507,15 +515,20 @@ class FalAIVideoConfig(BaseVideoConfig):
|
|||
self,
|
||||
raw_response: httpx.Response,
|
||||
response_data: Mapping[str, object],
|
||||
client: AsyncHTTPHandler | None,
|
||||
) -> str | None:
|
||||
result_request: Final[tuple[str, Mapping[str, str]] | None] = _result_request(raw_response, response_data)
|
||||
if result_request is None:
|
||||
return None
|
||||
result_url, result_headers = result_request
|
||||
result_response: Final[httpx.Response] = await self._async_client_factory().get(
|
||||
url=result_url,
|
||||
headers=result_headers,
|
||||
)
|
||||
result_client: Final[AsyncHTTPHandler] = client if client is not None else self._async_client_factory()
|
||||
try:
|
||||
result_response: Final[httpx.Response] = await result_client.get(
|
||||
url=result_url,
|
||||
headers=dict(result_headers), # mutable-ok: AsyncHTTPHandler.get only accepts a dict
|
||||
)
|
||||
except httpx.TransportError:
|
||||
return None
|
||||
return _terminal_result_error(result_response)
|
||||
|
||||
@staticmethod
|
||||
|
|
|
|||
|
|
@ -26,6 +26,7 @@ from litellm.types.videos.utils import (
|
|||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
|
||||
from litellm.llms.custom_httpx.http_handler import HTTPHandler
|
||||
|
||||
from ...base_llm.chat.transformation import BaseLLMException as _BaseLLMException
|
||||
|
||||
|
|
@ -386,6 +387,7 @@ class GeminiVideoConfig(BaseVideoConfig):
|
|||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
custom_llm_provider: str | None = None,
|
||||
client: "HTTPHandler | None" = None,
|
||||
) -> VideoObject:
|
||||
"""
|
||||
Transform the Veo operation status response.
|
||||
|
|
|
|||
|
|
@ -28,6 +28,7 @@ from litellm.types.videos.utils import (
|
|||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
|
||||
from litellm.llms.custom_httpx.http_handler import HTTPHandler
|
||||
|
||||
from ...base_llm.chat.transformation import BaseLLMException as _BaseLLMException
|
||||
|
||||
|
|
@ -437,6 +438,7 @@ class OpenAIVideoConfig(BaseVideoConfig):
|
|||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
custom_llm_provider: str | None = None,
|
||||
client: "HTTPHandler | None" = None,
|
||||
) -> VideoObject:
|
||||
"""
|
||||
Transform the OpenAI video retrieve response.
|
||||
|
|
|
|||
|
|
@ -616,6 +616,7 @@ class RunwayMLVideoConfig(BaseVideoConfig):
|
|||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
custom_llm_provider: str | None = None,
|
||||
client: HTTPHandler | None = None,
|
||||
) -> VideoObject:
|
||||
"""
|
||||
Transform the RunwayML video status retrieve response.
|
||||
|
|
|
|||
|
|
@ -36,6 +36,7 @@ if TYPE_CHECKING:
|
|||
from litellm.llms.base_llm.chat.transformation import (
|
||||
BaseLLMException as _BaseLLMException,
|
||||
)
|
||||
from litellm.llms.custom_httpx.http_handler import HTTPHandler
|
||||
|
||||
LiteLLMLoggingObj = _LiteLLMLoggingObj
|
||||
BaseLLMException = _BaseLLMException
|
||||
|
|
@ -491,6 +492,7 @@ class VertexAIVideoConfig(BaseVideoConfig, VertexBase):
|
|||
raw_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
custom_llm_provider: str | None = None,
|
||||
client: "HTTPHandler | None" = None,
|
||||
) -> VideoObject:
|
||||
"""
|
||||
Transform the Veo operation status response.
|
||||
|
|
|
|||
|
|
@ -1,7 +1,8 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import ssl
|
||||
import threading
|
||||
from collections.abc import Callable, Iterator, Mapping
|
||||
from collections.abc import Callable, Generator, Mapping
|
||||
from contextlib import contextmanager
|
||||
from dataclasses import dataclass
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
|
|
@ -38,7 +39,7 @@ class Wire:
|
|||
|
||||
|
||||
@contextmanager
|
||||
def wire_server(respond: Callable[[Request], Reply]) -> Iterator[Wire]:
|
||||
def wire_server(respond: Callable[[Request], Reply], tls: ssl.SSLContext | None = None) -> Generator[Wire, None, None]:
|
||||
"""Owned TCP peer; requests traverse the real HTTP client and serialization."""
|
||||
received: Final[SimpleQueue[Request]] = SimpleQueue()
|
||||
errors: Final[SimpleQueue[Exception]] = SimpleQueue()
|
||||
|
|
@ -50,7 +51,8 @@ def wire_server(respond: Callable[[Request], Reply]) -> Iterator[Wire]:
|
|||
|
||||
def respond(self) -> None:
|
||||
request: Final = Request(
|
||||
self.command, self.path,
|
||||
self.command,
|
||||
self.path,
|
||||
{name.lower(): value for name, value in self.headers.items()},
|
||||
self.rfile.read(int(self.headers.get("content-length", "0"))),
|
||||
)
|
||||
|
|
@ -99,11 +101,20 @@ def wire_server(respond: Callable[[Request], Reply]) -> Iterator[Wire]:
|
|||
class OwnedHTTPServer(ThreadingHTTPServer):
|
||||
daemon_threads = False
|
||||
|
||||
def server_bind(self) -> None:
|
||||
super().server_bind()
|
||||
if tls is not None:
|
||||
self.socket = tls.wrap_socket(self.socket, server_side=True)
|
||||
|
||||
with OwnedHTTPServer(("127.0.0.1", 0), Handler) as server:
|
||||
thread: Final = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.05})
|
||||
thread.start()
|
||||
try:
|
||||
yield Wire(f"http://127.0.0.1:{server.server_port}", received, disconnected)
|
||||
yield Wire(
|
||||
f"{'https' if tls is not None else 'http'}://127.0.0.1:{server.server_port}",
|
||||
received,
|
||||
disconnected,
|
||||
)
|
||||
finally:
|
||||
server.shutdown()
|
||||
thread.join(timeout=6)
|
||||
|
|
|
|||
|
|
@ -220,6 +220,15 @@
|
|||
"tests/integration/providers/test_fal_ai_video_wire.py::test_fal_h3_video_create_uses_canonical_body_and_status_path": [
|
||||
"other.provider_wire.fal_ai.video_queue_create_status_and_content_download"
|
||||
],
|
||||
"tests/integration/providers/test_fal_ai_video_wire.py::test_fal_result_probe_carries_the_deployment_extra_headers": [
|
||||
"other.provider_wire.fal_ai.video_result_probe_forwards_extra_headers"
|
||||
],
|
||||
"tests/integration/providers/test_fal_ai_video_wire.py::test_fal_result_probe_reuses_the_ssl_verify_false_client": [
|
||||
"other.provider_wire.fal_ai.video_result_probe_honors_ssl_verify"
|
||||
],
|
||||
"tests/integration/providers/test_fal_ai_video_wire.py::test_fal_provider_hanging_up_on_the_result_probe_keeps_the_completed_status": [
|
||||
"other.provider_wire.fal_ai.video_result_probe_hangup_stays_completed"
|
||||
],
|
||||
"tests/integration/mcp/test_mcp_lifecycle.py::test_saved_headers_reach_real_mcp_tool_and_survive_unrelated_edit": [
|
||||
"mcp.call_tool.saved_headers.reach_actual_transport"
|
||||
],
|
||||
|
|
|
|||
|
|
@ -1,9 +1,17 @@
|
|||
import datetime
|
||||
import ipaddress
|
||||
import json
|
||||
import ssl
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from integration._support.client import Gateway
|
||||
from cryptography import x509
|
||||
from cryptography.hazmat.primitives import hashes, serialization
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
from cryptography.x509.oid import NameOID
|
||||
from integration._support.client import JSON_OBJECT, Gateway, object_value, string_value
|
||||
from integration._support.wire import Reply, Request, wire_server
|
||||
|
||||
_MODEL: Final = "bytedance/seedance-2.5/text-to-video"
|
||||
|
|
@ -16,6 +24,37 @@ def _h3_queue_reply(request_id: str) -> Reply:
|
|||
return Reply(body=json.dumps({"status": "IN_QUEUE", "request_id": request_id, "queue_position": 0}).encode())
|
||||
|
||||
|
||||
def _write_self_signed_cert(cert_dir: Path) -> tuple[Path, Path]:
|
||||
key: Final = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||||
now: Final = datetime.datetime.now(datetime.timezone.utc)
|
||||
name: Final = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")])
|
||||
cert: Final = (
|
||||
x509.CertificateBuilder()
|
||||
.subject_name(name)
|
||||
.issuer_name(name)
|
||||
.public_key(key.public_key())
|
||||
.serial_number(x509.random_serial_number())
|
||||
.not_valid_before(now - datetime.timedelta(days=1))
|
||||
.not_valid_after(now + datetime.timedelta(days=7))
|
||||
.add_extension(
|
||||
x509.SubjectAlternativeName([x509.DNSName("localhost"), x509.IPAddress(ipaddress.ip_address("127.0.0.1"))]),
|
||||
critical=False,
|
||||
)
|
||||
.sign(key, hashes.SHA256())
|
||||
)
|
||||
cert_file: Final = cert_dir / "cert.pem"
|
||||
key_file: Final = cert_dir / "key.pem"
|
||||
cert_file.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
|
||||
key_file.write_bytes(
|
||||
key.private_bytes(
|
||||
serialization.Encoding.PEM,
|
||||
serialization.PrivateFormat.TraditionalOpenSSL,
|
||||
serialization.NoEncryption(),
|
||||
)
|
||||
)
|
||||
return cert_file, key_file
|
||||
|
||||
|
||||
@pytest.mark.covers("other.provider_wire.fal_ai.video_queue_create_status_and_content_download")
|
||||
def test_fal_video_create_status_and_content_follow_queue_wire_contract(gateway: Gateway) -> None:
|
||||
request_id: Final = "fal-req-" + uuid.uuid4().hex
|
||||
|
|
@ -178,12 +217,164 @@ def test_fal_video_failed_result_reports_failed_status_and_fal_error(gateway: Ga
|
|||
video_id: Final = created["id"]
|
||||
status: Final = gateway.get(f"/v1/videos/{video_id}")
|
||||
assert status["status"] == "failed"
|
||||
assert "input.reference_image_urls: Failed to download the file" in status["error"]["message"]
|
||||
assert "input.reference_image_urls: Failed to download the file" in string_value(
|
||||
object_value(status["error"])["message"]
|
||||
)
|
||||
content: Final = gateway.request("GET", f"/v1/videos/{video_id}/content")
|
||||
assert content.status_code == 422, content.text
|
||||
assert "Failed to download the file" in content.text
|
||||
|
||||
|
||||
@pytest.mark.covers("other.provider_wire.fal_ai.video_result_probe_forwards_extra_headers")
|
||||
def test_fal_result_probe_carries_the_deployment_extra_headers(gateway: Gateway) -> None:
|
||||
request_id: Final = "fal-probe-req-" + uuid.uuid4().hex
|
||||
marker: Final = uuid.uuid4().hex
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
assert request.headers["authorization"] == "Key synthetic-fal-key"
|
||||
if request.method == "POST":
|
||||
assert request.target == f"/{_H3_MODEL}"
|
||||
return Reply(
|
||||
body=json.dumps({"status": "IN_QUEUE", "request_id": request_id, "queue_position": 0}).encode()
|
||||
)
|
||||
assert request.method == "GET"
|
||||
if request.target == f"/minimax/h3/requests/{request_id}/status":
|
||||
return Reply(body=json.dumps({"status": "COMPLETED", "request_id": request_id}).encode())
|
||||
assert request.target == f"/minimax/h3/requests/{request_id}"
|
||||
if request.headers.get("x-integration-routing") != marker:
|
||||
return Reply(status=403, body=json.dumps({"detail": "routing header missing"}).encode())
|
||||
return Reply(body=json.dumps({"video": {"url": f"{wire_url}/files/{request_id}.mp4"}}).encode())
|
||||
|
||||
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
||||
wire_url: Final = wire.url
|
||||
model: Final = scenario.model(
|
||||
model=f"fal_ai/{_H3_MODEL}",
|
||||
api_base=wire.url,
|
||||
api_key="synthetic-fal-key",
|
||||
extra_headers={"x-integration-routing": marker},
|
||||
)
|
||||
created: Final = gateway.post(
|
||||
"/v1/videos",
|
||||
{
|
||||
"model": model,
|
||||
"prompt": "a paper boat drifting across a puddle after rain",
|
||||
"seconds": 6,
|
||||
"size": "2k",
|
||||
},
|
||||
)
|
||||
assert created["status"] == "queued"
|
||||
video_id: Final = created["id"]
|
||||
response: Final = gateway.request("GET", f"/v1/videos/{video_id}")
|
||||
assert response.status_code == 200, response.text
|
||||
status: Final = JSON_OBJECT.validate_json(response.content)
|
||||
assert status["status"] == "completed", status
|
||||
assert status["error"] is None, status
|
||||
assert [
|
||||
(request.method, request.target, request.headers.get("x-integration-routing")) for request in wire.drain()
|
||||
] == [
|
||||
("POST", f"/{_H3_MODEL}", marker),
|
||||
("GET", f"/minimax/h3/requests/{request_id}/status", marker),
|
||||
("GET", f"/minimax/h3/requests/{request_id}", marker),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.covers("other.provider_wire.fal_ai.video_result_probe_honors_ssl_verify")
|
||||
def test_fal_result_probe_reuses_the_ssl_verify_false_client(gateway: Gateway, tmp_path: Path) -> None:
|
||||
request_id: Final = "fal-tls-req-" + uuid.uuid4().hex
|
||||
cert_file, key_file = _write_self_signed_cert(tmp_path)
|
||||
context: Final = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
||||
context.load_cert_chain(certfile=cert_file, keyfile=key_file)
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
assert request.headers["authorization"] == "Key synthetic-fal-key"
|
||||
if request.method == "POST":
|
||||
assert request.target == f"/{_H3_MODEL}"
|
||||
return Reply(
|
||||
body=json.dumps({"status": "IN_QUEUE", "request_id": request_id, "queue_position": 0}).encode()
|
||||
)
|
||||
assert request.method == "GET"
|
||||
if request.target == f"/minimax/h3/requests/{request_id}/status":
|
||||
return Reply(body=json.dumps({"status": "COMPLETED", "request_id": request_id}).encode())
|
||||
assert request.target == f"/minimax/h3/requests/{request_id}"
|
||||
return Reply(body=json.dumps({"video": {"url": f"{wire_url}/files/{request_id}.mp4"}}).encode())
|
||||
|
||||
with wire_server(respond, tls=context) as wire, gateway.scenario() as scenario:
|
||||
wire_url: Final = wire.url
|
||||
model: Final = scenario.model(
|
||||
model=f"fal_ai/{_H3_MODEL}",
|
||||
api_base=wire.url,
|
||||
api_key="synthetic-fal-key",
|
||||
ssl_verify=False,
|
||||
)
|
||||
created: Final = gateway.post(
|
||||
"/v1/videos",
|
||||
{
|
||||
"model": model,
|
||||
"prompt": "a paper boat drifting across a puddle after rain",
|
||||
"seconds": 6,
|
||||
"size": "2k",
|
||||
},
|
||||
)
|
||||
assert created["status"] == "queued"
|
||||
video_id: Final = created["id"]
|
||||
response: Final = gateway.request("GET", f"/v1/videos/{video_id}")
|
||||
assert response.status_code == 200, response.text
|
||||
status: Final = JSON_OBJECT.validate_json(response.content)
|
||||
assert status["status"] == "completed", status
|
||||
assert status["error"] is None, status
|
||||
assert [(request.method, request.target) for request in wire.drain()] == [
|
||||
("POST", f"/{_H3_MODEL}"),
|
||||
("GET", f"/minimax/h3/requests/{request_id}/status"),
|
||||
("GET", f"/minimax/h3/requests/{request_id}"),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.covers("other.provider_wire.fal_ai.video_result_probe_hangup_stays_completed")
|
||||
def test_fal_provider_hanging_up_on_the_result_probe_keeps_the_completed_status(gateway: Gateway) -> None:
|
||||
request_id: Final = "fal-hangup-req-" + uuid.uuid4().hex
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
assert request.headers["authorization"] == "Key synthetic-fal-key"
|
||||
if request.method == "POST":
|
||||
assert request.target == f"/{_H3_MODEL}"
|
||||
return Reply(
|
||||
body=json.dumps({"status": "IN_QUEUE", "request_id": request_id, "queue_position": 0}).encode()
|
||||
)
|
||||
assert request.method == "GET"
|
||||
if request.target == f"/minimax/h3/requests/{request_id}/status":
|
||||
return Reply(body=json.dumps({"status": "COMPLETED", "request_id": request_id}).encode())
|
||||
assert request.target == f"/minimax/h3/requests/{request_id}"
|
||||
return Reply(chunks=(b"{",), abort_after=0)
|
||||
|
||||
with wire_server(respond) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(
|
||||
model=f"fal_ai/{_H3_MODEL}",
|
||||
api_base=wire.url,
|
||||
api_key="synthetic-fal-key",
|
||||
)
|
||||
created: Final = gateway.post(
|
||||
"/v1/videos",
|
||||
{
|
||||
"model": model,
|
||||
"prompt": "a paper boat drifting across a puddle after rain",
|
||||
"seconds": 6,
|
||||
"size": "2k",
|
||||
},
|
||||
)
|
||||
assert created["status"] == "queued"
|
||||
video_id: Final = created["id"]
|
||||
response: Final = gateway.request("GET", f"/v1/videos/{video_id}")
|
||||
assert response.status_code == 200, response.text
|
||||
status: Final = JSON_OBJECT.validate_json(response.content)
|
||||
assert status["status"] == "completed", status
|
||||
assert status["error"] is None, status
|
||||
assert [(request.method, request.target) for request in wire.drain()] == [
|
||||
("POST", f"/{_H3_MODEL}"),
|
||||
("GET", f"/minimax/h3/requests/{request_id}/status"),
|
||||
("GET", f"/minimax/h3/requests/{request_id}"),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.covers("other.provider_wire.fal_ai.h3_auto_duration_omits_duration_and_queues")
|
||||
def test_fal_h3_auto_duration_omits_duration_and_queues(gateway: Gateway) -> None:
|
||||
request_id: Final = "fal-h3-auto-" + uuid.uuid4().hex
|
||||
|
|
|
|||
|
|
@ -389,6 +389,117 @@ class TestFalAIVideoTransformation:
|
|||
assert "input.reference_image_urls: Failed to download the file" in video.error["message"]
|
||||
client.get.assert_awaited_once_with(url=result_url, headers=auth_headers)
|
||||
|
||||
def test_status_completed_result_probe_reuses_status_client_and_extra_headers(self):
|
||||
status_url = "https://queue.fal.run/minimax/h3/requests/abc/status"
|
||||
status_headers: Final = {
|
||||
"Authorization": "Key synthetic-fal-key",
|
||||
"Content-Type": "application/json",
|
||||
"X-Routing": "canary-7",
|
||||
}
|
||||
response: Final = httpx.Response(
|
||||
200,
|
||||
json={"request_id": "abc", "status": "COMPLETED"},
|
||||
request=httpx.Request("GET", status_url, headers=status_headers),
|
||||
)
|
||||
result_url: Final = status_url.removesuffix("/status")
|
||||
status_client: Final = Mock()
|
||||
status_client.get.return_value = httpx.Response(
|
||||
403, text="missing X-Routing", request=httpx.Request("GET", result_url)
|
||||
)
|
||||
factory_client: Final = Mock()
|
||||
config = FalAIVideoConfig(sync_client_factory=lambda: factory_client)
|
||||
|
||||
video = config.transform_video_status_retrieve_response(
|
||||
raw_response=response,
|
||||
logging_obj=self.logging_obj,
|
||||
custom_llm_provider="fal_ai",
|
||||
client=status_client,
|
||||
)
|
||||
|
||||
assert video.status == "failed"
|
||||
assert video.error == {"code": "fal_error", "message": "missing X-Routing"}
|
||||
factory_client.get.assert_not_called()
|
||||
status_client.get.assert_called_once()
|
||||
assert status_client.get.call_args.kwargs["url"] == result_url
|
||||
assert status_client.get.call_args.kwargs["headers"].items() >= status_headers.items()
|
||||
|
||||
def test_status_completed_result_probe_transport_error_keeps_completed(self):
|
||||
status_url = "https://queue.fal.run/minimax/h3/requests/abc/status"
|
||||
response: Final = httpx.Response(
|
||||
200,
|
||||
json={"request_id": "abc", "status": "COMPLETED"},
|
||||
request=httpx.Request("GET", status_url),
|
||||
)
|
||||
client: Final = Mock()
|
||||
client.get.side_effect = httpx.ReadError("connection reset by fal.ai")
|
||||
|
||||
video = FalAIVideoConfig().transform_video_status_retrieve_response(
|
||||
raw_response=response,
|
||||
logging_obj=self.logging_obj,
|
||||
custom_llm_provider="fal_ai",
|
||||
client=client,
|
||||
)
|
||||
|
||||
assert video.status == "completed"
|
||||
assert video.error is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_status_completed_result_probe_reuses_status_client_and_extra_headers(self):
|
||||
status_url = "https://queue.fal.run/minimax/h3/requests/abc/status"
|
||||
status_headers: Final = {
|
||||
"Authorization": "Key synthetic-fal-key",
|
||||
"Content-Type": "application/json",
|
||||
"X-Routing": "canary-7",
|
||||
}
|
||||
response: Final = httpx.Response(
|
||||
200,
|
||||
json={"request_id": "abc", "status": "COMPLETED"},
|
||||
request=httpx.Request("GET", status_url, headers=status_headers),
|
||||
)
|
||||
result_url: Final = status_url.removesuffix("/status")
|
||||
status_client: Final = Mock()
|
||||
status_client.get = AsyncMock(
|
||||
return_value=httpx.Response(403, text="missing X-Routing", request=httpx.Request("GET", result_url))
|
||||
)
|
||||
factory_client: Final = Mock()
|
||||
factory_client.get = AsyncMock()
|
||||
config = FalAIVideoConfig(async_client_factory=lambda: factory_client)
|
||||
|
||||
video = await config.async_transform_video_status_retrieve_response(
|
||||
raw_response=response,
|
||||
logging_obj=self.logging_obj,
|
||||
custom_llm_provider="fal_ai",
|
||||
client=status_client,
|
||||
)
|
||||
|
||||
assert video.status == "failed"
|
||||
assert video.error == {"code": "fal_error", "message": "missing X-Routing"}
|
||||
factory_client.get.assert_not_awaited()
|
||||
status_client.get.assert_awaited_once()
|
||||
assert status_client.get.await_args.kwargs["url"] == result_url
|
||||
assert status_client.get.await_args.kwargs["headers"].items() >= status_headers.items()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_status_completed_result_probe_transport_error_keeps_completed(self):
|
||||
status_url = "https://queue.fal.run/minimax/h3/requests/abc/status"
|
||||
response: Final = httpx.Response(
|
||||
200,
|
||||
json={"request_id": "abc", "status": "COMPLETED"},
|
||||
request=httpx.Request("GET", status_url),
|
||||
)
|
||||
client: Final = Mock()
|
||||
client.get = AsyncMock(side_effect=httpx.ConnectError("tls handshake failed"))
|
||||
|
||||
video = await FalAIVideoConfig().async_transform_video_status_retrieve_response(
|
||||
raw_response=response,
|
||||
logging_obj=self.logging_obj,
|
||||
custom_llm_provider="fal_ai",
|
||||
client=client,
|
||||
)
|
||||
|
||||
assert video.status == "completed"
|
||||
assert video.error is None
|
||||
|
||||
def test_status_in_progress_does_not_fetch_result(self):
|
||||
status_url = "https://queue.fal.run/minimax/h3/requests/abc/status"
|
||||
response = httpx.Response(
|
||||
|
|
@ -564,17 +675,23 @@ class TestFalAIVideoTransformation:
|
|||
row = litellm.model_cost[f"fal_ai/{H3_TEXT_MODEL}"]
|
||||
model_info = litellm.get_model_info(model=H3_TEXT_MODEL, custom_llm_provider="fal_ai")
|
||||
|
||||
assert video_generation_cost(
|
||||
model=H3_TEXT_MODEL,
|
||||
duration_seconds=5,
|
||||
custom_llm_provider="fal_ai",
|
||||
model_info=model_info,
|
||||
video_resolution="2K",
|
||||
) == 5 * row["output_cost_per_second_2k"]
|
||||
assert video_generation_cost(
|
||||
model=H3_TEXT_MODEL,
|
||||
duration_seconds=5,
|
||||
custom_llm_provider="fal_ai",
|
||||
model_info=model_info,
|
||||
video_resolution="768p",
|
||||
) == 5 * row["output_cost_per_second_768p"]
|
||||
assert (
|
||||
video_generation_cost(
|
||||
model=H3_TEXT_MODEL,
|
||||
duration_seconds=5,
|
||||
custom_llm_provider="fal_ai",
|
||||
model_info=model_info,
|
||||
video_resolution="2K",
|
||||
)
|
||||
== 5 * row["output_cost_per_second_2k"]
|
||||
)
|
||||
assert (
|
||||
video_generation_cost(
|
||||
model=H3_TEXT_MODEL,
|
||||
duration_seconds=5,
|
||||
custom_llm_provider="fal_ai",
|
||||
model_info=model_info,
|
||||
video_resolution="768p",
|
||||
)
|
||||
== 5 * row["output_cost_per_second_768p"]
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue