diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index a0f9f14..01c4b5d 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_body( 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, _ = await _request_body( + url, + method=method, + payload=payload, + headers=headers, + timeout=timeout, + ) + try: + 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})" + ) from e async def open_stream( diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index 3774708..0f46426 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -22,10 +22,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 _request_body, open_stream, post_json, request_json from .remote_signer import ( GetPaymentResponse, LivePaymentSession, @@ -126,6 +127,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 @@ -739,6 +743,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: @@ -799,16 +806,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_body( 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 (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})" + ) 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, @@ -818,6 +836,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: @@ -1136,6 +1156,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..213b8d4 --- /dev/null +++ b/tests/test_call_runner_raw.py @@ -0,0 +1,182 @@ +"""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.http import request_json +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_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]) + + 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) diff --git a/tests/test_live_runner.py b/tests/test_live_runner.py index 806715b..f199a07 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_body( url: str, *, method: str | None = None, payload: dict[str, object] | None = None, timeout: float, - ) -> dict[str, str]: + ) -> tuple[bytes, 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_body", side_effect=_request_body): 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_body( url: str, *, method: str | None = None, payload: dict[str, object] | None = None, timeout: float, - ) -> dict[str, str]: + ) -> tuple[bytes, 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_body", side_effect=_request_body): 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_body( 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]: 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_body", side_effect=_request_body), 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_body( 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]: 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_body", side_effect=_request_body), 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_body( 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]: 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_body", side_effect=_request_body), 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_body( 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]: 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_body", side_effect=_request_body), 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_body( 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]: 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_body", side_effect=_request_body), 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]: + return json.dumps(data).encode("utf-8"), "application/json" + + class _FakeO2RReader: instances: list[_FakeO2RReader] = []