diff --git a/CHANGELOG.md b/CHANGELOG.md index 1909d4a..6808d98 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Added + +- `call_runner` accepts a raw request body: `content` (bytes) with an optional + `content_type` (default `application/octet-stream`), as an alternative to the JSON + `payload`. Use it for apps that take an image, audio or other binary input. + ### Removed - gRPC orchestrator information and selection including protobuf bindings diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 4558c73..9625cd7 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -78,8 +78,12 @@ def _json_request_parts( *, method: str | None = None, payload: dict[str, Any] | None = None, + content: bytes | None = None, + content_type: str | None = None, headers: dict[str, str] | None = None, ) -> tuple[str, dict[str, str], bytes | None]: + if payload is not None and content is not None: + raise ValueError("pass either payload (JSON) or content (raw bytes), not both") req_headers: dict[str, str] = { "Accept": "application/json", "User-Agent": "livepeer-python-gateway/0.1", @@ -88,10 +92,13 @@ def _json_request_parts( if payload is not None: req_headers["Content-Type"] = "application/json" body = json.dumps(payload).encode("utf-8") + elif content is not None: + req_headers["Content-Type"] = content_type or "application/octet-stream" + body = bytes(content) if headers: req_headers.update(headers) - resolved_method = method.upper() if method else ("POST" if payload is not None else "GET") + resolved_method = method.upper() if method else ("POST" if body is not None else "GET") return resolved_method, req_headers, body @@ -228,6 +235,8 @@ async def _request_body( *, method: str | None = None, payload: dict[str, Any] | None = None, + content: bytes | None = None, + content_type: str | None = None, headers: dict[str, str] | None = None, timeout: float = 5.0, ) -> tuple[bytes, str]: @@ -245,6 +254,8 @@ async def _request_body( url, method=method, payload=payload, + content=content, + content_type=content_type, headers=headers, ) @@ -254,7 +265,7 @@ async def _request_body( async with aiohttp.ClientSession(timeout=client_timeout, connector=connector) as session: async with session.request(resolved_method, url, data=body, headers=req_headers) as resp: raw = await resp.read() - content_type = resp.content_type or "" + response_content_type = resp.content_type or "" if resp.status >= 400: _raise_http_json_error( resp.status, url, raw.decode(errors="replace") @@ -283,7 +294,7 @@ async def _request_body( f"HTTP JSON error: unexpected error: {e.__class__.__name__}: {e} (url={url})" ) from e - return raw, content_type + return raw, response_content_type async def request_json( @@ -321,6 +332,8 @@ async def open_stream( *, method: str | None = None, payload: dict[str, Any] | None = None, + content: bytes | None = None, + content_type: str | None = None, headers: dict[str, str] | None = None, connect_timeout: float = 10.0, ) -> tuple[aiohttp.ClientSession, aiohttp.ClientResponse]: @@ -336,9 +349,12 @@ async def open_stream( url, method=method, payload=payload, + content=content, + content_type=content_type, headers=headers, ) + timeout = aiohttp.ClientTimeout(total=None, sock_connect=connect_timeout, sock_read=None) session = aiohttp.ClientSession(timeout=timeout, connector=aiohttp.TCPConnector(ssl=False)) try: diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 8d80b3b..e9efb74 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -756,6 +756,8 @@ async def call_runner( *, runner: LiveRunnerInstance | None = ..., payload: dict[str, Any] | None = ..., + content: bytes | None = ..., + content_type: str | None = ..., method: str = ..., signer_url: str | None = ..., signer_headers: dict[str, str] | None = ..., @@ -773,6 +775,8 @@ async def call_runner( *, runner: LiveRunnerInstance | None = ..., payload: dict[str, Any] | None = ..., + content: bytes | None = ..., + content_type: str | None = ..., method: str = ..., signer_url: str | None = ..., signer_headers: dict[str, str] | None = ..., @@ -789,6 +793,8 @@ async def call_runner( *, runner: LiveRunnerInstance | None = None, payload: dict[str, Any] | None = None, + content: bytes | None = None, + content_type: str | None = None, method: str = "POST", signer_url: str | None = None, signer_headers: dict[str, str] | None = None, @@ -806,12 +812,26 @@ async def call_runner( ``application/json`` and ``+json`` types parse into ``result.data``; anything else (an image, ndjson) comes back unparsed in ``result.content`` + ``result.content_type``. + The request body is JSON (``payload``) or raw bytes (``content``), not both. ``content`` + is sent as is with ``Content-Type: content_type`` (default ``application/octet-stream``), + for apps that take an image, audio or any other binary body. + + ``timeout`` is the total time allowed for each buffered request attempt, so raise it + for large bodies. It restarts on every payment retry, so it does not bound the whole + call, and it is not applied when ``stream=True``. + The request asks for no particular format, so the app picks what it returns. """ runner_url = runner_url.strip() or (runner.url.strip() if runner is not None else "") if not runner_url: raise LivepeerGatewayError("Live runner call requires runner_url") - request_payload = payload or {} + if payload is not None and content is not None: + raise LivepeerGatewayError("Live runner call takes payload (JSON) or content (bytes), not both") + request_payload = None if content is not None else (payload or {}) + # Only a raw body needs these; JSON calls keep their existing request arguments. + body_kwargs: dict[str, Any] = ( + {"content": content, "content_type": content_type} if content is not None else {} + ) payment_type = _runner_payment_type(runner, payment_unit) if signer_url else "" max_price: LiveRunnerPriceInfo | None = None if signer_url and runner is not None and runner.price_info is not None: @@ -872,6 +892,7 @@ async def call_runner( method=method, payload=request_payload, headers=request_headers or None, + **body_kwargs, ) call_stream = LiveRunnerCallStream( resp.status, @@ -894,16 +915,17 @@ async def call_runner( else None ) try: - body, content_type = await _request_body( + body, response_content_type = await _request_body( runner_url, method=method, payload=request_payload, + **body_kwargs, **request_kwargs, ) finally: await _stop_funding(pay_task) # Non-JSON bodies (an image, ndjson) are handed back unparsed in `content`. - is_json = _is_json_content_type(content_type) + is_json = _is_json_content_type(response_content_type) data: dict[str, Any] = {} if is_json: try: @@ -911,7 +933,7 @@ async def call_runner( except (UnicodeDecodeError, json.JSONDecodeError) as e: raise LivepeerGatewayError( f"HTTP JSON error: endpoint did not return valid JSON: {e} " - f"(url={runner_url}, content_type={content_type})" + f"(url={runner_url}, content_type={response_content_type})" ) from e if not isinstance(data, dict): raise LivepeerGatewayError( @@ -927,7 +949,7 @@ async def call_runner( ), payment_session=None if payment_type == "fixed" else payment_session, content=None if is_json else body, - content_type=content_type, + content_type=response_content_type, ) except LivepeerHTTPError as e: if e.status_code != 402: diff --git a/tests/test_call_runner_raw.py b/tests/test_call_runner_raw.py index 213b8d4..443c690 100644 --- a/tests/test_call_runner_raw.py +++ b/tests/test_call_runner_raw.py @@ -1,9 +1,13 @@ -"""Tests for non-JSON (raw byte) responses in call_runner. +"""Tests for raw bytes in call_runner, in both directions. -Single-document JSON responses (``application/json`` or an RFC 6839 ``+json`` -suffix) keep today's behavior: parsed into ``result.data``, strict about being an +Responses: single-document JSON (``application/json`` or an RFC 6839 ``+json`` +suffix) keeps today's behavior: parsed into ``result.data``, strict about being an object. Anything else — binary, or a multi-document format like ndjson — returns the body unparsed in ``result.content`` with ``result.content_type`` set. + +Requests: ``payload`` keeps sending JSON. ``content`` sends the bytes as is with +``content_type`` (default ``application/octet-stream``), and the two are mutually +exclusive. """ from __future__ import annotations @@ -37,6 +41,9 @@ async def main(): return asyncio.run(main()) +# --- Responses: what the runner sends back -------------------------------------- + + def test_json_response_unchanged(): async def handler(request): return web.json_response({"message": "hello", "session_id": " s1 "}) @@ -180,3 +187,89 @@ async def scenario(base): with pytest.raises(LivepeerHTTPError): _run(app, scenario) + + +# --- Requests: what the client sends --------------------------------------------- + + +def _echo_app(seen: dict) -> web.Application: + """Record the request body, method and content type the server receives.""" + + async def handler(request): + seen["body"] = await request.read() + seen["content_type"] = request.headers.get("Content-Type", "") + seen["method"] = request.method + return web.json_response({"received": len(seen["body"])}) + + app = web.Application() + app.router.add_route("*", "/in", handler) + return app + + +def test_binary_request_sent_raw(): + seen: dict = {} + + async def scenario(base): + return await call_runner(f"{base}/in", content=FAKE_JPEG, content_type="image/jpeg") + + result = _run(_echo_app(seen), scenario) + assert seen["body"] == FAKE_JPEG + assert seen["content_type"] == "image/jpeg" + assert seen["method"] == "POST" + assert result.data == {"received": len(FAKE_JPEG)} + + +def test_binary_request_defaults_to_octet_stream(): + seen: dict = {} + + async def scenario(base): + return await call_runner(f"{base}/in", content=b"\x00\x01\x02") + + _run(_echo_app(seen), scenario) + assert seen["body"] == b"\x00\x01\x02" + assert seen["content_type"] == "application/octet-stream" + + +def test_empty_binary_request_is_posted(): + seen: dict = {} + + async def scenario(base): + return await call_runner(f"{base}/in", content=b"") + + _run(_echo_app(seen), scenario) + assert seen["method"] == "POST" + assert seen["body"] == b"" + + +def test_json_request_unchanged(): + seen: dict = {} + + async def scenario(base): + return await call_runner(f"{base}/in", payload={"x": 1}) + + _run(_echo_app(seen), scenario) + assert seen["content_type"] == "application/json" + assert seen["body"] == b'{"x": 1}' + + +def test_binary_request_with_stream(): + seen: dict = {} + + async def scenario(base): + async with await call_runner( + f"{base}/in", content=FAKE_JPEG, content_type="image/jpeg", stream=True + ) as stream: + return b"".join([chunk async for chunk in stream.aiter_bytes()]) + + body = _run(_echo_app(seen), scenario) + assert seen["body"] == FAKE_JPEG + assert seen["content_type"] == "image/jpeg" + assert b"received" in body + + +def test_payload_and_content_are_mutually_exclusive(): + async def scenario(base): + return await call_runner(f"{base}/in", payload={"x": 1}, content=b"y") + + with pytest.raises(LivepeerGatewayError): + _run(_echo_app({}), scenario) diff --git a/tests/test_live_runner.py b/tests/test_live_runner.py index 3d542fd..9f4f6ee 100644 --- a/tests/test_live_runner.py +++ b/tests/test_live_runner.py @@ -279,6 +279,77 @@ def _request_body( assert result.runner is runner + async def test_paid_call_resends_raw_content_on_retry(self) -> None: + calls: list[dict[str, object]] = [] + runner_url = "https://service.example.com/apps/runner-1/session" + runner = LiveRunnerInstance( + url=runner_url, + app="livepeer/app", + runner_id="runner-1", + mode="persistent", + orchestrator_url="https://service.example.com", + raw={}, + price_info=LiveRunnerPriceInfo(100, "wei", "seconds"), + ) + + class _PaymentSession: + def __init__(self, signer_url: str, **kwargs: object) -> None: + pass + + async def get_payment(self) -> object: + return SimpleNamespace(payment="payment-b64", seg_creds="seg-b64") + + def _request_body( + url: str, + *, + method: str | None = None, + payload: dict[str, object] | None = None, + content: bytes | None = None, + content_type: str | None = None, + headers: dict[str, str] | None = None, + timeout: float, + ) -> tuple[bytes, str]: + calls.append( + { + "payload": payload, + "content": content, + "content_type": content_type, + "headers": headers, + } + ) + if len(calls) == 1: + raise LivepeerHTTPError( + 402, url, _payment_challenge_body("manifest-1"), "payment required" + ) + return _json_data({"ok": True}) + + with ( + mock.patch.object(live_runner, "_request_body", side_effect=_request_body), + mock.patch.object(live_runner, "LivePaymentSession", _PaymentSession), + mock.patch.object( + live_runner, + "get_signer_info", + new_callable=mock.AsyncMock, + return_value=SimpleNamespace(address="opaque-payer", sig="opaque-signature"), + ), + ): + result = await call_runner( + runner=runner, + content=b"\x89PNG-bytes", + content_type="image/png", + signer_url="https://signer.example.com", + ) + + assert result.data == {"ok": True} + assert len(calls) == 2 + # The raw body is sent unchanged on the first attempt and again on the paid retry. + for call in calls: + assert call["payload"] is None + assert call["content"] == b"\x89PNG-bytes" + assert call["content_type"] == "image/png" + assert "Livepeer-Payment" not in calls[0]["headers"] # type: ignore[operator] + assert calls[1]["headers"]["Livepeer-Payment"] == "payment-b64" # type: ignore[index] + async def test_paid_call_retries_with_payment_headers(self) -> None: calls: list[ tuple[