diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 043d88a..9448530 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -240,20 +240,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: Optional[str] = None, payload: Optional[dict[str, Any]] = None, headers: Optional[dict[str, str]] = 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, @@ -267,14 +270,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})" @@ -297,7 +300,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, _ = 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 fbe3010..fe366c1 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -26,10 +26,11 @@ 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 -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, @@ -129,6 +130,9 @@ class LiveRunnerCallResult: repr=False, compare=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 = "" @dataclass @@ -740,6 +744,9 @@ 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. + + ``application/json`` and ``+json`` types parse into ``result.data``; anything else + (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: @@ -800,16 +807,27 @@ async def call_runner( resp.status, resp.headers, runner_url, runner, payment_session, session, resp, ) - data = await request_json( + body, content_type = await request_data( runner_url, method=method, payload=request_payload, **request_kwargs, ) - 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 `content`. + is_json = _is_json_content_type(content_type) + data: dict[str, Any] = {} + if is_json: + try: + data = json.loads(body) + 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, @@ -819,6 +837,8 @@ 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=None if is_json else body, + content_type=content_type, ) except LivepeerHTTPError as e: if e.status_code != 402: @@ -1137,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 new file mode 100644 index 0000000..8359300 --- /dev/null +++ b/tests/test_call_runner_raw.py @@ -0,0 +1,153 @@ +"""Tests for non-JSON (raw byte) responses in call_runner. + +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.content`` 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.content 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.content == FAKE_JPEG + assert result.content_type == "image/jpeg" + 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.content 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.content == 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") + + 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") 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(): + 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)