From 3700a70e319164d874d8a13d68abf88216c310b9 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Wed, 29 Jul 2026 20:21:02 +0200 Subject: [PATCH 1/7] feat: return raw bytes from call_runner for non-JSON responses Co-Authored-By: Claude Fable 5 --- src/livepeer_gateway/http.py | 53 +++++++++++--- src/livepeer_gateway/live_runner.py | 30 +++++++- tests/test_call_runner_raw.py | 110 ++++++++++++++++++++++++++++ 3 files changed, 181 insertions(+), 12 deletions(-) create mode 100644 tests/test_call_runner_raw.py diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index a0f9f14..7c45931 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -239,20 +239,23 @@ def get_json_sync( return request_json_sync(url, headers=headers, timeout=timeout) -async def request_json( +async def request_data( url: str, *, method: str | None = None, payload: dict[str, Any] | None = None, headers: dict[str, str] | None = None, timeout: float = 5.0, -) -> Any: +) -> tuple[bytes, str]: """ - Make an async JSON HTTP request and parse the JSON response. + Make an async JSON-payload HTTP request and return the raw response body. + + Returns ``(body, content_type)`` without assuming the response is JSON; + request semantics and error mapping match request_json. If method is None, defaults to POST when payload is provided, otherwise GET. - Raises LivepeerGatewayError on HTTP/network/JSON parsing errors. + Raises LivepeerGatewayError on HTTP/network errors. """ resolved_method, req_headers, body = _json_request_parts( url, @@ -266,14 +269,14 @@ async def request_json( connector = aiohttp.TCPConnector(ssl=False) 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.text() + raw = await resp.read() + content_type = resp.content_type or "" if resp.status >= 400: - _raise_http_json_error(resp.status, url, raw, dict(resp.headers.items())) - data: Any = json.loads(raw) + _raise_http_json_error( + resp.status, url, raw.decode(errors="replace"), dict(resp.headers.items()) + ) except (SignerRefreshRequired, SkipPaymentCycle, LivepeerGatewayError): raise - except json.JSONDecodeError as e: - raise LivepeerGatewayError(f"HTTP JSON error: endpoint did not return valid JSON: {e} (url={url})") from e except ConnectionRefusedError as e: raise LivepeerGatewayError( f"HTTP JSON error: connection refused (is the server running? is the host/port correct?) (url={url})" @@ -296,7 +299,37 @@ async def request_json( f"HTTP JSON error: unexpected error: {e.__class__.__name__}: {e} (url={url})" ) from e - return data + return raw, content_type + + +async def request_json( + url: str, + *, + method: Optional[str] = None, + payload: Optional[dict[str, Any]] = None, + headers: Optional[dict[str, str]] = None, + timeout: float = 5.0, +) -> Any: + """ + Make an async JSON HTTP request and parse the JSON response. + + If method is None, defaults to POST when payload is provided, otherwise GET. + + Raises LivepeerGatewayError on HTTP/network/JSON parsing errors. + """ + raw, _content_type = await request_data( + url, + method=method, + payload=payload, + headers=headers, + timeout=timeout, + ) + try: + return json.loads(raw) + except json.JSONDecodeError as e: + raise LivepeerGatewayError( + f"HTTP JSON error: endpoint did not return valid JSON: {e} (url={url})" + ) from e async def open_stream( diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 3774708..465ae58 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -25,7 +25,7 @@ from .channel_reader import ChannelReader from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired -from .http import open_stream, post_json, request_json +from .http import open_stream, post_json, request_data, request_json from .remote_signer import ( GetPaymentResponse, LivePaymentSession, @@ -126,6 +126,10 @@ class LiveRunnerCallResult: repr=False, compare=False, ) + # Non-JSON responses (an image, say) arrive unparsed: `raw` holds the body + # bytes and `content_type` its media type, while `data` stays empty. + raw: Optional[bytes] = field(default=None, repr=False) + content_type: str = "" @dataclass @@ -739,6 +743,10 @@ async def call_runner( With ``signer_url`` set, payment is automatic and **per call**: a 402 challenge is paid via the signer and retried (up to ``max_payment_challenge_retries``), one job, one upfront payment. Raises ``LivepeerHTTPError`` on non-402 errors. + + JSON responses (by content type) are parsed into ``result.data``; anything else + (an image, say) is returned unparsed in ``result.raw`` with ``result.content_type`` + set. """ runner_url = runner_url.strip() or (runner.url.strip() if runner is not None else "") if not runner_url: @@ -799,12 +807,29 @@ async def call_runner( resp.status, resp.headers, runner_url, runner, payment_session, session, resp, ) - data = await request_json( + raw, content_type = await request_data( runner_url, method=method, payload=request_payload, **request_kwargs, ) + if "json" not in content_type.lower(): + # Non-JSON body (an image, say): hand back the bytes unparsed. + return LiveRunnerCallResult( + {}, + runner_url=runner_url, + runner=runner, + session_id=session_id, + payment_session=None if payment_type == "fixed" else payment_session, + raw=raw, + content_type=content_type, + ) + try: + data = json.loads(raw) + except json.JSONDecodeError as e: + raise LivepeerGatewayError( + f"HTTP JSON error: endpoint did not return valid JSON: {e} (url={runner_url})" + ) from e if not isinstance(data, dict): raise LivepeerGatewayError( f"Live runner call expected JSON object, got {type(data).__name__}" @@ -818,6 +843,7 @@ async def call_runner( or (data["session_id"].strip() if isinstance(data.get("session_id"), str) else "") ), payment_session=None if payment_type == "fixed" else payment_session, + content_type=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 new file mode 100644 index 0000000..545a859 --- /dev/null +++ b/tests/test_call_runner_raw.py @@ -0,0 +1,110 @@ +"""Tests for non-JSON (raw byte) responses in call_runner. + +JSON responses (by content type) keep today's behavior: parsed into +``result.data``, strict about being an object. Any other content type returns +the body unparsed in ``result.raw`` with ``result.content_type`` set. +""" + +from __future__ import annotations + +import asyncio + +import pytest +from aiohttp import web + +from livepeer_gateway.errors import LivepeerGatewayError, LivepeerHTTPError +from livepeer_gateway.live_runner import call_runner + +FAKE_JPEG = b"\xff\xd8\xff\xe0" + b"jpeg-bytes" * 100 + + +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 test_json_response_unchanged(): + async def handler(request): + return web.json_response({"message": "hello", "session_id": " s1 "}) + + app = web.Application() + app.router.add_post("/call", handler) + + async def scenario(base): + return await call_runner(f"{base}/call", payload={"x": 1}) + + result = _run(app, scenario) + assert result.data == {"message": "hello", "session_id": " s1 "} + assert result.session_id == "s1" + assert result.raw is None + assert result.content_type == "application/json" + + +def test_binary_response_returns_raw(): + async def handler(request): + return web.Response(body=FAKE_JPEG, content_type="image/jpeg") + + app = web.Application() + app.router.add_post("/img", handler) + + async def scenario(base): + return await call_runner(f"{base}/img", payload={"prompt": "x"}) + + result = _run(app, scenario) + assert result.raw == FAKE_JPEG + assert result.content_type == "image/jpeg" + assert result.data == {} + + +def test_invalid_json_with_json_content_type_raises(): + async def handler(request): + return web.Response(text="not json", content_type="application/json") + + app = web.Application() + app.router.add_post("/bad", handler) + + async def scenario(base): + return await call_runner(f"{base}/bad", payload={}) + + with pytest.raises(LivepeerGatewayError, match="did not return valid JSON"): + _run(app, scenario) + + +def test_json_array_still_rejected(): + async def handler(request): + return web.json_response([1, 2, 3]) + + app = web.Application() + app.router.add_post("/arr", handler) + + async def scenario(base): + return await call_runner(f"{base}/arr", payload={}) + + with pytest.raises(LivepeerGatewayError, match="expected JSON object"): + _run(app, scenario) + + +def test_http_error_still_raises_with_binary_endpoint(): + async def handler(request): + return web.Response(status=404, text="nope") + + app = web.Application() + app.router.add_post("/missing", handler) + + async def scenario(base): + return await call_runner(f"{base}/missing", payload={}) + + with pytest.raises(LivepeerHTTPError): + _run(app, scenario) From f45cb26f8d4e2b3d6764c14ae37f7114a750a8cb Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 11:11:13 +0200 Subject: [PATCH 2/7] refactor: match single-document JSON by media subtype MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Content-type detection was a substring test (`"json" in content_type`), which is right for `+json` types only by accident and wrong for multi-document formats: `application/jsonl` (this repo's own trickle channel default), `application/x-ndjson`, and `application/json-seq` all matched and then failed in json.loads, so a working runner fronting a streaming API got "did not return valid JSON" instead of its bytes — the same bug class this branch set out to fix. Match on the media subtype instead, via aiohttp's own mimetype parser: `application/json` or the RFC 6839 `+json` structured suffix. Vendor types (`application/vnd.acme.v1+json`) keep parsing without being listed; multi-document bodies fall through to `raw`. Only four classifications change, all previously raising. Also fold the raw early return into the single existing return, so `payment_session=None if payment_type == "fixed" else payment_session` stays in one place — payment-type handling here has been reverted twice and there is no paid-binary test to catch the two sites drifting. The `session_id` expression reduces to the early return's behavior when `data` is empty, so this is equivalent. Add `content_type` to the JSON parse error: it is the fact that routed the response into parsing, and without it a misclassification is undebuggable from the message alone. Co-Authored-By: Claude Opus 5 (1M context) --- src/livepeer_gateway/http.py | 2 +- src/livepeer_gateway/live_runner.py | 48 ++++++++++++++-------------- tests/test_call_runner_raw.py | 49 +++++++++++++++++++++++++++-- 3 files changed, 71 insertions(+), 28 deletions(-) diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 7c45931..86b7f88 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -317,7 +317,7 @@ async def request_json( Raises LivepeerGatewayError on HTTP/network/JSON parsing errors. """ - raw, _content_type = await request_data( + raw, _ = await request_data( url, method=method, payload=payload, diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 465ae58..de08171 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -22,6 +22,7 @@ from urllib.parse import quote, urlparse, urlunparse import aiohttp +from aiohttp.helpers import parse_mimetype from .channel_reader import ChannelReader from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired @@ -744,9 +745,8 @@ async def call_runner( paid via the signer and retried (up to ``max_payment_challenge_retries``), one job, one upfront payment. Raises ``LivepeerHTTPError`` on non-402 errors. - JSON responses (by content type) are parsed into ``result.data``; anything else - (an image, say) is returned unparsed in ``result.raw`` with ``result.content_type`` - set. + ``application/json`` and ``+json`` types parse into ``result.data``; anything else + (an image, ndjson) comes back unparsed in ``result.raw`` + ``result.content_type``. """ runner_url = runner_url.strip() or (runner.url.strip() if runner is not None else "") if not runner_url: @@ -813,27 +813,21 @@ async def call_runner( payload=request_payload, **request_kwargs, ) - if "json" not in content_type.lower(): - # Non-JSON body (an image, say): hand back the bytes unparsed. - return LiveRunnerCallResult( - {}, - runner_url=runner_url, - runner=runner, - session_id=session_id, - payment_session=None if payment_type == "fixed" else payment_session, - raw=raw, - content_type=content_type, - ) - try: - data = json.loads(raw) - except json.JSONDecodeError as e: - raise LivepeerGatewayError( - f"HTTP JSON error: endpoint did not return valid JSON: {e} (url={runner_url})" - ) from e - if not isinstance(data, dict): - raise LivepeerGatewayError( - f"Live runner call expected JSON object, got {type(data).__name__}" - ) + # Non-JSON bodies (an image, ndjson) are handed back unparsed in `raw`. + is_json = _is_json_content_type(content_type) + data: dict[str, Any] = {} + if is_json: + try: + data = json.loads(raw) + except 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})" + ) from e + if not isinstance(data, dict): + raise LivepeerGatewayError( + f"Live runner call expected JSON object, got {type(data).__name__}" + ) return LiveRunnerCallResult( data, runner_url=runner_url, @@ -843,6 +837,7 @@ async def call_runner( or (data["session_id"].strip() if isinstance(data.get("session_id"), str) else "") ), payment_session=None if payment_type == "fixed" else payment_session, + raw=None if is_json else raw, content_type=content_type, ) except LivepeerHTTPError as e: @@ -1162,6 +1157,11 @@ def _validate_trickle_channel_requests(channels: list[LiveRunnerTrickleChannelRe raise TypeError("trickle channel mime_type must be str") +def _is_json_content_type(content_type: str) -> bool: + mime = parse_mimetype(content_type) + return mime.subtype == "json" or mime.suffix == "json" + + def _is_trickle_channel_response(value: object) -> bool: if not isinstance(value, dict): return False diff --git a/tests/test_call_runner_raw.py b/tests/test_call_runner_raw.py index 545a859..5db6b56 100644 --- a/tests/test_call_runner_raw.py +++ b/tests/test_call_runner_raw.py @@ -1,7 +1,8 @@ """Tests for non-JSON (raw byte) responses in call_runner. -JSON responses (by content type) keep today's behavior: parsed into -``result.data``, strict about being an object. Any other content type returns +Single-document JSON responses (``application/json`` or an RFC 6839 ``+json`` +suffix) keep 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.raw`` with ``result.content_type`` set. """ @@ -68,6 +69,46 @@ async def scenario(base): assert result.data == {} +def test_json_suffix_content_type_is_parsed(): + """RFC 6839 ``+json`` types are single JSON documents, so they parse.""" + + async def handler(request): + return web.Response( + text='{"message": "hello"}', content_type="application/vnd.acme.v1+json" + ) + + app = web.Application() + app.router.add_post("/vnd", handler) + + async def scenario(base): + return await call_runner(f"{base}/vnd", payload={}) + + result = _run(app, scenario) + assert result.data == {"message": "hello"} + assert result.raw is None + assert result.content_type == "application/vnd.acme.v1+json" + + +def test_ndjson_returns_raw(): + """Multi-document formats json.loads can't parse come back as bytes.""" + + body = b'{"token": "Hello"}\n{"token": " world"}\n' + + async def handler(request): + return web.Response(body=body, content_type="application/x-ndjson") + + app = web.Application() + app.router.add_post("/ndjson", handler) + + async def scenario(base): + return await call_runner(f"{base}/ndjson", payload={}) + + result = _run(app, scenario) + assert result.raw == body + assert result.content_type == "application/x-ndjson" + assert result.data == {} + + def test_invalid_json_with_json_content_type_raises(): async def handler(request): return web.Response(text="not json", content_type="application/json") @@ -78,8 +119,10 @@ async def handler(request): async def scenario(base): return await call_runner(f"{base}/bad", payload={}) - with pytest.raises(LivepeerGatewayError, match="did not return valid JSON"): + with pytest.raises(LivepeerGatewayError, match="did not return valid JSON") as excinfo: _run(app, scenario) + # The content type is in the message: it is what routed us into parsing. + assert "content_type=application/json" in str(excinfo.value) def test_json_array_still_rejected(): From 7b4bdb265a75ab48aad3fc1c54d9180d483dc26d Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 11:20:57 +0200 Subject: [PATCH 3/7] docs: trim result field comment and helper docstring `content_type` is self-documenting from the field name; the invariant worth stating is that a populated `raw` means an empty `data`. The subtype helper matches its bare neighbors in this module. Co-Authored-By: Claude Opus 5 (1M context) --- src/livepeer_gateway/live_runner.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index de08171..b0b954f 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -127,8 +127,7 @@ class LiveRunnerCallResult: repr=False, compare=False, ) - # Non-JSON responses (an image, say) arrive unparsed: `raw` holds the body - # bytes and `content_type` its media type, while `data` stays empty. + # Non-JSON responses (an image, say) arrive unparsed in `raw`; `data` stays empty. raw: Optional[bytes] = field(default=None, repr=False) content_type: str = "" From 62f4bd4c3a8da5d00cb7cf6010481b4312e3f3d2 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Thu, 30 Jul 2026 12:09:01 +0200 Subject: [PATCH 4/7] refactor: name the non-JSON body field `content` MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `raw` already means "the original JSON dict, un-normalized" on three dataclasses in this SDK — LiveRunnerSessionEvent.raw, LiveRunnerInstance.raw, LiveVideoToVideo.raw — and examples dump it with json.dumps(x.raw). A `raw: Optional[bytes]` on LiveRunnerCallResult overloads the name to mean the opposite, sharply so on one expression chain: `result.raw` (bytes) sitting one dot from `result.runner.raw` (dict). `content` matches its sibling `content_type` and the ecosystem convention for a response body as bytes (requests/httpx `.content`) — where `.raw` instead means an unread stream object, so the old name actively misled. Nothing consumes the field yet (runner-app-examples#45 and api-proxy both use the streaming path), so this is free now and permanent once released. Semantics unchanged: still `Optional[bytes] = None`, since `b""` is a valid empty body and `None` is the only unambiguous "this was JSON" sentinel. Also renames the local `raw` to `body` — it holds bytes, while `raw` in this codebase reads as a dict. Co-Authored-By: Claude Opus 5 (1M context) --- src/livepeer_gateway/live_runner.py | 14 +++++++------- tests/test_call_runner_raw.py | 10 +++++----- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index b0b954f..551917f 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -127,8 +127,8 @@ class LiveRunnerCallResult: repr=False, compare=False, ) - # Non-JSON responses (an image, say) arrive unparsed in `raw`; `data` stays empty. - raw: Optional[bytes] = field(default=None, repr=False) + # Non-JSON responses (an image, say) arrive unparsed in `content`; `data` stays empty. + content: Optional[bytes] = field(default=None, repr=False) content_type: str = "" @@ -745,7 +745,7 @@ async def call_runner( one upfront payment. Raises ``LivepeerHTTPError`` on non-402 errors. ``application/json`` and ``+json`` types parse into ``result.data``; anything else - (an image, ndjson) comes back unparsed in ``result.raw`` + ``result.content_type``. + (an image, ndjson) comes back unparsed in ``result.content`` + ``result.content_type``. """ runner_url = runner_url.strip() or (runner.url.strip() if runner is not None else "") if not runner_url: @@ -806,18 +806,18 @@ async def call_runner( resp.status, resp.headers, runner_url, runner, payment_session, session, resp, ) - raw, content_type = await request_data( + body, content_type = await request_data( runner_url, method=method, payload=request_payload, **request_kwargs, ) - # Non-JSON bodies (an image, ndjson) are handed back unparsed in `raw`. + # Non-JSON bodies (an image, ndjson) are handed back unparsed in `content`. is_json = _is_json_content_type(content_type) data: dict[str, Any] = {} if is_json: try: - data = json.loads(raw) + data = json.loads(body) except json.JSONDecodeError as e: raise LivepeerGatewayError( f"HTTP JSON error: endpoint did not return valid JSON: {e} " @@ -836,7 +836,7 @@ async def call_runner( or (data["session_id"].strip() if isinstance(data.get("session_id"), str) else "") ), payment_session=None if payment_type == "fixed" else payment_session, - raw=None if is_json else raw, + content=None if is_json else body, content_type=content_type, ) except LivepeerHTTPError as e: diff --git a/tests/test_call_runner_raw.py b/tests/test_call_runner_raw.py index 5db6b56..8359300 100644 --- a/tests/test_call_runner_raw.py +++ b/tests/test_call_runner_raw.py @@ -3,7 +3,7 @@ Single-document JSON responses (``application/json`` or an RFC 6839 ``+json`` suffix) keep 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.raw`` with ``result.content_type`` set. +the body unparsed in ``result.content`` with ``result.content_type`` set. """ from __future__ import annotations @@ -49,7 +49,7 @@ async def scenario(base): result = _run(app, scenario) assert result.data == {"message": "hello", "session_id": " s1 "} assert result.session_id == "s1" - assert result.raw is None + assert result.content is None assert result.content_type == "application/json" @@ -64,7 +64,7 @@ async def scenario(base): return await call_runner(f"{base}/img", payload={"prompt": "x"}) result = _run(app, scenario) - assert result.raw == FAKE_JPEG + assert result.content == FAKE_JPEG assert result.content_type == "image/jpeg" assert result.data == {} @@ -85,7 +85,7 @@ async def scenario(base): result = _run(app, scenario) assert result.data == {"message": "hello"} - assert result.raw is None + assert result.content is None assert result.content_type == "application/vnd.acme.v1+json" @@ -104,7 +104,7 @@ async def scenario(base): return await call_runner(f"{base}/ndjson", payload={}) result = _run(app, scenario) - assert result.raw == body + assert result.content == body assert result.content_type == "application/x-ndjson" assert result.data == {} From 0233659eb4cfddd4bfad70ed638f145fa1320937 Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Fri, 31 Jul 2026 09:17:24 -0700 Subject: [PATCH 5/7] fix: preserve JSON response encoding --- src/livepeer_gateway/http.py | 15 ++--- src/livepeer_gateway/live_runner.py | 6 +- tests/test_call_runner_raw.py | 52 +++++++++++++++++ tests/test_live_runner.py | 90 ++++++++++++++++------------- 4 files changed, 114 insertions(+), 49 deletions(-) diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 86b7f88..78a8db5 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -246,12 +246,12 @@ async def request_data( payload: dict[str, Any] | None = None, headers: dict[str, str] | None = None, timeout: float = 5.0, -) -> tuple[bytes, str]: +) -> tuple[bytes, str, str]: """ Make an async JSON-payload HTTP request and return the raw response body. - Returns ``(body, content_type)`` without assuming the response is JSON; - request semantics and error mapping match request_json. + Returns ``(body, content_type, encoding)`` without assuming the response is + JSON; request semantics and error mapping match request_json. If method is None, defaults to POST when payload is provided, otherwise GET. @@ -271,6 +271,7 @@ async def request_data( async with session.request(resolved_method, url, data=body, headers=req_headers) as resp: raw = await resp.read() content_type = resp.content_type or "" + encoding = resp.get_encoding() if resp.status >= 400: _raise_http_json_error( resp.status, url, raw.decode(errors="replace"), dict(resp.headers.items()) @@ -299,7 +300,7 @@ async def request_data( f"HTTP JSON error: unexpected error: {e.__class__.__name__}: {e} (url={url})" ) from e - return raw, content_type + return raw, content_type, encoding async def request_json( @@ -317,7 +318,7 @@ async def request_json( Raises LivepeerGatewayError on HTTP/network/JSON parsing errors. """ - raw, _ = await request_data( + raw, _, encoding = await request_data( url, method=method, payload=payload, @@ -325,8 +326,8 @@ async def request_json( timeout=timeout, ) try: - return json.loads(raw) - except json.JSONDecodeError as e: + return json.loads(raw.decode(encoding)) + except (UnicodeDecodeError, json.JSONDecodeError) as e: raise LivepeerGatewayError( f"HTTP JSON error: endpoint did not return valid JSON: {e} (url={url})" ) from e diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 551917f..99f4a97 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -806,7 +806,7 @@ async def call_runner( resp.status, resp.headers, runner_url, runner, payment_session, session, resp, ) - body, content_type = await request_data( + body, content_type, encoding = await request_data( runner_url, method=method, payload=request_payload, @@ -817,8 +817,8 @@ async def call_runner( data: dict[str, Any] = {} if is_json: try: - data = json.loads(body) - except json.JSONDecodeError as e: + data = json.loads(body.decode(encoding)) + 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})" diff --git a/tests/test_call_runner_raw.py b/tests/test_call_runner_raw.py index 8359300..4fc6024 100644 --- a/tests/test_call_runner_raw.py +++ b/tests/test_call_runner_raw.py @@ -14,6 +14,7 @@ from aiohttp import web from livepeer_gateway.errors import LivepeerGatewayError, LivepeerHTTPError +from livepeer_gateway.http import request_json from livepeer_gateway.live_runner import call_runner FAKE_JPEG = b"\xff\xd8\xff\xe0" + b"jpeg-bytes" * 100 @@ -125,6 +126,57 @@ async def scenario(base): assert "content_type=application/json" in str(excinfo.value) +def test_json_response_uses_declared_charset(): + body = '{"message": "café"}'.encode("iso-8859-1") + + async def handler(request): + return web.Response( + body=body, + content_type="application/json", + charset="iso-8859-1", + ) + + app = web.Application() + app.router.add_route("*", "/charset", handler) + + async def scenario(base): + direct = await request_json(f"{base}/charset") + result = await call_runner(f"{base}/charset", payload={}) + return direct, result + + direct, result = _run(app, scenario) + assert direct == {"message": "café"} + assert result.data == {"message": "café"} + + +def test_invalid_json_encoding_raises_gateway_error(): + async def handler(request): + return web.Response( + body=b'{"message": "\xff"}', + content_type="application/json", + ) + + app = web.Application() + app.router.add_route("*", "/bad-encoding", handler) + + async def scenario(base): + errors = [] + for request in ( + request_json(f"{base}/bad-encoding"), + call_runner(f"{base}/bad-encoding", payload={}), + ): + try: + await request + except Exception as exc: + errors.append(exc) + return errors + + errors = _run(app, scenario) + assert len(errors) == 2 + assert all(isinstance(error, LivepeerGatewayError) for error in errors) + assert all("did not return valid JSON" in str(error) for error in errors) + + def test_json_array_still_rejected(): async def handler(request): return web.json_response([1, 2, 3]) diff --git a/tests/test_live_runner.py b/tests/test_live_runner.py index 806715b..5960805 100644 --- a/tests/test_live_runner.py +++ b/tests/test_live_runner.py @@ -86,17 +86,17 @@ class TestLiveRunnerSession: async def test_call_runner_returns_json_and_metadata(self) -> None: calls: list[tuple[str, str | None, dict[str, object] | None, float]] = [] - def _request_json( + def _request_data( url: str, *, method: str | None = None, payload: dict[str, object] | None = None, timeout: float, - ) -> dict[str, str]: + ) -> tuple[bytes, str, str]: calls.append((url, method, payload, timeout)) - return {"session_id": "session-1", "ok": "true"} + return _json_data({"session_id": "session-1", "ok": "true"}) - with mock.patch.object(live_runner, "request_json", side_effect=_request_json): + with mock.patch.object(live_runner, "request_data", side_effect=_request_data): result = await call_runner( "https://service.example.com/apps/runner-1/app", payload={"hello": "world"}, @@ -184,20 +184,22 @@ async def test_call_runner_can_attach_runner_instance(self) -> None: raw={"label": "echo"}, ) - def _request_json( + def _request_data( url: str, *, method: str | None = None, payload: dict[str, object] | None = None, timeout: float, - ) -> dict[str, str]: + ) -> tuple[bytes, str, str]: del method, payload, timeout - return { - "session_id": "session-1", - "app_url": "https://service.example.com/app", - } + return _json_data( + { + "session_id": "session-1", + "app_url": "https://service.example.com/app", + } + ) - with mock.patch.object(live_runner, "request_json", side_effect=_request_json): + with mock.patch.object(live_runner, "request_data", side_effect=_request_data): result = await call_runner(runner=runner) assert result.runner is runner @@ -220,25 +222,27 @@ def __init__(self, signer_url: str, **kwargs: object) -> None: async def get_payment(self) -> object: return SimpleNamespace(payment="payment-b64", seg_creds="seg-b64") - def _request_json( + def _request_data( url: str, *, method: str | None = None, payload: dict[str, object] | None = None, headers: dict[str, str] | None = None, timeout: float, - ) -> dict[str, str]: + ) -> tuple[bytes, str, str]: calls.append((url, method, payload, headers, timeout)) if len([call for call in calls if call[0] == runner_url]) == 1: body = _payment_challenge_body("manifest-1") raise LivepeerHTTPError(402, url, body, "payment required") - return { - "session_id": "session-1", - "app_url": "https://service.example.com/app", - } + return _json_data( + { + "session_id": "session-1", + "app_url": "https://service.example.com/app", + } + ) with ( - mock.patch.object(live_runner, "request_json", side_effect=_request_json), + mock.patch.object(live_runner, "request_data", side_effect=_request_data), mock.patch.object(live_runner, "LivePaymentSession", _PaymentSession), mock.patch.object( live_runner, @@ -305,23 +309,23 @@ def __init__(self, signer_url: str, **kwargs: object) -> None: async def get_payment(self) -> object: return SimpleNamespace(payment="payment-b64", seg_creds="seg-b64") - def _request_json( + def _request_data( url: str, *, method: str | None = None, payload: dict[str, object] | None = None, headers: dict[str, str] | None = None, timeout: float, - ) -> dict[str, str]: + ) -> tuple[bytes, str, str]: del method, payload, timeout if headers and "Livepeer-Payment" in headers: - return {"session_id": "session-1"} + return _json_data({"session_id": "session-1"}) raise LivepeerHTTPError( 402, url, _payment_challenge_body("manifest-scope"), "payment required" ) with ( - mock.patch.object(live_runner, "request_json", side_effect=_request_json), + mock.patch.object(live_runner, "request_data", side_effect=_request_data), mock.patch.object(live_runner, "LivePaymentSession", _PaymentSession), mock.patch.object( live_runner, @@ -376,14 +380,14 @@ async def get_payment(self) -> object: seg_creds=f"fixed-segment-{payment_number}", ) - def _request_json( + def _request_data( url: str, *, method: str | None = None, payload: dict[str, object] | None = None, headers: dict[str, str] | None = None, timeout: float, - ) -> dict[str, str]: + ) -> tuple[bytes, str, str]: nonlocal runner_calls del method, payload, timeout runner_calls += 1 @@ -395,13 +399,15 @@ def _request_json( "payment required", ) assert headers["Livepeer-Payment"] == "fixed-payment-2" - return { - "session_id": "fixed-manifest", - "app_url": "https://service.example.com/app", - } + return _json_data( + { + "session_id": "fixed-manifest", + "app_url": "https://service.example.com/app", + } + ) with ( - mock.patch.object(live_runner, "request_json", side_effect=_request_json), + mock.patch.object(live_runner, "request_data", side_effect=_request_data), mock.patch.object(live_runner, "LivePaymentSession", _PaymentSession), mock.patch.object( live_runner, @@ -448,22 +454,24 @@ async def get_payment(self) -> object: raise SignerRefreshRequired("refresh") return SimpleNamespace(payment="payment-2", seg_creds="seg-2") - def _request_json( + def _request_data( url: str, *, method: str | None = None, payload: dict[str, object] | None = None, headers: dict[str, str] | None = None, timeout: float, - ) -> dict[str, str]: + ) -> tuple[bytes, str, str]: nonlocal unpaid_count del timeout calls.append((url, method, payload, headers)) if headers and "Livepeer-Payment" in headers: - return { - "session_id": "session-2", - "app_url": "https://service.example.com/app", - } + return _json_data( + { + "session_id": "session-2", + "app_url": "https://service.example.com/app", + } + ) unpaid_count += 1 raise LivepeerHTTPError( 402, @@ -473,7 +481,7 @@ def _request_json( ) with ( - mock.patch.object(live_runner, "request_json", side_effect=_request_json), + mock.patch.object(live_runner, "request_data", side_effect=_request_data), mock.patch.object(live_runner, "LivePaymentSession", _PaymentSession), mock.patch.object( live_runner, @@ -540,14 +548,14 @@ async def get_payment(self) -> object: payment_attempts += 1 raise SignerRefreshRequired("fixed price not found for session") - def _request_json( + def _request_data( url: str, *, method: str | None = None, payload: dict[str, object] | None = None, headers: dict[str, str] | None = None, timeout: float, - ) -> dict[str, str]: + ) -> tuple[bytes, str, str]: nonlocal unpaid_count del method, payload, timeout calls.append(headers) @@ -560,7 +568,7 @@ def _request_json( ) with ( - mock.patch.object(live_runner, "request_json", side_effect=_request_json), + mock.patch.object(live_runner, "request_data", side_effect=_request_data), mock.patch.object(live_runner, "LivePaymentSession", _PaymentSession), mock.patch.object( live_runner, @@ -596,6 +604,10 @@ def _payment_challenge_body(manifest_id: str) -> str: ) +def _json_data(data: dict[str, object]) -> tuple[bytes, str, str]: + return json.dumps(data).encode("utf-8"), "application/json", "utf-8" + + class _FakeO2RReader: instances: list[_FakeO2RReader] = [] From f474038b154232c325072e2bacb99c4b013a550e Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Fri, 31 Jul 2026 09:27:39 -0700 Subject: [PATCH 6/7] refactor: keep request_data return shape --- src/livepeer_gateway/http.py | 13 ++++++------- src/livepeer_gateway/live_runner.py | 4 ++-- tests/test_call_runner_raw.py | 23 ----------------------- tests/test_live_runner.py | 18 +++++++++--------- 4 files changed, 17 insertions(+), 41 deletions(-) diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 78a8db5..e870b76 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -246,12 +246,12 @@ async def request_data( payload: dict[str, Any] | None = None, headers: dict[str, str] | None = None, timeout: float = 5.0, -) -> tuple[bytes, str, str]: +) -> tuple[bytes, str]: """ Make an async JSON-payload HTTP request and return the raw response body. - Returns ``(body, content_type, encoding)`` without assuming the response is - JSON; request semantics and error mapping match request_json. + Returns ``(body, content_type)`` without assuming the response is JSON; + request semantics and error mapping match request_json. If method is None, defaults to POST when payload is provided, otherwise GET. @@ -271,7 +271,6 @@ async def request_data( async with session.request(resolved_method, url, data=body, headers=req_headers) as resp: raw = await resp.read() content_type = resp.content_type or "" - encoding = resp.get_encoding() if resp.status >= 400: _raise_http_json_error( resp.status, url, raw.decode(errors="replace"), dict(resp.headers.items()) @@ -300,7 +299,7 @@ async def request_data( f"HTTP JSON error: unexpected error: {e.__class__.__name__}: {e} (url={url})" ) from e - return raw, content_type, encoding + return raw, content_type async def request_json( @@ -318,7 +317,7 @@ async def request_json( Raises LivepeerGatewayError on HTTP/network/JSON parsing errors. """ - raw, _, encoding = await request_data( + raw, _ = await request_data( url, method=method, payload=payload, @@ -326,7 +325,7 @@ async def request_json( timeout=timeout, ) try: - return json.loads(raw.decode(encoding)) + return json.loads(raw) except (UnicodeDecodeError, json.JSONDecodeError) as e: raise LivepeerGatewayError( f"HTTP JSON error: endpoint did not return valid JSON: {e} (url={url})" diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 99f4a97..0acec4d 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -806,7 +806,7 @@ async def call_runner( resp.status, resp.headers, runner_url, runner, payment_session, session, resp, ) - body, content_type, encoding = await request_data( + body, content_type = await request_data( runner_url, method=method, payload=request_payload, @@ -817,7 +817,7 @@ async def call_runner( data: dict[str, Any] = {} if is_json: try: - data = json.loads(body.decode(encoding)) + data = json.loads(body) except (UnicodeDecodeError, json.JSONDecodeError) as e: raise LivepeerGatewayError( f"HTTP JSON error: endpoint did not return valid JSON: {e} " diff --git a/tests/test_call_runner_raw.py b/tests/test_call_runner_raw.py index 4fc6024..213b8d4 100644 --- a/tests/test_call_runner_raw.py +++ b/tests/test_call_runner_raw.py @@ -126,29 +126,6 @@ async def scenario(base): assert "content_type=application/json" in str(excinfo.value) -def test_json_response_uses_declared_charset(): - body = '{"message": "café"}'.encode("iso-8859-1") - - async def handler(request): - return web.Response( - body=body, - content_type="application/json", - charset="iso-8859-1", - ) - - app = web.Application() - app.router.add_route("*", "/charset", handler) - - async def scenario(base): - direct = await request_json(f"{base}/charset") - result = await call_runner(f"{base}/charset", payload={}) - return direct, result - - direct, result = _run(app, scenario) - assert direct == {"message": "café"} - assert result.data == {"message": "café"} - - def test_invalid_json_encoding_raises_gateway_error(): async def handler(request): return web.Response( diff --git a/tests/test_live_runner.py b/tests/test_live_runner.py index 5960805..cd7fb8b 100644 --- a/tests/test_live_runner.py +++ b/tests/test_live_runner.py @@ -92,7 +92,7 @@ def _request_data( method: str | None = None, payload: dict[str, object] | None = None, timeout: float, - ) -> tuple[bytes, str, str]: + ) -> tuple[bytes, str]: calls.append((url, method, payload, timeout)) return _json_data({"session_id": "session-1", "ok": "true"}) @@ -190,7 +190,7 @@ def _request_data( method: str | None = None, payload: dict[str, object] | None = None, timeout: float, - ) -> tuple[bytes, str, str]: + ) -> tuple[bytes, str]: del method, payload, timeout return _json_data( { @@ -229,7 +229,7 @@ def _request_data( payload: dict[str, object] | None = None, headers: dict[str, str] | None = None, timeout: float, - ) -> tuple[bytes, str, str]: + ) -> tuple[bytes, str]: calls.append((url, method, payload, headers, timeout)) if len([call for call in calls if call[0] == runner_url]) == 1: body = _payment_challenge_body("manifest-1") @@ -316,7 +316,7 @@ def _request_data( payload: dict[str, object] | None = None, headers: dict[str, str] | None = None, timeout: float, - ) -> tuple[bytes, str, str]: + ) -> tuple[bytes, str]: del method, payload, timeout if headers and "Livepeer-Payment" in headers: return _json_data({"session_id": "session-1"}) @@ -387,7 +387,7 @@ def _request_data( payload: dict[str, object] | None = None, headers: dict[str, str] | None = None, timeout: float, - ) -> tuple[bytes, str, str]: + ) -> tuple[bytes, str]: nonlocal runner_calls del method, payload, timeout runner_calls += 1 @@ -461,7 +461,7 @@ def _request_data( payload: dict[str, object] | None = None, headers: dict[str, str] | None = None, timeout: float, - ) -> tuple[bytes, str, str]: + ) -> tuple[bytes, str]: nonlocal unpaid_count del timeout calls.append((url, method, payload, headers)) @@ -555,7 +555,7 @@ def _request_data( payload: dict[str, object] | None = None, headers: dict[str, str] | None = None, timeout: float, - ) -> tuple[bytes, str, str]: + ) -> tuple[bytes, str]: nonlocal unpaid_count del method, payload, timeout calls.append(headers) @@ -604,8 +604,8 @@ def _payment_challenge_body(manifest_id: str) -> str: ) -def _json_data(data: dict[str, object]) -> tuple[bytes, str, str]: - return json.dumps(data).encode("utf-8"), "application/json", "utf-8" +def _json_data(data: dict[str, object]) -> tuple[bytes, str]: + return json.dumps(data).encode("utf-8"), "application/json" class _FakeO2RReader: From b97ab536eb4434df1ae962dca520a14c32808133 Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Fri, 31 Jul 2026 10:24:00 -0700 Subject: [PATCH 7/7] refactor: make request body helper internal --- src/livepeer_gateway/http.py | 4 ++-- src/livepeer_gateway/live_runner.py | 4 ++-- tests/test_live_runner.py | 28 ++++++++++++++-------------- 3 files changed, 18 insertions(+), 18 deletions(-) diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index e870b76..01c4b5d 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -239,7 +239,7 @@ def get_json_sync( return request_json_sync(url, headers=headers, timeout=timeout) -async def request_data( +async def _request_body( url: str, *, method: str | None = None, @@ -317,7 +317,7 @@ async def request_json( Raises LivepeerGatewayError on HTTP/network/JSON parsing errors. """ - raw, _ = await request_data( + raw, _ = await _request_body( url, method=method, payload=payload, diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 0acec4d..0f46426 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -26,7 +26,7 @@ from .channel_reader import ChannelReader from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired -from .http import open_stream, post_json, request_data, request_json +from .http import _request_body, open_stream, post_json, request_json from .remote_signer import ( GetPaymentResponse, LivePaymentSession, @@ -806,7 +806,7 @@ async def call_runner( resp.status, resp.headers, runner_url, runner, payment_session, session, resp, ) - body, content_type = await request_data( + body, content_type = await _request_body( runner_url, method=method, payload=request_payload, diff --git a/tests/test_live_runner.py b/tests/test_live_runner.py index cd7fb8b..f199a07 100644 --- a/tests/test_live_runner.py +++ b/tests/test_live_runner.py @@ -86,7 +86,7 @@ class TestLiveRunnerSession: async def test_call_runner_returns_json_and_metadata(self) -> None: calls: list[tuple[str, str | None, dict[str, object] | None, float]] = [] - def _request_data( + def _request_body( url: str, *, method: str | None = None, @@ -96,7 +96,7 @@ def _request_data( calls.append((url, method, payload, timeout)) return _json_data({"session_id": "session-1", "ok": "true"}) - with mock.patch.object(live_runner, "request_data", side_effect=_request_data): + with mock.patch.object(live_runner, "_request_body", side_effect=_request_body): result = await call_runner( "https://service.example.com/apps/runner-1/app", payload={"hello": "world"}, @@ -184,7 +184,7 @@ async def test_call_runner_can_attach_runner_instance(self) -> None: raw={"label": "echo"}, ) - def _request_data( + def _request_body( url: str, *, method: str | None = None, @@ -199,7 +199,7 @@ def _request_data( } ) - with mock.patch.object(live_runner, "request_data", side_effect=_request_data): + with mock.patch.object(live_runner, "_request_body", side_effect=_request_body): result = await call_runner(runner=runner) assert result.runner is runner @@ -222,7 +222,7 @@ def __init__(self, signer_url: str, **kwargs: object) -> None: async def get_payment(self) -> object: return SimpleNamespace(payment="payment-b64", seg_creds="seg-b64") - def _request_data( + def _request_body( url: str, *, method: str | None = None, @@ -242,7 +242,7 @@ def _request_data( ) with ( - mock.patch.object(live_runner, "request_data", side_effect=_request_data), + mock.patch.object(live_runner, "_request_body", side_effect=_request_body), mock.patch.object(live_runner, "LivePaymentSession", _PaymentSession), mock.patch.object( live_runner, @@ -309,7 +309,7 @@ def __init__(self, signer_url: str, **kwargs: object) -> None: async def get_payment(self) -> object: return SimpleNamespace(payment="payment-b64", seg_creds="seg-b64") - def _request_data( + def _request_body( url: str, *, method: str | None = None, @@ -325,7 +325,7 @@ def _request_data( ) with ( - mock.patch.object(live_runner, "request_data", side_effect=_request_data), + mock.patch.object(live_runner, "_request_body", side_effect=_request_body), mock.patch.object(live_runner, "LivePaymentSession", _PaymentSession), mock.patch.object( live_runner, @@ -380,7 +380,7 @@ async def get_payment(self) -> object: seg_creds=f"fixed-segment-{payment_number}", ) - def _request_data( + def _request_body( url: str, *, method: str | None = None, @@ -407,7 +407,7 @@ def _request_data( ) with ( - mock.patch.object(live_runner, "request_data", side_effect=_request_data), + mock.patch.object(live_runner, "_request_body", side_effect=_request_body), mock.patch.object(live_runner, "LivePaymentSession", _PaymentSession), mock.patch.object( live_runner, @@ -454,7 +454,7 @@ async def get_payment(self) -> object: raise SignerRefreshRequired("refresh") return SimpleNamespace(payment="payment-2", seg_creds="seg-2") - def _request_data( + def _request_body( url: str, *, method: str | None = None, @@ -481,7 +481,7 @@ def _request_data( ) with ( - mock.patch.object(live_runner, "request_data", side_effect=_request_data), + mock.patch.object(live_runner, "_request_body", side_effect=_request_body), mock.patch.object(live_runner, "LivePaymentSession", _PaymentSession), mock.patch.object( live_runner, @@ -548,7 +548,7 @@ async def get_payment(self) -> object: payment_attempts += 1 raise SignerRefreshRequired("fixed price not found for session") - def _request_data( + def _request_body( url: str, *, method: str | None = None, @@ -568,7 +568,7 @@ def _request_data( ) with ( - mock.patch.object(live_runner, "request_data", side_effect=_request_data), + mock.patch.object(live_runner, "_request_body", side_effect=_request_body), mock.patch.object(live_runner, "LivePaymentSession", _PaymentSession), mock.patch.object( live_runner,