diff --git a/examples/echo/client.py b/examples/echo/client.py index 0b91738..35d71da 100755 --- a/examples/echo/client.py +++ b/examples/echo/client.py @@ -5,13 +5,12 @@ import asyncio import sys import time -from contextlib import nullcontext, suppress +from contextlib import nullcontext from pathlib import Path import av from livepeer_gateway.errors import LivepeerGatewayError -from livepeer_gateway.live_runner import stop_runner_session from livepeer_gateway.media_output import MediaOutput from livepeer_gateway.media_publish import MediaPublish from livepeer_gateway.http import post_json @@ -32,6 +31,7 @@ def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run the proxied echo Live Runner demo.") parser.add_argument("input") parser.add_argument("--discovery", default=DEFAULT_DISCOVERY) + parser.add_argument("--signer-url", default="", help="Remote signer URL; enables on-chain payment (self-funding session).") parser.add_argument("--output", default=DEFAULT_OUTPUT) parser.add_argument("--radius", type=int, default=75) parser.add_argument("--max-frames", type=int, default=0, help="Stop after this many input video frames (0 = full file).") @@ -120,42 +120,42 @@ async def main() -> None: if not input_path.exists(): raise SystemExit(f"input file does not exist: {input_path}") - session = None - try: - session = await reserve_session(discovery_url=args.discovery, app=ECHO_APP_ID) - _log("runner_url:", session.runner.url if session.runner is not None else session.runner_url) - _log("session_id:", session.session_id) - _log("app_url:", session.app_url) - - echo = await post_json(f"{session.app_url.rstrip('/')}/echo", {"radius": args.radius}) - in_url = _channel_url(echo, "in") - out_url = _channel_url(echo, "out") - _log("in:", in_url) - _log("out:", out_url) - - with nullcontext(sys.stdout.buffer) if output_stdout else output_path.open("wb") as fh: - def _write_chunk(chunk: bytes) -> None: - fh.write(chunk) - if output_stdout: - fh.flush() - - async with MediaOutput(out_url, on_bytes=_write_chunk): - await _publish_video( - input_path, - in_url, - max_frames=max(0, args.max_frames), - app_url=session.app_url, - blur=args.blur, - ) - _log("publish complete; waiting for output to drain...") - fh.flush() + # The session funds itself while open (auto_pay) and stops the runner + # session on exit; offchain (no --signer-url) the payment loop is a no-op. + async with await reserve_session( + discovery_url=args.discovery, + app=ECHO_APP_ID, + signer_url=args.signer_url or None, + ) as session: + _log("runner_url:", session.runner.url if session.runner is not None else session.runner_url) + _log("session_id:", session.session_id) + _log("app_url:", session.app_url) + + echo = await post_json(f"{session.app_url.rstrip('/')}/echo", {"radius": args.radius}) + in_url = _channel_url(echo, "in") + out_url = _channel_url(echo, "out") + _log("in:", in_url) + _log("out:", out_url) + + with nullcontext(sys.stdout.buffer) if output_stdout else output_path.open("wb") as fh: + def _write_chunk(chunk: bytes) -> None: + fh.write(chunk) + if output_stdout: + fh.flush() + + async with MediaOutput(out_url, on_bytes=_write_chunk): + await _publish_video( + input_path, + in_url, + max_frames=max(0, args.max_frames), + app_url=session.app_url, + blur=args.blur, + ) + _log("publish complete; waiting for output to drain...") + fh.flush() except LivepeerGatewayError as exc: raise SystemExit(f"ERROR: {exc}") from exc - finally: - if session is not None: - with suppress(Exception): - await stop_runner_session(session) if __name__ == "__main__": diff --git a/pyproject.toml b/pyproject.toml index 0a829fc..892e671 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,6 +19,7 @@ generate-lp-rpc = "livepeer_gateway.codegen:main" [project.optional-dependencies] dev = [ "grpcio-tools>=1.65.0", + "pytest>=8.0", ] examples = [ diff --git a/src/livepeer_gateway/__init__.py b/src/livepeer_gateway/__init__.py index f474df5..c1bfd2a 100644 --- a/src/livepeer_gateway/__init__.py +++ b/src/livepeer_gateway/__init__.py @@ -57,6 +57,7 @@ create_trickle_channels, register_runner, remove_trickle_channels, + run_session_payments, stop_runner_session, ) from .discovery import discover_orchestrators, discover_runners @@ -143,6 +144,7 @@ "orchestrator_selector", "runner_selector", "reserve_session", + "run_session_payments", "StartJobRequest", "call_runner", "create_proxy", diff --git a/src/livepeer_gateway/http.py b/src/livepeer_gateway/http.py index 043d88a..db4057b 100644 --- a/src/livepeer_gateway/http.py +++ b/src/livepeer_gateway/http.py @@ -371,6 +371,50 @@ async def get_json( return await request_json(url, headers=headers, timeout=timeout) +async def post_empty( + url: str, + *, + headers: Optional[dict[str, str]] = None, + timeout: float = 5.0, +) -> None: + """ + POST an empty body to `url` and discard the response. + + Certificate verification is disabled, matching every other HTTP helper in + the SDK. Error responses raise the same typed errors as request_json + (SignerRefreshRequired for 480, SkipPaymentCycle for 482, LivepeerHTTPError + otherwise) so callers can branch on status codes. + """ + try: + client_timeout = aiohttp.ClientTimeout(total=timeout) + connector = aiohttp.TCPConnector(ssl=False) + async with aiohttp.ClientSession(timeout=client_timeout, connector=connector) as session: + async with session.post(url, data=b"", headers=headers) as resp: + if resp.status >= 400: + raw = await resp.text() + _raise_http_json_error(resp.status, url, raw, dict(resp.headers.items())) + await resp.read() + except (SignerRefreshRequired, SkipPaymentCycle, LivepeerGatewayError): + raise + except ConnectionRefusedError as e: + raise LivepeerGatewayError( + f"HTTP empty POST error: connection refused (is the server running? is the host/port correct?) (url={url})" + ) from e + except getattr(aiohttp, "ClientConnectorError", ()) as e: + os_error = getattr(e, "os_error", None) + if isinstance(os_error, ConnectionRefusedError): + raise LivepeerGatewayError( + f"HTTP empty POST error: connection refused (is the server running? is the host/port correct?) (url={url})" + ) from e + raise LivepeerGatewayError( + f"HTTP empty POST error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})" + ) from e + except (aiohttp.ClientError, asyncio.TimeoutError) as e: + raise LivepeerGatewayError( + f"HTTP empty POST error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})" + ) from e + + def _parse_http_url(url: str, *, context: str = "URL") -> ParseResult: """ Normalize a URL for HTTP(S) endpoints. diff --git a/src/livepeer_gateway/live_runner.py b/src/livepeer_gateway/live_runner.py index fbe3010..1b7aff6 100644 --- a/src/livepeer_gateway/live_runner.py +++ b/src/livepeer_gateway/live_runner.py @@ -28,8 +28,8 @@ import aiohttp from .channel_reader import ChannelReader -from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired -from .http import open_stream, post_json, request_json +from .errors import LivepeerGatewayError, LivepeerHTTPError, SignerRefreshRequired, SkipPaymentCycle +from .http import open_stream, post_empty, post_json, request_json from .remote_signer import ( GetPaymentResponse, LivePaymentSession, @@ -110,6 +110,114 @@ class LiveRunnerSession: app_url: str runner_url: str runner: Optional[LiveRunnerInstance] = None + # Session control base URL from the reserve response. Payments go to + # {control_url}/payment, which is scoped to this session and 404s once the + # orchestrator releases it. Empty for orchestrators that predate it. + control_url: str = "" + # Present when the session was reserved on-chain (signer_url given). + # reserve_session() starts the payment loop automatically (auto_pay=True); + # drive it manually with start_payments() / stop_payments() otherwise. + # run_session_payments() is the underlying loop. + payment_session: Optional[LivePaymentSession] = field( + default=None, + repr=False, + compare=False, + ) + # Seconds between session payments while the session is held open. Must stay + # at or below the orchestrator's livePaymentInterval (5s default) so credits + # lead its server-side debit ticker. Default 3s keeps margin under that. + payment_interval: float = 3.0 + # Background payment task, owned by this session once start_payments() runs. + _payment_task: Optional[asyncio.Task] = field( + default=None, + repr=False, + compare=False, + ) + # Set once the orchestrator reports the session gone (payment 404). + _released: asyncio.Event = field( + default_factory=asyncio.Event, + repr=False, + compare=False, + ) + + @property + def released(self) -> bool: + """True once the orchestrator has released this session.""" + return self._released.is_set() + + async def wait_released(self) -> None: + """Block until the orchestrator releases this session. + + Only fires when the payment loop observes the release (payment 404); + offchain sessions and sessions without a running payment loop never set + it. + """ + await self._released.wait() + + def _mark_released(self) -> None: + self._released.set() + + def start_payments(self) -> Optional[asyncio.Task]: + """Start the background payment loop that keeps this session funded. + + Idempotent and safe to call repeatedly. No-op offchain (no + ``payment_session``). Requires a running event loop; if called from sync + code it logs a warning and returns ``None``. Returns the task, or + ``None`` if payments could not be started. + """ + if getattr(self, "_payment_task", None) is not None: + return self._payment_task + if self.payment_session is None: + return None + try: + loop = asyncio.get_running_loop() + except RuntimeError: + _LOG.warning( + "No running event loop; session payments not started. " + "Call session.start_payments() from async code to enable." + ) + return None + task = loop.create_task(run_session_payments(self, interval=self.payment_interval)) + object.__setattr__(self, "_payment_task", task) + return task + + async def stop_payments(self) -> None: + """Cancel the payment loop without stopping the session. + + Symmetrical complement to start_payments(); the session stays reserved + (and the orchestrator keeps debiting it), so use this only to hand off + or drain funding. No-op when no loop is running. + """ + task = getattr(self, "_payment_task", None) + object.__setattr__(self, "_payment_task", None) + if task is not None and not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + async def aclose(self) -> None: + """Cancel the payment loop (if running) and stop the runner session. + + Best-effort: both steps run even if one fails, and the first + non-cancellation error is re-raised after cleanup. The stop call is + skipped when the orchestrator already released the session. + """ + awaitables: list[Awaitable[Any]] = [self.stop_payments()] + if not self.released: + awaitables.append(stop_runner_session(self)) + results = await asyncio.gather(*awaitables, return_exceptions=True) + for result in results: + if isinstance(result, BaseException) and not isinstance(result, asyncio.CancelledError): + raise result + + async def __aenter__(self) -> "LiveRunnerSession": + self.start_payments() + return self + + async def __aexit__(self, *exc_info: object) -> None: + await self.aclose() @dataclass(frozen=True) @@ -129,6 +237,9 @@ class LiveRunnerCallResult: repr=False, compare=False, ) + # Orchestrator debit cadence in seconds, from the payment challenge's + # payment_interval_ms. None when the orchestrator does not report it. + server_payment_interval: Optional[float] = None @dataclass @@ -145,8 +256,15 @@ class LiveRunnerCallStream: runner_url: str runner: Optional[LiveRunnerInstance] payment_session: Optional[LivePaymentSession] - _session: aiohttp.ClientSession = field(repr=False, compare=False) - _response: aiohttp.ClientResponse = field(repr=False, compare=False) + # Session backing this single-shot request, from the payment challenge's + # manifest_id. Empty offchain: a stream body cannot supply the JSON-path + # session_id fallback. + session_id: str = "" + # Orchestrator debit cadence in seconds, from the payment challenge's + # payment_interval_ms. None when the orchestrator does not report it. + server_payment_interval: Optional[float] = None + _session: aiohttp.ClientSession = field(repr=False, compare=False, kw_only=True) + _response: aiohttp.ClientResponse = field(repr=False, compare=False, kw_only=True) @property def content_type(self) -> str: @@ -296,10 +414,10 @@ async def close(self) -> None: _LOG.warning("Skipping live runner unregister without heartbeat secret") return try: - await _post_empty( + await post_empty( _join_endpoint(self.orchestrator_url, f"/runners/{quote(self.runner_id, safe='')}/unregister"), - {"Authorization": secret}, - self._timeout, + headers={"Authorization": secret}, + timeout=self._timeout, ) except Exception: _LOG.debug("Live runner unregister failed", exc_info=True) @@ -797,7 +915,17 @@ async def call_runner( headers=request_headers or None, ) return LiveRunnerCallStream( - resp.status, resp.headers, runner_url, runner, payment_session, session, resp, + status=resp.status, + headers=resp.headers, + runner_url=runner_url, + runner=runner, + session_id=session_id, + payment_session=None if payment_type == "fixed" else payment_session, + server_payment_interval=( + challenge.payment_interval_s if challenge is not None else None + ), + _session=session, + _response=resp, ) data = await request_json( @@ -819,6 +947,9 @@ 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, + server_payment_interval=( + challenge.payment_interval_s if challenge is not None else None + ), ) except LivepeerHTTPError as e: if e.status_code != 402: @@ -836,6 +967,9 @@ class _RunnerPaymentChallenge: payment_params: str orchestrator_url: str manifest_id: str + # Orchestrator debit cadence in seconds (payment_interval_ms); None when + # the orchestrator does not report it. + payment_interval_s: Optional[float] = None def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentChallenge: @@ -856,10 +990,16 @@ def _parse_runner_payment_challenge(error: LivepeerHTTPError) -> _RunnerPaymentC if not isinstance(manifest_id, str) or not manifest_id: raise LivepeerGatewayError("Live runner payment challenge missing manifest_id") + interval_ms = data.get("payment_interval_ms") + payment_interval_s: Optional[float] = None + if isinstance(interval_ms, (int, float)) and not isinstance(interval_ms, bool) and interval_ms > 0: + payment_interval_s = float(interval_ms) / 1000.0 + return _RunnerPaymentChallenge( payment_params=payment_params, orchestrator_url=orchestrator_url, manifest_id=manifest_id, + payment_interval_s=payment_interval_s, ) @@ -934,25 +1074,6 @@ def _live_runner_price_info_from_json(value: object) -> Optional[LiveRunnerPrice ) -def _live_runner_session_from_json( - data: dict[str, Any], - *, - runner_url: str, - runner: Optional[LiveRunnerInstance], -) -> LiveRunnerSession: - session_id = data.get("session_id") - app_url = data.get("app_url") - if not isinstance(session_id, str) or not session_id.strip(): - raise LivepeerGatewayError("Live runner session reserve response missing session_id") - if not isinstance(app_url, str) or not app_url.strip(): - raise LivepeerGatewayError("Live runner session reserve response missing app_url") - return LiveRunnerSession( - session_id=session_id.strip(), - app_url=app_url.strip(), - runner_url=runner_url, - runner=runner, - ) - async def stop_runner_session( session: LiveRunnerSession | LiveRunnerSessionRequest, *, @@ -977,10 +1098,10 @@ async def stop_runner_session( url = _join_endpoint(control_url, "stop") if isinstance(token, str) and token.strip(): request_headers = {"Livepeer-Session-Token": token} - await _post_empty( + await post_empty( url, - request_headers, - timeout, + headers=request_headers, + timeout=timeout, ) @@ -1146,25 +1267,6 @@ def _is_trickle_channel_response(value: object) -> bool: ) and ("internal_url" not in value or isinstance(value.get("internal_url"), str)) -async def _post_empty(url: str, headers: dict[str, str], timeout: float) -> None: - try: - client_timeout = aiohttp.ClientTimeout(total=timeout) - connector = aiohttp.TCPConnector(ssl=False) - async with aiohttp.ClientSession(timeout=client_timeout, connector=connector) as session: - async with session.post(url, data=b"", headers=headers) as resp: - body = await resp.text() - if resp.status >= 400: - raise LivepeerGatewayError( - f"HTTP empty POST error: HTTP {resp.status}; body={body!r}" - ) - except LivepeerGatewayError: - raise - except getattr(aiohttp, "ClientConnectorError", ()) as e: - raise LivepeerGatewayError(f"HTTP empty POST error: {getattr(e, 'message', e)}") from e - except (aiohttp.ClientError, asyncio.TimeoutError) as e: - raise LivepeerGatewayError(f"HTTP empty POST error: {getattr(e, 'message', e)}") from e - - def _detect_gpu_pynvml() -> Optional[LiveRunnerGPU]: try: import pynvml # type: ignore[import-not-found] @@ -1316,3 +1418,75 @@ def _decode_maybe_bytes(value: object) -> str: if isinstance(value, bytes): return value.decode("utf-8", errors="replace") return str(value or "") + + +async def run_session_payments( + session: LiveRunnerSession, + *, + interval: float = 3.0, +) -> None: + """Keep a reserved live-runner session funded for its whole lifetime. + + After the reservation payment, go-livepeer holds the session as a prepaid + balance: a server-side ticker (``-livePaymentInterval``, default 5s) debits it + and releases the session once it runs dry. The client must keep crediting it + out-of-band, so this pushes payments on a cadence below that server tick. It + sends an initial payment immediately, then one every ``interval`` seconds until + cancelled. + + The immediate first payment matters: a cold start can leave a long gap after + the reservation payment, so top up before the first sleep. The orchestrator + can answer a payment with a skip signal (HTTP 482, ``SkipPaymentCycle``) when + the balance is still sufficient; that is a normal "paid up" response, not a + failure. Other per-cycle failures are logged and retried rather than killing + the loop. + + When the session carries a ``control_url``, payments go to the session-scoped + ``{control_url}/payment`` endpoint, which knows whether the session still + exists. A 404 there means the orchestrator released the session: the loop + stops and marks the session released (``session.released`` / + ``session.wait_released()``). A 409 means the session is fixed-price and + needs no follow-up payments. Without a ``control_url`` (older orchestrators) + the generic ``/payment`` endpoint is used, which credits blindly and cannot + detect a released session. + + No-op for offchain sessions (no ``payment_session``). ``interval`` must stay + at or below the orchestrator's ``livePaymentInterval`` (5s default) so the + balance stays ahead; tune it to the deployment. + """ + payment_session = session.payment_session + if payment_session is None: + return + payment_url = _join_endpoint(session.control_url, "payment") if session.control_url else None + while True: + try: + await payment_session.send_payment(payment_url=payment_url) + except SkipPaymentCycle as exc: + # Orchestrator says the balance is current; this is a healthy gate, not an error. + _LOG.debug("Live runner session payment skipped (balance current): %s", exc) + except LivepeerHTTPError as exc: + if exc.status_code == 404: + _LOG.warning( + "Live runner session %s released by orchestrator; stopping payments", + session.session_id, + ) + session._mark_released() + return + if exc.status_code == 409: + _LOG.info( + "Live runner session %s is fixed-price; no follow-up payments needed", + session.session_id, + ) + return + if exc.status_code == 403: + _LOG.error( + "Live runner session %s payment rejected (mismatched session/payment); " + "stopping payments: %s", + session.session_id, + exc, + ) + return + _LOG.warning("Live runner session payment failed: %s", exc) + except Exception as exc: # noqa: BLE001 - keep the loop alive across transient failures + _LOG.warning("Live runner session payment failed: %s", exc) + await asyncio.sleep(interval) diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index fc94463..b3f5b5a 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -1,6 +1,5 @@ from __future__ import annotations -import asyncio import base64 import json import logging @@ -12,7 +11,6 @@ from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen -import aiohttp from . import lp_rpc_pb2 from .async_cache import async_lru_cache @@ -240,44 +238,41 @@ async def get_payment(self) -> GetPaymentResponse: await self._refresh_payment_params(orchestrator_url) attempts += 1 - async def send_payment(self, orchestrator_url: Optional[str] = None) -> None: + async def send_payment( + self, + orchestrator_url: Optional[str] = None, + *, + payment_url: Optional[str] = None, + ) -> None: + """Generate a payment and POST it to the orchestrator. + + ``payment_url`` targets a specific payment endpoint, such as the + session-scoped ``{control_url}/payment`` which 404s once the session is + released. Without it the payment goes to the orchestrator's generic + ``/payment`` endpoint, which credits the payer balance blindly. + + Raises LivepeerHTTPError on error responses so callers can branch on + the status code, and SkipPaymentCycle when the signer gates the cycle. + """ if not self._signer_url: return - target = orchestrator_url or self._orchestrator_url - if not target: - raise PaymentError("orchestrator_url is required before sending payment") + from .http import _http_origin, post_empty - from .http import _extract_error_message_from_body, _http_origin + if payment_url: + url = payment_url + else: + target = orchestrator_url or self._orchestrator_url + if not target: + raise PaymentError("orchestrator_url is required before sending payment") + url = f"{_http_origin(target)}/payment" payment = await self.get_payment() - url = f"{_http_origin(target)}/payment" headers = { "Livepeer-Payment": payment.payment, - "Livepeer-Segment": payment.seg_creds, + "Livepeer-Segment": payment.seg_creds or "", } - try: - timeout = aiohttp.ClientTimeout(total=5.0) - async with aiohttp.ClientSession(timeout=timeout) as session: - async with session.post(url, data=b"", headers=headers) as resp: - if resp.status >= 400: - body = await resp.text() - message = _extract_error_message_from_body(body) - body_part = f"; body={message!r}" if message else "" - raise PaymentError( - f"HTTP payment error: HTTP {resp.status} from endpoint (url={url}){body_part}" - ) - await resp.read() - except PaymentError: - raise - except getattr(aiohttp, "ClientConnectorError", ()) as e: - raise PaymentError( - f"HTTP payment error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})" - ) from e - except (aiohttp.ClientError, asyncio.TimeoutError) as e: - raise PaymentError( - f"HTTP payment error: failed to reach endpoint: {getattr(e, 'message', e)} (url={url})" - ) from e + await post_empty(url, headers=headers, timeout=5.0) async def _payment_request(self) -> GetPaymentResponse: from .http import _http_origin, post_json diff --git a/src/livepeer_gateway/selection.py b/src/livepeer_gateway/selection.py index 5cb8e82..0b8509e 100644 --- a/src/livepeer_gateway/selection.py +++ b/src/livepeer_gateway/selection.py @@ -271,6 +271,12 @@ async def runner_selector( ) +# Fraction of the orchestrator's debit interval to pay at, so credits lead the +# server-side ticker with margin. 5s server tick -> 3s client cadence. +_PAYMENT_INTERVAL_MARGIN = 0.6 +_DEFAULT_PAYMENT_INTERVAL_S = 3.0 + + async def reserve_session( *, signer_url: Optional[str] = None, @@ -281,7 +287,23 @@ async def reserve_session( app: Optional[FilterValue] = None, gpu: Optional[FilterValue] = None, timeout: float = 5.0, + payment_interval: Optional[float] = None, + auto_pay: bool = True, ) -> LiveRunnerSession: + """Reserve a persistent live-runner session on the best available runner. + + On-chain sessions are self-funding: the returned session already runs its + payment loop (``auto_pay=True``), so any transport against ``app_url`` + (websocket, trickle, SSE, plain HTTP) stays paid for without further calls. + Stop it with ``aclose()`` or by using the session as an async context + manager; the loop also stops on its own when the orchestrator reports the + session released. Pass ``auto_pay=False`` to own the lifecycle manually via + ``start_payments()`` / ``stop_payments()``. + + ``payment_interval`` overrides the payment cadence in seconds. By default it + is derived from the orchestrator's advertised ``payment_interval_ms`` (with + margin), falling back to 3s when the orchestrator does not report one. + """ cursor = await runner_selector( orchestrators=orchestrators, signer_url=signer_url, @@ -295,16 +317,29 @@ async def reserve_session( result = await cursor.next() session_id = result.data.get("session_id") app_url = result.data.get("app_url") + control_url = result.data.get("control_url") if not isinstance(session_id, str) or not session_id.strip(): raise LivepeerGatewayError("runner session response missing session_id") if not isinstance(app_url, str) or not app_url.strip(): raise LivepeerGatewayError("runner session response missing app_url") - return LiveRunnerSession( + interval = payment_interval + if interval is None: + if result.server_payment_interval is not None: + interval = result.server_payment_interval * _PAYMENT_INTERVAL_MARGIN + else: + interval = _DEFAULT_PAYMENT_INTERVAL_S + session = LiveRunnerSession( session_id=session_id.strip(), app_url=app_url.strip(), runner_url=result.runner_url, runner=result.runner, + control_url=control_url.strip() if isinstance(control_url, str) else "", + payment_interval=interval, + payment_session=result.payment_session, ) + if auto_pay: + session.start_payments() + return session def _runner_candidates_from_discovery(entries: Sequence[dict[str, Any]]) -> list[LiveRunnerInstance]: diff --git a/tests/test_live_runner_payments.py b/tests/test_live_runner_payments.py new file mode 100644 index 0000000..52e76e2 --- /dev/null +++ b/tests/test_live_runner_payments.py @@ -0,0 +1,491 @@ +"""Unit tests for live-runner session payment lifecycle. + +Covers the session-owned payment loop added on top of run_session_payments: +- run_session_payments pays immediately, then on interval, and is a no-op offchain +- session-scoped payment endpoint ({control_url}/payment) and liveness handling: + 404 stops the loop and marks the session released, 409 (fixed-price) stops it +- LiveRunnerSession.start_payments is idempotent, offchain-safe, loop-aware +- stop_payments cancels only the loop; aclose / async-context-manager also stop + the session (skipping the stop call when the orchestrator already released it) +- payment challenge parsing picks up payment_interval_ms +- call_runner(stream=True) mirrors the JSON path's payment semantics (challenge + session_id, server_payment_interval, fixed-price payment_session drop) +- reserve_session auto-starts payments and derives the cadence from the challenge +""" +from __future__ import annotations + +import asyncio +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest + +from livepeer_gateway import live_runner, selection +from livepeer_gateway.errors import LivepeerHTTPError, SkipPaymentCycle +from livepeer_gateway.live_runner import ( + LiveRunnerCallResult, + LiveRunnerCallStream, + LiveRunnerSession, + _parse_runner_payment_challenge, + call_runner, + run_session_payments, +) +from livepeer_gateway.selection import reserve_session + + +class _FakePaymentSession: + """Duck-typed stand-in for LivePaymentSession (only send_payment is used).""" + + def __init__(self) -> None: + self.calls = 0 + self.payment_urls: list[str | None] = [] + self.paid = asyncio.Event() + + async def send_payment( + self, + orchestrator_url: str | None = None, + *, + payment_url: str | None = None, + ) -> None: + self.calls += 1 + self.payment_urls.append(payment_url) + self.paid.set() + + +def _session(payment_session=None, interval: float = 10.0, control_url: str = "") -> LiveRunnerSession: + return LiveRunnerSession( + session_id="sess-1", + app_url="http://app", + runner_url="http://runner", + control_url=control_url, + payment_session=payment_session, + payment_interval=interval, + ) + + +def _http_error(status: int) -> LivepeerHTTPError: + return LivepeerHTTPError(status, "http://orch/payment") + + +def test_run_session_payments_noop_offchain() -> None: + async def go() -> None: + # Returns immediately when there is no payment_session. + await asyncio.wait_for(run_session_payments(_session(None), interval=0.01), timeout=1.0) + + asyncio.run(go()) + + +def test_run_session_payments_pays_immediately() -> None: + async def go() -> None: + ps = _FakePaymentSession() + # Long interval: only the immediate first payment should land before cancel. + task = asyncio.create_task(run_session_payments(_session(ps), interval=10.0)) + await asyncio.wait_for(ps.paid.wait(), timeout=1.0) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert ps.calls >= 1 + + asyncio.run(go()) + + +def test_run_session_payments_uses_session_scoped_endpoint() -> None: + async def go() -> None: + ps = _FakePaymentSession() + sess = _session(ps, control_url="https://orch/apps/r1/session/sess-1") + task = asyncio.create_task(run_session_payments(sess, interval=10.0)) + await asyncio.wait_for(ps.paid.wait(), timeout=1.0) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert ps.payment_urls[0] == "https://orch/apps/r1/session/sess-1/payment" + + asyncio.run(go()) + + +def test_run_session_payments_without_control_url_uses_generic_endpoint() -> None: + async def go() -> None: + ps = _FakePaymentSession() + task = asyncio.create_task(run_session_payments(_session(ps), interval=10.0)) + await asyncio.wait_for(ps.paid.wait(), timeout=1.0) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + # Old orchestrators without control_url fall back to send_payment's default. + assert ps.payment_urls[0] is None + + asyncio.run(go()) + + +def test_run_session_payments_survives_payment_error() -> None: + async def go() -> None: + ps = _FakePaymentSession() + original = ps.send_payment + attempts = {"n": 0} + + async def flaky(orchestrator_url: str | None = None, *, payment_url: str | None = None) -> None: + attempts["n"] += 1 + if attempts["n"] == 1: + raise RuntimeError("transient signer error") + await original(orchestrator_url, payment_url=payment_url) + + ps.send_payment = flaky # type: ignore[assignment] + task = asyncio.create_task(run_session_payments(_session(ps), interval=0.01)) + await asyncio.wait_for(ps.paid.wait(), timeout=1.0) # set on the 2nd, successful cycle + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert attempts["n"] >= 2 + + asyncio.run(go()) + + +def test_run_session_payments_treats_skip_cycle_as_paid_up() -> None: + async def go() -> None: + ps = _FakePaymentSession() + original = ps.send_payment + attempts = {"n": 0} + + async def skip_then_pay(orchestrator_url: str | None = None, *, payment_url: str | None = None) -> None: + attempts["n"] += 1 + if attempts["n"] == 1: + raise SkipPaymentCycle("HTTP 482 (skip payment cycle)") # orchestrator: balance current + await original(orchestrator_url, payment_url=payment_url) + + ps.send_payment = skip_then_pay # type: ignore[assignment] + task = asyncio.create_task(run_session_payments(_session(ps), interval=0.01)) + await asyncio.wait_for(ps.paid.wait(), timeout=1.0) # set on the 2nd cycle, after the skip + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + # The skip did not kill the loop; it kept going and paid on the next cycle. + assert attempts["n"] >= 2 + + asyncio.run(go()) + + +def test_run_session_payments_stops_and_marks_released_on_404() -> None: + async def go() -> None: + ps = _FakePaymentSession() + + async def gone(orchestrator_url: str | None = None, *, payment_url: str | None = None) -> None: + raise _http_error(404) + + ps.send_payment = gone # type: ignore[assignment] + sess = _session(ps, control_url="https://orch/apps/r1/session/sess-1") + # The loop must return on its own (no cancel) once the orchestrator 404s. + await asyncio.wait_for(run_session_payments(sess, interval=0.01), timeout=1.0) + assert sess.released + await asyncio.wait_for(sess.wait_released(), timeout=1.0) + + asyncio.run(go()) + + +def test_run_session_payments_stops_on_409_fixed_price() -> None: + async def go() -> None: + ps = _FakePaymentSession() + + async def fixed(orchestrator_url: str | None = None, *, payment_url: str | None = None) -> None: + raise _http_error(409) + + ps.send_payment = fixed # type: ignore[assignment] + sess = _session(ps, control_url="https://orch/apps/r1/session/sess-1") + await asyncio.wait_for(run_session_payments(sess, interval=0.01), timeout=1.0) + # Fixed-price sessions simply need no more payments; they are not released. + assert not sess.released + + asyncio.run(go()) + + +def test_run_session_payments_stops_on_403_mismatch() -> None: + async def go() -> None: + ps = _FakePaymentSession() + + async def mismatched(orchestrator_url: str | None = None, *, payment_url: str | None = None) -> None: + raise _http_error(403) + + ps.send_payment = mismatched # type: ignore[assignment] + sess = _session(ps, control_url="https://orch/apps/r1/session/sess-1") + # A session/payment mismatch is fatal for the loop but says nothing about + # the session itself, so it must return on its own without marking released. + await asyncio.wait_for(run_session_payments(sess, interval=0.01), timeout=1.0) + assert not sess.released + + asyncio.run(go()) + + +def test_run_session_payments_retries_other_http_errors() -> None: + async def go() -> None: + ps = _FakePaymentSession() + original = ps.send_payment + attempts = {"n": 0} + + async def flaky(orchestrator_url: str | None = None, *, payment_url: str | None = None) -> None: + attempts["n"] += 1 + if attempts["n"] == 1: + raise _http_error(500) + await original(orchestrator_url, payment_url=payment_url) + + ps.send_payment = flaky # type: ignore[assignment] + task = asyncio.create_task(run_session_payments(_session(ps), interval=0.01)) + await asyncio.wait_for(ps.paid.wait(), timeout=1.0) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert attempts["n"] >= 2 + + asyncio.run(go()) + + +def test_start_payments_noop_offchain() -> None: + async def go() -> None: + sess = _session(None) + assert sess.start_payments() is None + assert sess._payment_task is None + + asyncio.run(go()) + + +def test_start_payments_without_running_loop_returns_none() -> None: + # No running loop: logs a warning and skips rather than raising. + sess = _session(_FakePaymentSession()) + assert sess.start_payments() is None + assert sess._payment_task is None + + +def test_start_payments_is_idempotent() -> None: + async def go() -> None: + sess = _session(_FakePaymentSession()) + t1 = sess.start_payments() + t2 = sess.start_payments() + assert t1 is not None + assert t1 is t2 + t1.cancel() + with pytest.raises(asyncio.CancelledError): + await t1 + + asyncio.run(go()) + + +def test_stop_payments_cancels_only_loop() -> None: + async def go() -> None: + sess = _session(_FakePaymentSession()) + task = sess.start_payments() + assert task is not None + with patch.object(live_runner, "stop_runner_session", new=AsyncMock()) as stop: + await sess.stop_payments() + stop.assert_not_awaited() + assert task.done() + assert sess._payment_task is None + # A stopped loop can be restarted (drain / hand-off flows). + restarted = sess.start_payments() + assert restarted is not None and restarted is not task + restarted.cancel() + with pytest.raises(asyncio.CancelledError): + await restarted + + asyncio.run(go()) + + +def test_aclose_cancels_loop_and_stops_session() -> None: + async def go() -> None: + sess = _session(_FakePaymentSession()) + sess.start_payments() + task = sess._payment_task + assert task is not None + with patch.object(live_runner, "stop_runner_session", new=AsyncMock()) as stop: + await sess.aclose() + stop.assert_awaited_once() + assert task.done() + + asyncio.run(go()) + + +def test_aclose_skips_stop_when_released() -> None: + async def go() -> None: + sess = _session(_FakePaymentSession()) + sess._mark_released() + with patch.object(live_runner, "stop_runner_session", new=AsyncMock()) as stop: + await sess.aclose() + stop.assert_not_awaited() + + asyncio.run(go()) + + +def test_async_context_manager_starts_and_stops() -> None: + async def go() -> None: + ps = _FakePaymentSession() + sess = _session(ps) + with patch.object(live_runner, "stop_runner_session", new=AsyncMock()) as stop: + async with sess as entered: + assert entered is sess + await asyncio.wait_for(ps.paid.wait(), timeout=1.0) + assert sess._payment_task is not None + stop.assert_awaited_once() + assert sess._payment_task is None + + asyncio.run(go()) + + +def _challenge_error(payload: dict) -> LivepeerHTTPError: + return LivepeerHTTPError(402, "http://runner/session", body=json.dumps(payload)) + + +def test_parse_challenge_reads_payment_interval_ms() -> None: + challenge = _parse_runner_payment_challenge( + _challenge_error( + { + "payment_params": "params", + "orchestrator": "https://orch", + "manifest_id": "sess-1", + "payment_interval_ms": 5000, + } + ) + ) + assert challenge.payment_interval_s == 5.0 + + +def test_parse_challenge_without_payment_interval_ms() -> None: + challenge = _parse_runner_payment_challenge( + _challenge_error( + { + "payment_params": "params", + "orchestrator": "https://orch", + "manifest_id": "sess-1", + } + ) + ) + assert challenge.payment_interval_s is None + + +_STREAM_CHALLENGE = { + "payment_params": "params", + "orchestrator": "https://orch", + "manifest_id": "sess-1", + "payment_interval_ms": 5000, +} + + +async def _call_stream(payment_unit: str) -> tuple[LiveRunnerCallStream, AsyncMock, _FakePaymentSession]: + """Drive call_runner(stream=True) through a 402 challenge and a paid retry.""" + ps = _FakePaymentSession() + payment = SimpleNamespace(payment="payment-b64", seg_creds="seg-b64") + open_stream = AsyncMock( + side_effect=[ + _challenge_error(_STREAM_CHALLENGE), + (AsyncMock(), SimpleNamespace(status=200, headers={"Content-Type": "text/event-stream"})), + ] + ) + with ( + patch.object( + live_runner, + "get_signer_info", + new=AsyncMock(return_value=SimpleNamespace(address="0xPayer")), + ), + patch.object(live_runner, "_get_runner_payment", new=AsyncMock(return_value=(ps, payment))), + patch.object(live_runner, "open_stream", new=open_stream), + ): + stream = await call_runner( + "https://orch/apps/r1/app", + stream=True, + signer_url="https://signer", + payment_unit=payment_unit, + ) + return stream, open_stream, ps + + +def test_call_runner_stream_carries_payment_semantics() -> None: + async def go() -> None: + stream, open_stream, ps = await _call_stream("seconds") + # The paid retry carries the payment material on the streaming request. + headers = open_stream.call_args.kwargs["headers"] + assert headers["Livepeer-Payment"] == "payment-b64" + assert headers["Livepeer-Segment"] == "seg-b64" + # Stream results mirror the JSON path: challenge manifest id and cadence. + assert stream.session_id == "sess-1" + assert stream.server_payment_interval == 5.0 + assert stream.payment_session is ps + + asyncio.run(go()) + + +def test_call_runner_stream_fixed_price_drops_payment_session() -> None: + async def go() -> None: + stream, _, _ = await _call_stream("fixed") + # Fixed-price pays once inline; no session to hand to a payment loop. + assert stream.session_id == "sess-1" + assert stream.payment_session is None + + asyncio.run(go()) + + +class _FakeCursor: + def __init__(self, result: LiveRunnerCallResult) -> None: + self._result = result + + async def next(self) -> LiveRunnerCallResult: + return self._result + + +def _reserve_result(ps: _FakePaymentSession | None, server_payment_interval: float | None = None) -> LiveRunnerCallResult: + return LiveRunnerCallResult( + { + "session_id": "sess-1", + "app_url": "https://orch/apps/r1/session/sess-1/app", + "control_url": "https://orch/apps/r1/session/sess-1", + }, + runner_url="https://orch/apps/r1/session", + payment_session=ps, # type: ignore[arg-type] + server_payment_interval=server_payment_interval, + ) + + +def test_reserve_session_auto_starts_payments_and_parses_control_url() -> None: + async def go() -> None: + ps = _FakePaymentSession() + cursor = _FakeCursor(_reserve_result(ps)) + with patch.object(selection, "runner_selector", new=AsyncMock(return_value=cursor)): + sess = await reserve_session(signer_url="https://signer") + assert sess.control_url == "https://orch/apps/r1/session/sess-1" + assert sess._payment_task is not None + await asyncio.wait_for(ps.paid.wait(), timeout=1.0) + assert ps.payment_urls[0] == "https://orch/apps/r1/session/sess-1/payment" + await sess.stop_payments() + + asyncio.run(go()) + + +def test_reserve_session_auto_pay_false_does_not_start() -> None: + async def go() -> None: + cursor = _FakeCursor(_reserve_result(_FakePaymentSession())) + with patch.object(selection, "runner_selector", new=AsyncMock(return_value=cursor)): + sess = await reserve_session(signer_url="https://signer", auto_pay=False) + assert sess._payment_task is None + + asyncio.run(go()) + + +def test_reserve_session_derives_interval_from_challenge() -> None: + async def go() -> None: + cursor = _FakeCursor(_reserve_result(_FakePaymentSession(), server_payment_interval=5.0)) + with patch.object(selection, "runner_selector", new=AsyncMock(return_value=cursor)): + sess = await reserve_session(signer_url="https://signer", auto_pay=False) + # 60% of the orchestrator's 5s debit tick. + assert sess.payment_interval == pytest.approx(3.0) + + asyncio.run(go()) + + +def test_reserve_session_explicit_interval_wins() -> None: + async def go() -> None: + cursor = _FakeCursor(_reserve_result(_FakePaymentSession(), server_payment_interval=5.0)) + with patch.object(selection, "runner_selector", new=AsyncMock(return_value=cursor)): + sess = await reserve_session( + signer_url="https://signer", + payment_interval=1.5, + auto_pay=False, + ) + assert sess.payment_interval == 1.5 + + asyncio.run(go()) diff --git a/uv.lock b/uv.lock index 9bc317a..36694fd 100644 --- a/uv.lock +++ b/uv.lock @@ -161,6 +161,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/83/41/7f13361db54d7e02f11552575c0384dadaf0918138f4eaa82ea03a9f9580/av-16.1.0-cp314-cp314t-win_amd64.whl", hash = "sha256:6f90dc082ff2068ddbe77618400b44d698d25d9c4edac57459e250c16b33d700", size = 31948164, upload-time = "2026-01-11T09:59:19.501Z" }, ] +[[package]] +name = "colorama" +version = "0.4.6" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697, upload-time = "2022-10-25T02:36:22.414Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, +] + [[package]] name = "frozenlist" version = "1.8.0" @@ -343,6 +352,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0e/61/66938bbb5fc52dbdf84594873d5b51fb1f7c7794e9c0f5bd885f30bc507b/idna-3.11-py3-none-any.whl", hash = "sha256:771a87f49d9defaf64091e6e6fe9c18d4833f140bd19464795bc32d966ca37ea", size = 71008, upload-time = "2025-10-12T14:55:18.883Z" }, ] +[[package]] +name = "iniconfig" +version = "2.3.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, +] + [[package]] name = "livepeer-gateway" version = "0.1.0" @@ -357,6 +375,7 @@ dependencies = [ [package.optional-dependencies] dev = [ { name = "grpcio-tools" }, + { name = "pytest" }, ] examples = [ { name = "numpy" }, @@ -372,6 +391,7 @@ requires-dist = [ { name = "numpy", marker = "extra == 'examples'", specifier = ">=2.2.6" }, { name = "opencv-python-headless", marker = "extra == 'examples'", specifier = ">=4.13.0.90" }, { name = "protobuf", specifier = ">=4.25.0" }, + { name = "pytest", marker = "extra == 'dev'", specifier = ">=8.0" }, ] provides-extras = ["dev", "examples"] @@ -553,6 +573,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/4a/90/b338326131ccb2aaa3c2c85d00f41822c0050139a4bfe723cfd95455bd2d/opencv_python_headless-4.13.0.92-cp37-abi3-win_amd64.whl", hash = "sha256:77a82fe35ddcec0f62c15f2ba8a12ecc2ed4207c17b0902c7a3151ae29f37fb6", size = 40070414, upload-time = "2026-02-05T07:02:26.448Z" }, ] +[[package]] +name = "packaging" +version = "26.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d7/f1/e7a6dd94a8d4a5626c03e4e99c87f241ba9e350cd9e6d75123f992427270/packaging-26.2.tar.gz", hash = "sha256:ff452ff5a3e828ce110190feff1178bb1f2ea2281fa2075aadb987c2fb221661", size = 228134, upload-time = "2026-04-24T20:15:23.917Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/df/b2/87e62e8c3e2f4b32e5fe99e0b86d576da1312593b39f47d8ceef365e95ed/packaging-26.2-py3-none-any.whl", hash = "sha256:5fc45236b9446107ff2415ce77c807cee2862cb6fac22b8a73826d0693b0980e", size = 100195, upload-time = "2026-04-24T20:15:22.081Z" }, +] + +[[package]] +name = "pluggy" +version = "1.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" }, +] + [[package]] name = "propcache" version = "0.4.1" @@ -652,6 +690,31 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0e/15/4f02896cc3df04fc465010a4c6a0cd89810f54617a32a70ef531ed75d61c/protobuf-6.33.2-py3-none-any.whl", hash = "sha256:7636aad9bb01768870266de5dc009de2d1b936771b38a793f73cbbf279c91c5c", size = 170501, upload-time = "2025-12-06T00:17:52.211Z" }, ] +[[package]] +name = "pygments" +version = "2.20.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/c3/b2/bc9c9196916376152d655522fdcebac55e66de6603a76a02bca1b6414f6c/pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f", size = 4955991, upload-time = "2026-03-29T13:29:33.898Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" }, +] + +[[package]] +name = "pytest" +version = "9.1.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "iniconfig" }, + { name = "packaging" }, + { name = "pluggy" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e4/47/b9efed96c114afcfa3c9d3fe98a76a1d14c74a9e266d397cf6eb64be5e01/pytest-9.1.1.tar.gz", hash = "sha256:1088fbde8f2b49d95a549a195707afa7a76a3ce9bcadc26b6d71f0ffda5fe313", size = 1636369, upload-time = "2026-06-19T10:58:32.857Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" }, +] + [[package]] name = "setuptools" version = "80.9.0"