From 16fc82f44574186eda0cbe45f8296e52e0f10749 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 15 Oct 2026 15:41:05 +0200 Subject: [PATCH 1/6] feat(call_runner): send raw bytes as the request body The response side already returns raw bytes. The request side only took a JSON payload, so apps taking an image or audio needed a base64 shim. `data` and `content_type` send the bytes as is, exclusive with `payload`. Co-Authored-By: Claude Sonnet 5.5 --- CHANGELOG.md | 6 ++ src/livepeer_gateway/http.py | 19 +++- src/livepeer_gateway/live_runner.py | 29 ++++-- tests/test_call_runner_request_bytes.py | 115 ++++++++++++++++++++++++ 4 files changed, 163 insertions(+), 6 deletions(-) create mode 100644 tests/test_call_runner_request_bytes.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1909d4a..1c65e6f 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: `data` (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..16a7d00 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, + data: 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 data is not None: + raise ValueError("pass either payload (JSON) or data (raw bytes), not both") req_headers: dict[str, str] = { "Accept": "application/json", "User-Agent": "livepeer-python-gateway/0.1", @@ -88,10 +92,14 @@ def _json_request_parts( if payload is not None: req_headers["Content-Type"] = "application/json" body = json.dumps(payload).encode("utf-8") + elif data is not None: + req_headers["Content-Type"] = content_type or "application/octet-stream" + body = bytes(data) if headers: req_headers.update(headers) - resolved_method = method.upper() if method else ("POST" if payload is not None else "GET") + has_body = body is not None + resolved_method = method.upper() if method else ("POST" if has_body else "GET") return resolved_method, req_headers, body @@ -228,6 +236,8 @@ async def _request_body( *, method: str | None = None, payload: dict[str, Any] | None = None, + data: bytes | None = None, + content_type: str | None = None, headers: dict[str, str] | None = None, timeout: float = 5.0, ) -> tuple[bytes, str]: @@ -245,6 +255,8 @@ async def _request_body( url, method=method, payload=payload, + data=data, + content_type=content_type, headers=headers, ) @@ -321,6 +333,8 @@ async def open_stream( *, method: str | None = None, payload: dict[str, Any] | None = None, + data: 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 +350,12 @@ async def open_stream( url, method=method, payload=payload, + data=data, + 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..3b6bc67 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 = ..., + data: 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 = ..., + data: 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, + data: 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,23 @@ 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 (``data``), not both. ``data`` 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. Raise ``timeout`` for + large bodies, since it bounds the whole request. + 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 data is not None: + raise LivepeerGatewayError("Live runner call takes payload (JSON) or data (bytes), not both") + request_payload = None if data is not None else (payload or {}) + # Only a raw body needs these; JSON calls keep their existing request arguments. + body_kwargs: dict[str, Any] = ( + {"data": data, "content_type": content_type} if data 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 +889,7 @@ async def call_runner( method=method, payload=request_payload, headers=request_headers or None, + **body_kwargs, ) call_stream = LiveRunnerCallStream( resp.status, @@ -894,16 +912,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 +930,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 +946,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_request_bytes.py b/tests/test_call_runner_request_bytes.py new file mode 100644 index 0000000..1046e61 --- /dev/null +++ b/tests/test_call_runner_request_bytes.py @@ -0,0 +1,115 @@ +"""Tests for raw byte request bodies in call_runner. + +``payload`` keeps sending JSON. ``data`` sends the bytes as is with the given +content type, and the two are mutually exclusive. +""" + +from __future__ import annotations + +import asyncio + +import pytest +from aiohttp import web + +from livepeer_gateway.errors import LivepeerGatewayError +from livepeer_gateway.live_runner import call_runner + +FAKE_PNG = b"\x89PNG\r\n\x1a\n" + bytes(range(256)) * 40 + + +def _run(app: web.Application, scenario): + """Serve `app` on an ephemeral port and run `scenario(base_url)`.""" + + async def main(): + runner = web.AppRunner(app) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + port = site._server.sockets[0].getsockname()[1] + try: + return await scenario(f"http://127.0.0.1:{port}") + finally: + await runner.cleanup() + + return asyncio.run(main()) + + +def _echo_app(seen: dict) -> web.Application: + async def handler(request: web.Request) -> web.Response: + 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_bytes_sent_as_is_with_content_type(): + seen: dict = {} + + async def scenario(base): + return await call_runner(f"{base}/in", data=FAKE_PNG, content_type="image/png") + + result = _run(_echo_app(seen), scenario) + assert seen["body"] == FAKE_PNG + assert seen["content_type"] == "image/png" + assert seen["method"] == "POST" + assert result.data == {"received": len(FAKE_PNG)} + + +def test_bytes_default_content_type_is_octet_stream(): + seen: dict = {} + + async def scenario(base): + return await call_runner(f"{base}/in", data=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_bytes_body_is_sent_as_post(): + seen: dict = {} + + async def scenario(base): + return await call_runner(f"{base}/in", data=b"") + + _run(_echo_app(seen), scenario) + assert seen["method"] == "POST" + assert seen["body"] == b"" + + +def test_json_payload_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_bytes_with_stream(): + seen: dict = {} + + async def scenario(base): + async with await call_runner( + f"{base}/in", data=FAKE_PNG, content_type="image/png", 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_PNG + assert seen["content_type"] == "image/png" + assert b"received" in body + + +def test_payload_and_data_are_mutually_exclusive(): + async def scenario(base): + return await call_runner(f"{base}/in", payload={"x": 1}, data=b"y") + + with pytest.raises(LivepeerGatewayError): + _run(_echo_app({}), scenario) From cbbe63bd699c6e48293ea1c577fc6ef4193034ef Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 15 Oct 2026 16:00:05 +0200 Subject: [PATCH 2/6] refactor(http): inline the body check for the default method Co-Authored-By: Claude Sonnet 5.5 --- src/livepeer_gateway/http.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 16a7d00..abb6f31 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -98,8 +98,7 @@ def _json_request_parts( if headers: req_headers.update(headers) - has_body = body is not None - resolved_method = method.upper() if method else ("POST" if has_body else "GET") + resolved_method = method.upper() if method else ("POST" if body is not None else "GET") return resolved_method, req_headers, body From 49820dd95499f3246c43ed0ebd09e455ded2e969 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 15 Oct 2026 16:42:20 +0200 Subject: [PATCH 3/6] refactor(call_runner): name the raw request body `content` Matches `result.content` on the response side, and stops `data` meaning the JSON dict on the result but raw bytes on the request. Co-Authored-By: Claude Sonnet 5.5 --- CHANGELOG.md | 2 +- src/livepeer_gateway/http.py | 18 +++++++++--------- src/livepeer_gateway/live_runner.py | 18 +++++++++--------- tests/test_call_runner_request_bytes.py | 12 ++++++------ 4 files changed, 25 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1c65e6f..6808d98 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ### Added -- `call_runner` accepts a raw request body: `data` (bytes) with an optional +- `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. diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index abb6f31..ab7aa05 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -78,12 +78,12 @@ def _json_request_parts( *, method: str | None = None, payload: dict[str, Any] | None = None, - data: bytes | 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 data is not None: - raise ValueError("pass either payload (JSON) or data (raw bytes), not both") + 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", @@ -92,9 +92,9 @@ def _json_request_parts( if payload is not None: req_headers["Content-Type"] = "application/json" body = json.dumps(payload).encode("utf-8") - elif data is not None: + elif content is not None: req_headers["Content-Type"] = content_type or "application/octet-stream" - body = bytes(data) + body = bytes(content) if headers: req_headers.update(headers) @@ -235,7 +235,7 @@ async def _request_body( *, method: str | None = None, payload: dict[str, Any] | None = None, - data: bytes | None = None, + content: bytes | None = None, content_type: str | None = None, headers: dict[str, str] | None = None, timeout: float = 5.0, @@ -254,7 +254,7 @@ async def _request_body( url, method=method, payload=payload, - data=data, + content=content, content_type=content_type, headers=headers, ) @@ -332,7 +332,7 @@ async def open_stream( *, method: str | None = None, payload: dict[str, Any] | None = None, - data: bytes | None = None, + content: bytes | None = None, content_type: str | None = None, headers: dict[str, str] | None = None, connect_timeout: float = 10.0, @@ -349,7 +349,7 @@ async def open_stream( url, method=method, payload=payload, - data=data, + content=content, content_type=content_type, headers=headers, ) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 3b6bc67..f95cd3e 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -756,7 +756,7 @@ async def call_runner( *, runner: LiveRunnerInstance | None = ..., payload: dict[str, Any] | None = ..., - data: bytes | None = ..., + content: bytes | None = ..., content_type: str | None = ..., method: str = ..., signer_url: str | None = ..., @@ -775,7 +775,7 @@ async def call_runner( *, runner: LiveRunnerInstance | None = ..., payload: dict[str, Any] | None = ..., - data: bytes | None = ..., + content: bytes | None = ..., content_type: str | None = ..., method: str = ..., signer_url: str | None = ..., @@ -793,7 +793,7 @@ async def call_runner( *, runner: LiveRunnerInstance | None = None, payload: dict[str, Any] | None = None, - data: bytes | None = None, + content: bytes | None = None, content_type: str | None = None, method: str = "POST", signer_url: str | None = None, @@ -812,8 +812,8 @@ 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 (``data``), not both. ``data`` is - sent as is with ``Content-Type: content_type`` (default ``application/octet-stream``), + 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. Raise ``timeout`` for large bodies, since it bounds the whole request. @@ -822,12 +822,12 @@ async def call_runner( 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") - if payload is not None and data is not None: - raise LivepeerGatewayError("Live runner call takes payload (JSON) or data (bytes), not both") - request_payload = None if data is not None else (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] = ( - {"data": data, "content_type": content_type} if data is not None else {} + {"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 diff --git a/tests/test_call_runner_request_bytes.py b/tests/test_call_runner_request_bytes.py index 1046e61..16bc587 100644 --- a/tests/test_call_runner_request_bytes.py +++ b/tests/test_call_runner_request_bytes.py @@ -1,6 +1,6 @@ """Tests for raw byte request bodies in call_runner. -``payload`` keeps sending JSON. ``data`` sends the bytes as is with the given +``payload`` keeps sending JSON. ``content`` sends the bytes as is with the given content type, and the two are mutually exclusive. """ @@ -50,7 +50,7 @@ def test_bytes_sent_as_is_with_content_type(): seen: dict = {} async def scenario(base): - return await call_runner(f"{base}/in", data=FAKE_PNG, content_type="image/png") + return await call_runner(f"{base}/in", content=FAKE_PNG, content_type="image/png") result = _run(_echo_app(seen), scenario) assert seen["body"] == FAKE_PNG @@ -63,7 +63,7 @@ def test_bytes_default_content_type_is_octet_stream(): seen: dict = {} async def scenario(base): - return await call_runner(f"{base}/in", data=b"\x00\x01\x02") + return await call_runner(f"{base}/in", content=b"\x00\x01\x02") _run(_echo_app(seen), scenario) assert seen["body"] == b"\x00\x01\x02" @@ -74,7 +74,7 @@ def test_empty_bytes_body_is_sent_as_post(): seen: dict = {} async def scenario(base): - return await call_runner(f"{base}/in", data=b"") + return await call_runner(f"{base}/in", content=b"") _run(_echo_app(seen), scenario) assert seen["method"] == "POST" @@ -97,7 +97,7 @@ def test_bytes_with_stream(): async def scenario(base): async with await call_runner( - f"{base}/in", data=FAKE_PNG, content_type="image/png", stream=True + f"{base}/in", content=FAKE_PNG, content_type="image/png", stream=True ) as stream: return b"".join([chunk async for chunk in stream.aiter_bytes()]) @@ -109,7 +109,7 @@ async def scenario(base): def test_payload_and_data_are_mutually_exclusive(): async def scenario(base): - return await call_runner(f"{base}/in", payload={"x": 1}, data=b"y") + return await call_runner(f"{base}/in", payload={"x": 1}, content=b"y") with pytest.raises(LivepeerGatewayError): _run(_echo_app({}), scenario) From 08be2bf6e684eb22087f18d0776660e8bb879cff Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 15 Oct 2026 16:43:17 +0200 Subject: [PATCH 4/6] refactor(http): keep the response type apart from the request one Co-Authored-By: Claude Sonnet 5.5 --- src/livepeer_gateway/http.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index ab7aa05..9625cd7 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -265,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") @@ -294,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( From e2031968e9f651a734cc018fff094704220200b2 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 15 Oct 2026 16:43:52 +0200 Subject: [PATCH 5/6] docs(call_runner): say what timeout does and does not bound Co-Authored-By: Claude Sonnet 5.5 --- src/livepeer_gateway/live_runner.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index f95cd3e..e9efb74 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -814,8 +814,11 @@ async def call_runner( 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. Raise ``timeout`` for - large bodies, since it bounds the whole request. + 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. """ From 0f48e7763c7a116393c8ae5de5a14de0e5c889bd Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 15 Oct 2026 16:53:54 +0200 Subject: [PATCH 6/6] test(call_runner): put request bytes tests with the raw response ones Also cover that a paid retry re-sends the raw body unchanged. Co-Authored-By: Claude Sonnet 5.5 --- tests/test_call_runner_raw.py | 99 +++++++++++++++++++- tests/test_call_runner_request_bytes.py | 115 ------------------------ tests/test_live_runner.py | 71 +++++++++++++++ 3 files changed, 167 insertions(+), 118 deletions(-) delete mode 100644 tests/test_call_runner_request_bytes.py 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_call_runner_request_bytes.py b/tests/test_call_runner_request_bytes.py deleted file mode 100644 index 16bc587..0000000 --- a/tests/test_call_runner_request_bytes.py +++ /dev/null @@ -1,115 +0,0 @@ -"""Tests for raw byte request bodies in call_runner. - -``payload`` keeps sending JSON. ``content`` sends the bytes as is with the given -content type, and the two are mutually exclusive. -""" - -from __future__ import annotations - -import asyncio - -import pytest -from aiohttp import web - -from livepeer_gateway.errors import LivepeerGatewayError -from livepeer_gateway.live_runner import call_runner - -FAKE_PNG = b"\x89PNG\r\n\x1a\n" + bytes(range(256)) * 40 - - -def _run(app: web.Application, scenario): - """Serve `app` on an ephemeral port and run `scenario(base_url)`.""" - - async def main(): - runner = web.AppRunner(app) - await runner.setup() - site = web.TCPSite(runner, "127.0.0.1", 0) - await site.start() - port = site._server.sockets[0].getsockname()[1] - try: - return await scenario(f"http://127.0.0.1:{port}") - finally: - await runner.cleanup() - - return asyncio.run(main()) - - -def _echo_app(seen: dict) -> web.Application: - async def handler(request: web.Request) -> web.Response: - 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_bytes_sent_as_is_with_content_type(): - seen: dict = {} - - async def scenario(base): - return await call_runner(f"{base}/in", content=FAKE_PNG, content_type="image/png") - - result = _run(_echo_app(seen), scenario) - assert seen["body"] == FAKE_PNG - assert seen["content_type"] == "image/png" - assert seen["method"] == "POST" - assert result.data == {"received": len(FAKE_PNG)} - - -def test_bytes_default_content_type_is_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_bytes_body_is_sent_as_post(): - 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_payload_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_bytes_with_stream(): - seen: dict = {} - - async def scenario(base): - async with await call_runner( - f"{base}/in", content=FAKE_PNG, content_type="image/png", 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_PNG - assert seen["content_type"] == "image/png" - assert b"received" in body - - -def test_payload_and_data_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[