diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9efbb27ca..4f438a43c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -80,7 +80,8 @@ jobs: - name: Run e2e tests run: | source .venv/bin/activate - pytest tests/e2e/ -v -p no:xdist --timeout=600 --reruns 1 --reruns-delay 5 --log-cli-level=INFO -o "addopts=" + # sandbox cloud checks run explicitly against an authorized dev environment. + pytest tests/e2e/ --ignore=tests/e2e/test_sandbox.py -v -p no:xdist --timeout=600 --reruns 1 --reruns-delay 5 --log-cli-level=INFO -o "addopts=" env: # the e2e provisioner still drives runpod-flash, whose import-time # sunset FutureWarning is an error under pytest.ini filterwarnings diff --git a/CHANGELOG.md b/CHANGELOG.md index be9484adf..43526f669 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,11 @@ ### Features * Apps expose `NetworkVolume` and `GlobalVolume` resources with explicit path-keyed mounts, lazy create-if-missing provisioning, and execution-bound filesystem access. Global volumes use GraphQL for lookup and creation. +* Sandboxes support shared storage references, structured ports, execution details, command timeouts, detached execution, and lifetime updates through synchronous and asynchronous handles. + +### Fixes + +* Sandbox ownership cleanup survives interrupted handle delivery and overlapping context entry. Concurrent requests settle during loop shutdown, and failed creation preserves recoverable sandbox IDs and cleanup errors. ### Bug Fixes diff --git a/README.md b/README.md index 6f5a3fc4d..0bf5e98f6 100644 --- a/README.md +++ b/README.md @@ -73,6 +73,107 @@ rp flash dev main.py # live dev session: edit, re-run, logs stream back rp flash deploy # deploy production endpoints ``` +## Sandboxes + +Sandboxes are an access-controlled preview. An authorized account and a +sandbox-enabled API environment are required. Configure its API origin with +`RUNPOD_API_BASE_URL` or the handle's `base_url` option. Authenticate with +`RUNPOD_API_KEY`, `runpod.api_key`, or a handle-specific `api_key`. + +Supply exactly one of `image` or `template_id`. Construction is local; entering +an owned context creates the sandbox and leaving it terminates remote compute. + +```python +from runpod import Sandbox + +with Sandbox( + image="python:3.12-slim", + cmd=["sleep", "infinity"], + disk_gb=20, + idle_timeout_seconds=300, + max_lifetime_seconds=1200, +) as sandbox: + result = sandbox.exec(["python", "-c", "print('hello from a sandbox')"], check=True) + print(result.output) + sandbox.extend(seconds=300) + sandbox.update(idle_timeout_seconds=600) +``` + +Use `AsyncioSandbox` with awaited factory, execution, and lifecycle methods: + +```python +import asyncio +from runpod import AsyncioSandbox + +async def main(): + async with AsyncioSandbox(image="python:3.12-slim", cmd=["sleep", "infinity"]) as sandbox: + result = await sandbox.exec(["python", "-c", "print('hello')"], check=True) + print(result.output) + +asyncio.run(main()) +``` + +Create options include: + +- `mounts={"/data": NetworkVolume("volume-id", create=False), + "/models": GlobalVolume("global-volume-id")}` for one volume of each kind. + Paths must be absolute and non-overlapping. Network storage pins placement to + its datacenter; `data_center_ids` must include that datacenter if supplied. + Creating a named network volume outside an app requires its explicit + `datacenter`. Sandbox contexts never mount remote storage on the client. +- `ports={3000: "http", 3001: "tcp", 53: ("tcp", "udp")}` for exposed ports. + TCP and UDP exposure depends on backend public-port support. +- `cmd` and `entrypoint` as argument lists, `registry_auth` as a saved credential + name or ID, and `start_ssh=True` for SSH setup. SSH also requires registered + keys and an image supporting that setup. + +Omitted fields stay omitted. The SDK sends `mounts={}`, `ports={}`, and empty +command lists explicitly, preserving the distinction from absent configuration. +The container stops when its main command exits. + +`Sandbox.create(...)` returns an owned handle. `get(...)` and `list(...)` return +borrowed handles whose contexts only close local connections. `terminate()` +releases remote compute; `close()` releases local resources and permits reuse. +Each handle permits one active context; overlapping entry is rejected. + +Failed creation triggers bounded cleanup. `SandboxCreationError`, exported from +`runpod.sandbox`, preserves the backend's `sandbox_id`, HTTP status, and error +details. If termination also fails, the cleanup error is chained as `__cause__`; +the ID remains available for explicit recovery with `get(...)` and `terminate()`. + +`info`, `state`, `compute`, and `expires_at` are cached snapshots. `refresh()` +fetches a new snapshot, including mount, port, SSH, and placement metadata. +Unavailable configuration remains `None`, distinct from empty configuration. + +`idle_timeout_seconds` measures inactivity; `max_lifetime_seconds` measures total +age from creation. `update(...)` changes only supplied lifetime settings. +`extend(seconds=...)` adds to the current lifetime after refreshing it. Calls +through one handle are serialized; independent clients can race because the +backend extension contract accepts an absolute lifetime. + +`exec` returns combined output plus available `stdout`, `stderr`, `exit_code`, +`duration_ms`, and `truncated` details. `check=True` raises +`SandboxExecutionError` on a reported command failure, retaining the result. +Only explicit `sandbox_starting` conflicts are retried within `startup_timeout`. +Other conflicts, malformed responses, and ambiguous transport failures propagate. + +`exec(..., timeout_seconds=30)` requests a command timeout from 1 to 50 seconds. +The backend currently caps foreground execution at about four seconds. +`request_timeout` separately limits the client HTTP wait. +For longer work, use `background=True` and poll for completion. A successful +detached result confirms that the command started, not that it finished; redirect +output to a file to retrieve it afterward. These options require backend support +for `timeoutSeconds` and `background`. + +`logs(source="container", tail=10)` streams main-process logs; `source="system"` +selects lifecycle events. Exec output is returned by `exec`. Use `with` and +regular iteration for synchronous streams, or `async with` and `async for` for +asynchronous streams. Save an event's `id` as `last_event_id` to resume, or filter +with `since`. + +These features require backend support for the sandbox create, exec, and lifetime +contracts, plus compatible SDK/runtime releases for apps worker mount access. + ## Contributing Pull requests and issues are welcome — see the [contributing guide](CONTRIBUTING.md) to get started. diff --git a/runpod/__init__.py b/runpod/__init__.py index 427023537..59247a3e4 100644 --- a/runpod/__init__.py +++ b/runpod/__init__.py @@ -44,6 +44,8 @@ "AsyncioEndpoint": "runpod.endpoint", "AsyncioJob": "runpod.endpoint", "Endpoint": "runpod.endpoint", + "AsyncioSandbox": "runpod.sandbox", + "Sandbox": "runpod.sandbox", # apps surface "Api": "runpod.apps", "App": "runpod.apps", diff --git a/runpod/api/rest.py b/runpod/api/rest.py index 5f77b6190..e1ee375b6 100644 --- a/runpod/api/rest.py +++ b/runpod/api/rest.py @@ -29,8 +29,10 @@ def _resolve_api_key(api_key: Optional[str]) -> str: return effective_api_key -def _build_url(path: str) -> str: - api_url_base = os.environ.get("RUNPOD_API_BASE_URL", "https://api.runpod.io") +def _build_url(path: str, base_url: Optional[str] = None) -> str: + api_url_base = base_url or os.environ.get( + "RUNPOD_API_BASE_URL", "https://api.runpod.io" + ) return f"{api_url_base.rstrip('/')}/{path.lstrip('/')}" @@ -43,7 +45,14 @@ def _build_headers(api_key: str) -> dict[str, str]: } -def _raise_for_error(status_code: int, method: str, path: str, text: str = "") -> None: +def _raise_for_status( + status_code: int, + payload: Mapping[str, Any], + text: str, + method: str, + path: str, +) -> None: + """map http error details consistently across rest transports.""" if status_code == HTTP_STATUS_UNAUTHORIZED: raise error.AuthenticationError( "Unauthorized request, please check your API key." @@ -52,12 +61,6 @@ def _raise_for_error(status_code: int, method: str, path: str, text: str = "") - if status_code < HTTP_STATUS_BAD_REQUEST: return - try: - payload = json_module.loads(text) - except ValueError: - payload = {} - if not isinstance(payload, dict): - payload = {} message = payload.get("detail") or payload.get("title") if not message: message = text or f"Request failed with status {status_code}" @@ -70,6 +73,20 @@ def _raise_for_error(status_code: int, method: str, path: str, text: str = "") - ) +def _raise_for_error(status_code: int, method: str, path: str, text: str = "") -> None: + if status_code == HTTP_STATUS_UNAUTHORIZED: + _raise_for_status(status_code, {}, "", method, path) + if status_code < HTTP_STATUS_BAD_REQUEST: + return + try: + payload = json_module.loads(text) + except ValueError: + payload = {} + _raise_for_status( + status_code, payload if isinstance(payload, dict) else {}, text, method, path + ) + + def run_rest_request( method: str, path: str, diff --git a/runpod/api/sandboxes.py b/runpod/api/sandboxes.py new file mode 100644 index 000000000..80ead99a5 --- /dev/null +++ b/runpod/api/sandboxes.py @@ -0,0 +1,565 @@ +"""Asynchronous REST and streaming transport for CPU sandboxes.""" + +from __future__ import annotations + +import asyncio +import codecs +import json +import math +import re +from collections.abc import AsyncIterator, Mapping, Sequence +from typing import Any +from urllib.parse import quote, urlencode + +import aiohttp +from yarl import URL + +from runpod import error +from runpod.api.rest import ( + _build_headers, + _build_url, + _raise_for_status, + _resolve_api_key, +) + +_SANDBOX_PATH = "/v2/sandboxes" +_LINE_END = re.compile(r"\r\n|\r|\n") + + +class SandboxCreationError(error.QueryError): + """a failed creation with a recoverable, potentially billable sandbox id.""" + + def __init__(self, sandbox_id: str, cause: error.QueryError) -> None: + super().__init__( + cause.message, + cause.query, + status_code=cause.status_code, + errors=cause.errors, + ) + self.sandbox_id = sandbox_id + + +class SandboxConflictError(error.QueryError): + """a structured rejection whose code determines whether startup can be retried.""" + + def __init__(self, payload: Mapping[str, Any], method: str, path: str) -> None: + super().__init__( + payload.get("detail") or payload.get("title") or "Sandbox conflict", + f"{method} {path}", + status_code=409, + errors=payload.get("errors"), + ) + self.code = payload.get("code") + + +def _sandbox_path(sandbox_id: str) -> str: + return f"{_SANDBOX_PATH}/{quote(sandbox_id, safe='')}" + + +async def _raise_response_error( + response: aiohttp.ClientResponse, method: str, path: str +) -> None: + if response.status == 401: + _raise_for_status(response.status, {}, "", method, path) + if 200 <= response.status < 300: + return + raw = await response.read() + try: + payload = json.loads(raw) + except (ValueError, UnicodeError): + payload = {} + if not isinstance(payload, dict): + payload = {} + sandbox_id = ( + payload.get("sandboxId") + if method == "POST" and path == _SANDBOX_PATH + else None + ) + if not isinstance(sandbox_id, str) or not sandbox_id: + sandbox_id = None + if response.status == 409 and sandbox_id is None: + raise SandboxConflictError(payload, method, path) + try: + text = raw.decode(response.charset or "utf-8", errors="replace") + except LookupError: + text = raw.decode("utf-8", errors="replace") + try: + _raise_for_status(response.status, payload, text, method, path) + except error.QueryError as cause: + if sandbox_id is not None: + raise SandboxCreationError(sandbox_id, cause) from cause + raise + # Redirects are deliberately not followed, including on read operations. + raise error.QueryError( + text or f"Unexpected HTTP status {response.status}", + f"{method} {path}", + status_code=response.status, + errors=payload.get("errors"), + ) + + +async def _sse_lines(content: aiohttp.StreamReader) -> AsyncIterator[str]: + """Decode only the current line, allowing arbitrary UTF-8/chunk boundaries.""" + decoder = codecs.getincrementaldecoder("utf-8-sig")() + pending = "" + skip_lf = False + async for chunk in content.iter_chunked(8192): + text = decoder.decode(chunk) + if not text: + continue + if skip_lf: + if text.startswith("\n"): + text = text[1:] + skip_lf = False + text = pending + text + start = 0 + for match in _LINE_END.finditer(text): + yield text[start : match.start()] + start = match.end() + skip_lf = text.endswith("\r") + pending = text[start:] + pending += decoder.decode(b"", final=True) + if pending: + yield pending + + +class AsyncSandboxLogStream: + """One SSE connection; use an async context or ``aclose`` on early exit. + + ``open`` performs the HTTP handshake without waiting for a log event. + Exhaustion, cancellation, parsing errors and server timeout frames close + the response. No reconnect is attempted, even when a cursor is supplied. + """ + + def __init__( + self, + api: AsyncSandboxAPI, + path: str, + params: list[tuple[str, str]], + last_event_id: str | None, + ) -> None: + self._api = api + self._path = path + self._params = params + self._last_event_id = last_event_id + self._response: aiohttp.ClientResponse | None = None + self._open_task: asyncio.Task[None] | None = None + self._iterator: AsyncIterator[dict[str, Any]] | None = None + self._closed = False + + async def _perform_open(self) -> None: + try: + if self._closed: + raise RuntimeError("Sandbox log stream is closed") + headers = self._api._headers() + headers["Accept"] = "text/event-stream" + if self._last_event_id is not None: + headers["Last-Event-ID"] = self._last_event_id + self._response = await self._api._get_session().request( + "GET", + self._api._url(self._path, self._params), + headers=headers, + timeout=aiohttp.ClientTimeout( + total=None, + connect=self._api.request_timeout, + sock_connect=self._api.request_timeout, + sock_read=None, + ), + allow_redirects=False, + ) + await _raise_response_error(self._response, "GET", self._path) + if self._response.content_type != "text/event-stream": + raise error.QueryError( + "Expected a text/event-stream sandbox log response", + f"GET {self._path}", + status_code=self._response.status, + ) + self._iterator = self._events(self._response) + except BaseException: + self._dispose() + raise + + async def open(self) -> AsyncSandboxLogStream: + """Open once, with a bounded handshake but no total stream timeout.""" + if self._closed: + raise RuntimeError("Sandbox log stream is closed") + if self._open_task is None: + self._open_task = asyncio.create_task( + asyncio.wait_for( + self._perform_open(), timeout=self._api.request_timeout + ) + ) + try: + await self._open_task + except BaseException: + await self.aclose() + raise + return self + + def _dispose(self) -> None: + self._closed = True + if self._response is not None: + self._response.close() + self._response = None + self._iterator = None + self._api._streams.discard(self) + + async def aclose(self) -> None: + """Release this stream, including a handshake still in progress.""" + self._dispose() + if self._open_task is not None and not self._open_task.done(): + self._open_task.cancel() + await asyncio.gather(self._open_task, return_exceptions=True) + + async def __aenter__(self) -> AsyncSandboxLogStream: + return await self.open() + + async def __aexit__(self, exc_type, exc_value, traceback) -> None: + await self.aclose() + + def __aiter__(self) -> AsyncSandboxLogStream: + return self + + async def __anext__(self) -> dict[str, Any]: + if self._closed: + raise StopAsyncIteration + try: + await self.open() + assert self._iterator is not None + return await anext(self._iterator) + except BaseException: + await self.aclose() + raise + + async def _events( + self, response: aiohttp.ClientResponse + ) -> AsyncIterator[dict[str, Any]]: + data: list[str] = [] + event_type = "" + event_id: str | None = None + async for line in _sse_lines(response.content): + if not line: + if event_type == "timeout": + return + if event_type not in ("", "message"): + raise error.QueryError( + f"Sandbox log event {event_type!r}: " + "\n".join(data), + f"GET {self._path}", + ) + if data: + payload = json.loads("\n".join(data)) + if not isinstance(payload, dict): + raise ValueError("Sandbox log event must be a JSON object") + yield { + "source": payload["source"], + "line": payload["line"], + "ts": payload["ts"], + "id": event_id, + } + data = [] + event_type = "" + continue + if line.startswith(":"): + continue + field, separator, value = line.partition(":") + if separator and value.startswith(" "): + value = value[1:] + if field == "data": + data.append(value) + elif field == "event": + event_type = value + elif field == "id" and "\x00" not in value: + event_id = value + # SSE retry directives are intentionally ignored: no reconnects. + # Per SSE, EOF does not dispatch an event lacking its blank-line boundary. + + +class AsyncSandboxAPI: + """Lazy, owned aiohttp transport. Each operation makes one HTTP request. + + ``close`` releases streams and sockets, but a later request can open a new + session. Use the instance from one event loop while its session is active. + Readiness and retry policies belong to the higher-level sandbox handles. + """ + + def __init__( + self, + api_key: str | None = None, + base_url: str | None = None, + request_timeout: float = 30, + ) -> None: + if not math.isfinite(request_timeout) or request_timeout <= 0: + raise ValueError("request_timeout must be a finite positive number") + self.api_key = api_key + self.base_url = base_url + self.request_timeout = request_timeout + self._session: aiohttp.ClientSession | None = None + self._streams: set[AsyncSandboxLogStream] = set() + + def _headers(self) -> dict[str, str]: + return _build_headers(_resolve_api_key(self.api_key)) + + def _url(self, path: str, params: list[tuple[str, str]] | None = None) -> URL: + url = _build_url(path, self.base_url) + if params: + url += "?" + urlencode(params, safe="=") + # Preserve quoted ID segments (including encoded slashes) verbatim. + return URL(url, encoded=True) + + def _get_session(self) -> aiohttp.ClientSession: + if self._session is None or self._session.closed: + self._session = aiohttp.ClientSession( + timeout=aiohttp.ClientTimeout(total=self.request_timeout), + cookie_jar=aiohttp.DummyCookieJar(), + ) + # aiohttp otherwise transparently replays disconnected GET/DELETE + # requests. it currently exposes no public switch for this policy. + if hasattr(self._session, "_retry_connection"): + self._session._retry_connection = False + return self._session + + async def _request( + self, + method: str, + path: str, + *, + params: list[tuple[str, str]] | None = None, + body: Mapping[str, Any] | None = None, + ) -> dict[str, Any] | None: + headers = self._headers() + async with self._get_session().request( + method, + self._url(path, params), + headers=headers, + json=body, + allow_redirects=False, + ) as response: + await _raise_response_error(response, method, path) + if response.status == 204: + return None + try: + payload = await response.json(content_type=None) + except ValueError as cause: + raise error.QueryError( + "Sandbox API response is not valid JSON", + f"{method} {path}", + status_code=response.status, + ) from cause + if not isinstance(payload, dict): + raise error.QueryError( + "Sandbox API response must be a JSON object", + f"{method} {path}", + status_code=response.status, + ) + return payload + + async def _list_volumes(self, path: str, collection: str) -> list[dict[str, Any]]: + records = [] + params = [] + seen_cursors = set() + while True: + payload = await self._request("GET", path, params=params) + page = payload.get(collection) if isinstance(payload, dict) else None + if not isinstance(page, list) or any( + not isinstance(record, dict) for record in page + ): + raise error.QueryError( + f"invalid {collection} list response", f"GET {path}" + ) + pagination = payload.get("pagination") + if not isinstance(pagination, dict) or not isinstance( + pagination.get("hasNextPage"), bool + ): + raise error.QueryError( + f"invalid {collection} pagination response", f"GET {path}" + ) + records.extend(page) + if not pagination["hasNextPage"]: + return records + cursor = pagination.get("nextCursor") + if not isinstance(cursor, str) or not cursor or cursor in seen_cursors: + raise error.QueryError( + f"{collection} pagination requires a unique nextCursor", + f"GET {path}", + ) + seen_cursors.add(cursor) + params = [("cursor", cursor)] + + async def list_global_volumes(self) -> list[dict[str, Any]]: + return await self._list_volumes("/v2/global-volumes", "globalVolumes") + + async def create_global_volume(self, name: str) -> dict[str, Any]: + payload = await self._request("POST", "/v2/global-volumes", body={"name": name}) + if ( + not isinstance(payload, dict) + or not isinstance(payload.get("id"), str) + or not payload["id"] + ): + raise error.QueryError( + "global volume creation returned an invalid response; " + "the volume may have been created, check existing volumes before retrying", + "POST /v2/global-volumes", + ) + return payload + + async def list_network_volumes(self) -> list[dict[str, Any]]: + """list storage through this handle's credentials and control-plane URL.""" + return await self._list_volumes("/v2/network-volumes", "networkVolumes") + + async def network_volume_datacenters(self) -> set[str]: + payload = await self._request("GET", "/v2/catalog/datacenters") + datacenters = payload.get("dataCenters") if isinstance(payload, dict) else None + if not isinstance(datacenters, list): + raise error.QueryError("datacenter catalog is missing dataCenters") + supported = set() + for dc in datacenters: + if ( + not isinstance(dc, dict) + or not isinstance(dc.get("id"), str) + or not dc["id"] + or not isinstance(dc.get("networkVolumeTypes"), list) + or any( + not isinstance(tier, str) or not tier + for tier in dc["networkVolumeTypes"] + ) + ): + raise error.QueryError( + "datacenter catalog has invalid networkVolumeTypes" + ) + if dc["networkVolumeTypes"]: + supported.add(dc["id"]) + return supported + + async def create_network_volume( + self, name: str, size: int, data_center_id: str + ) -> dict[str, Any]: + payload = await self._request( + "POST", + "/v2/network-volumes", + body={"name": name, "size": size, "dataCenter": data_center_id}, + ) + if payload is None: + raise ValueError("Network volume creation returned no response") + return payload + + async def list_registry_auths(self) -> list[dict[str, Any]]: + payload = await self._request("GET", "/v2/registries") + if payload is None: + raise ValueError("Registry listing returned no response") + return payload["registries"] + + async def create(self, body: Mapping[str, Any]) -> dict[str, Any]: + payload = await self._request( + "POST", + _SANDBOX_PATH, + body={key: value for key, value in body.items() if value is not None}, + ) + if payload is None: + raise ValueError("Sandbox creation returned no snapshot") + return payload + + async def list( + self, + *, + state: str | None = None, + labels: Mapping[str, str] | None = None, + ) -> list[dict[str, Any]]: + params = [] + if state is not None: + params.append(("state", state)) + if labels is not None: + params.extend(("labels", f"{key}={value}") for key, value in labels.items()) + payload = await self._request("GET", _SANDBOX_PATH, params=params) + if payload is None: + raise ValueError("Sandbox listing returned no response") + sandboxes = payload["sandboxes"] + if not isinstance(sandboxes, list) or any( + not isinstance(item, dict) for item in sandboxes + ): + raise ValueError("Sandbox listing must contain an array of objects") + return sandboxes + + async def get(self, sandbox_id: str) -> dict[str, Any]: + payload = await self._request("GET", _sandbox_path(sandbox_id)) + if payload is None: + raise ValueError("Sandbox lookup returned no snapshot") + return payload + + async def update(self, sandbox_id: str, body: Mapping[str, Any]) -> dict[str, Any]: + payload = await self._request("PATCH", _sandbox_path(sandbox_id), body=body) + if payload is None: + raise ValueError("Sandbox update returned no snapshot") + return payload + + async def extend( + self, sandbox_id: str, max_lifetime_seconds: int + ) -> dict[str, Any]: + payload = await self._request( + "POST", + f"{_sandbox_path(sandbox_id)}/extend", + body={"maxLifetimeSeconds": max_lifetime_seconds}, + ) + if payload is None: + raise ValueError("Sandbox extension returned no snapshot") + return payload + + async def terminate(self, sandbox_id: str) -> None: + await self._request("DELETE", _sandbox_path(sandbox_id)) + + async def exec( + self, + sandbox_id: str, + command: Sequence[str], + *, + timeout_seconds: int | None = None, + background: bool = False, + ) -> dict[str, Any]: + path = f"{_sandbox_path(sandbox_id)}/exec" + body: dict[str, Any] = {"command": list(command)} + if timeout_seconds is not None: + body["timeoutSeconds"] = timeout_seconds + if background: + body["background"] = True + payload = await self._request("POST", path, body=body) + if payload is None or not isinstance(payload.get("output"), str): + raise error.QueryError( + "Sandbox execution response must contain a string 'output'", + f"POST {path}", + ) + return payload + + def logs( + self, + sandbox_id: str, + *, + source: str | None = None, + tail: int | None = None, + since: str | None = None, + last_event_id: str | None = None, + ) -> AsyncSandboxLogStream: + params = [ + (key, str(value)) + for key, value in (("source", source), ("tail", tail), ("since", since)) + if value is not None + ] + stream = AsyncSandboxLogStream( + self, f"{_sandbox_path(sandbox_id)}/logs", params, last_event_id + ) + self._streams.add(stream) + return stream + + async def close(self) -> None: + """Close all active streams and the owned session; allow later reuse.""" + streams = tuple(self._streams) + session, self._session = self._session, None + # Dispose every response before the first await, including on cancellation. + for stream in streams: + stream._dispose() + if stream._open_task is not None and not stream._open_task.done(): + stream._open_task.cancel() + try: + for stream in streams: + await stream.aclose() + finally: + if session is not None: + await session.close() diff --git a/runpod/apps/volume.py b/runpod/apps/volume.py index 943c80b51..25d65c153 100644 --- a/runpod/apps/volume.py +++ b/runpod/apps/volume.py @@ -331,7 +331,11 @@ def validate_mounts(mounts: Mapping[str, Volume], kind: str, is_cpu: bool) -> No if kind in ("queue", "api") and mounts: if len(mounts) != 1 or next(iter(mounts)) != str(ENDPOINT_MOUNT_PATH): raise VolumeError("endpoints support one volume mounted at /runpod-volume") - if is_cpu and any(isinstance(volume, GlobalVolume) for volume in mounts.values()): + if ( + kind != "sandbox" + and is_cpu + and any(isinstance(volume, GlobalVolume) for volume in mounts.values()) + ): raise VolumeError("global volumes require gpu compute") diff --git a/runpod/sandbox/__init__.py b/runpod/sandbox/__init__.py new file mode 100644 index 000000000..f6d2fd128 --- /dev/null +++ b/runpod/sandbox/__init__.py @@ -0,0 +1,34 @@ +"""Sync and async sandbox lifecycles, execution results, and log streams.""" + +from runpod.api.sandboxes import SandboxCreationError + +from .asyncio import AsyncioSandbox, AsyncioSandboxLogStream +from .models import ( + ExecResult, + LogEvent, + LogSource, + SandboxCompute, + SandboxExecutionError, + SandboxInfo, + SandboxStartupTimeout, + SandboxState, + SandboxStateError, +) +from .sync import Sandbox, SandboxLogs + +__all__ = [ + "AsyncioSandbox", + "AsyncioSandboxLogStream", + "Sandbox", + "SandboxLogs", + "ExecResult", + "LogEvent", + "LogSource", + "SandboxCompute", + "SandboxExecutionError", + "SandboxCreationError", + "SandboxInfo", + "SandboxStartupTimeout", + "SandboxState", + "SandboxStateError", +] diff --git a/runpod/sandbox/asyncio.py b/runpod/sandbox/asyncio.py new file mode 100644 index 000000000..e1c602791 --- /dev/null +++ b/runpod/sandbox/asyncio.py @@ -0,0 +1,729 @@ +"""Asynchronous sandbox handles, lifecycle management, and typed log streams.""" + +import asyncio +import math +from collections.abc import Awaitable, Callable, Mapping, Sequence +from dataclasses import replace +from datetime import datetime +from types import TracebackType +from typing import Any, Literal, Optional, TypeVar + +from runpod.api.sandboxes import ( + AsyncSandboxAPI, + SandboxConflictError, + SandboxCreationError, +) +from runpod.apps.registry import resolve_registry_auth +from runpod.apps.volume import Volume, VolumeResolver, normalize_mounts, validate_mounts +from runpod.error import QueryError +from runpod.sandbox.models import ( + ExecResult, + LogEvent, + LogSource, + SandboxCompute, + SandboxExecutionError, + SandboxInfo, + SandboxStartupTimeout, + SandboxState, + SandboxStateError, +) + +_T = TypeVar("_T") + + +def _timeout(value: float, name: str, *, allow_zero: bool = False) -> float: + if not math.isfinite(value) or value < 0 or (value == 0 and not allow_zero): + raise ValueError( + f"{name} must be finite and {'nonnegative' if allow_zero else 'positive'}" + ) + return value + + +def _string_list( + value: Optional[Sequence[str]], name: str, *, nonempty: bool = False +) -> Optional[list[str]]: + if value is None: + return None + if isinstance(value, (str, bytes)) or not isinstance(value, Sequence): + raise TypeError(f"{name} must be a sequence of strings") + if any(not isinstance(item, str) for item in value): + raise TypeError(f"{name} must be a sequence of strings") + if nonempty and (not value or any(not item for item in value)): + raise ValueError(f"{name} must contain at least one nonempty string") + return list(value) + + +def _ports( + ports: Optional[Mapping[int, str | tuple[str, ...]]], +) -> Optional[list[dict[str, Any]]]: + if ports is None: + return None + if not isinstance(ports, Mapping): + raise TypeError("ports must map port numbers to protocols") + rows = [] + for port, protocols in ports.items(): + if ( + isinstance(port, bool) + or not isinstance(port, int) + or not 1 <= port <= 65535 + ): + raise ValueError("port numbers must be integers from 1 to 65535") + if isinstance(protocols, str): + protocols = (protocols,) + if not isinstance(protocols, tuple) or not protocols: + raise TypeError("port protocols must be a string or nonempty tuple") + transports = set() + for protocol in protocols: + if protocol not in ("http", "tcp", "udp"): + raise ValueError("port protocols must be http, tcp, or udp") + transport = "udp" if protocol == "udp" else "tcp" + if transport in transports: + raise ValueError(f"port {port} has overlapping protocols") + transports.add(transport) + rows.append({"port": port, "protocol": protocol}) + if len(rows) > 15: + raise ValueError("sandboxes support at most 15 exposed ports") + return rows + + +def _sandbox_mounts( + mounts: Optional[Mapping[str, Volume]], +) -> Optional[dict[str, Volume]]: + if mounts is None: + return None + normalized = normalize_mounts(mounts) + validate_mounts(normalized, "sandbox", is_cpu=True) + return normalized + + +async def _cleanup( + operation: Awaitable[None], + timeout: float, + original: Optional[BaseException] = None, +) -> None: + """finish bounded cleanup; callers propagate an existing error on success.""" + task = asyncio.create_task(asyncio.wait_for(operation, timeout)) + interrupted = original + while not task.done(): + try: + await asyncio.shield(task) + except asyncio.CancelledError as error: + if interrupted is None: + interrupted = error + except BaseException: + # task.result() below propagates the failure with its cleanup context. + break + try: + task.result() + except BaseException as error: + if interrupted is not None: + raise interrupted from error + raise + if interrupted is not None and original is None: + raise interrupted + + +class AsyncioSandbox: + """a cached sandbox handle; construction never performs network I/O. + + supply at most one of ``image`` or ``template_id``; omitting both uses the + server's default image. unspecified resource and lifetime options are left + to the server's account policy. supply ``cpu_flavor_id`` and ``vcpu_count`` + together, or omit both for server defaults. memory is derived by the server + from the CPU flavor and vCPU count. + created handles own their remote sandbox in a context; ``get`` and ``list`` + return borrowed handles whose context only closes local resources. + """ + + def __init__( + self, + *, + image: Optional[str] = None, + template_id: Optional[str] = None, + name: Optional[str] = None, + cpu_flavor_id: Optional[str] = None, + vcpu_count: Optional[int] = None, + disk_gb: Optional[int] = None, + data_center_ids: Optional[Sequence[str]] = None, + mounts: Optional[Mapping[str, Volume]] = None, + ports: Optional[ + Mapping[ + int, + Literal["http", "tcp", "udp"] + | tuple[Literal["http", "tcp", "udp"], ...], + ] + ] = None, + cmd: Optional[Sequence[str]] = None, + entrypoint: Optional[Sequence[str]] = None, + start_ssh: Optional[bool] = None, + registry_auth: Optional[str] = None, + env: Optional[Mapping[str, str]] = None, + idle_timeout_seconds: Optional[int] = None, + max_lifetime_seconds: Optional[int] = None, + labels: Optional[Mapping[str, str]] = None, + api_key: Optional[str] = None, + base_url: Optional[str] = None, + request_timeout: float = 30, + startup_timeout: float = 60, + ) -> None: + if image is not None and template_id is not None: + raise ValueError("Supply at most one of image or template_id") + if (cpu_flavor_id is None) != (vcpu_count is None): + raise ValueError("Supply cpu_flavor_id and vcpu_count together") + self._initialize(api_key, base_url, request_timeout, startup_timeout) + self._mounts = _sandbox_mounts(mounts) + self._registry_auth = registry_auth + self._create_body = { + "image": image, + "templateId": template_id, + "name": name, + "disk": disk_gb, + "dataCenterIds": _string_list( + data_center_ids, "data_center_ids", nonempty=True + ), + "ports": _ports(ports), + "cmd": _string_list(cmd, "cmd"), + "entrypoint": _string_list(entrypoint, "entrypoint"), + "startSsh": start_ssh, + "env": dict(env) if env is not None else None, + "idleTimeoutSeconds": idle_timeout_seconds, + "maxLifetimeSeconds": max_lifetime_seconds, + "labels": dict(labels) if labels is not None else None, + } + if cpu_flavor_id is not None and vcpu_count is not None: + self._create_body["cpu"] = {"id": cpu_flavor_id, "vcpuCount": vcpu_count} + + def _initialize( + self, + api_key: Optional[str], + base_url: Optional[str], + request_timeout: float, + startup_timeout: float, + ) -> None: + self._request_timeout = _timeout(request_timeout, "request_timeout") + self._startup_timeout = _timeout( + startup_timeout, "startup_timeout", allow_zero=True + ) + self._api = AsyncSandboxAPI(api_key, base_url, request_timeout) + self._info: Optional[SandboxInfo] = None + self._sandbox_id: Optional[str] = None + self._create_body: Optional[dict[str, Any]] = None + self._mounts: Optional[dict[str, Volume]] = None + self._registry_auth: Optional[str] = None + self._lifetime_lock = asyncio.Lock() + self._owned = False + self._entering = False + self._creating = False + self._streams: set[AsyncioSandboxLogStream] = set() + + @classmethod + async def create(cls, **options: Any) -> "AsyncioSandbox": + """Create and return an owned handle, without claiming container readiness.""" + sandbox = cls(**options) + await sandbox._create() + return sandbox + + @classmethod + async def get( + cls, + sandbox_id: str, + *, + api_key: Optional[str] = None, + base_url: Optional[str] = None, + request_timeout: float = 30, + startup_timeout: float = 60, + ) -> "AsyncioSandbox": + """Fetch a borrowed handle; leaving its context does not terminate it.""" + sandbox = cls.__new__(cls) + sandbox._initialize(api_key, base_url, request_timeout, startup_timeout) + sandbox._sandbox_id = sandbox_id + try: + await sandbox.refresh() + except BaseException as error: + await _cleanup(sandbox.close(), request_timeout, error) + raise + return sandbox + + @classmethod + async def list( + cls, + *, + state: Optional[SandboxState] = None, + labels: Optional[Mapping[str, str]] = None, + api_key: Optional[str] = None, + base_url: Optional[str] = None, + request_timeout: float = 30, + startup_timeout: float = 60, + ) -> list["AsyncioSandbox"]: + """Return borrowed snapshots with independent, not-yet-open sessions.""" + _timeout(request_timeout, "request_timeout") + _timeout(startup_timeout, "startup_timeout", allow_zero=True) + api = AsyncSandboxAPI(api_key, base_url, request_timeout) + try: + rows = await api.list(state=state, labels=labels) + snapshots = [SandboxInfo.from_dict(row) for row in rows] + except BaseException as error: + await _cleanup(api.close(), request_timeout, error) + raise + else: + await _cleanup(api.close(), request_timeout) + handles = [] + for info in snapshots: + sandbox = cls.__new__(cls) + sandbox._initialize(api_key, base_url, request_timeout, startup_timeout) + sandbox._info = info + sandbox._sandbox_id = info.id + handles.append(sandbox) + return handles + + @property + def info(self) -> SandboxInfo: + """The latest server snapshot; raises before creation, never fetches.""" + if self._info is None: + raise RuntimeError("Sandbox has not been created or fetched") + return self._info + + @property + def id(self) -> str: + """The cached sandbox identifier.""" + return self.info.id + + @property + def state(self) -> SandboxState: + """The cached lifecycle state, not a container-readiness guarantee.""" + return self.info.state + + @property + def compute(self) -> Optional[SandboxCompute]: + """The cached allocation, if one currently exists.""" + return self.info.compute + + @property + def expires_at(self) -> datetime: + """The cached maximum-lifetime deadline.""" + return self.info.expires_at + + async def _create(self) -> None: + if self._creating: + raise RuntimeError("Sandbox creation is already in progress") + if self._sandbox_id is not None: + return + if self._create_body is None: + raise RuntimeError("A borrowed sandbox cannot create a new resource") + self._creating = True + + async def create_remote() -> None: + body = dict(self._create_body) + if self._mounts is not None: + resolved = await VolumeResolver(api=self._api).resolve_mounts( + self._mounts + ) + wire_mounts = {} + ids = set() + for mount in resolved: + if mount["id"] in ids: + raise ValueError("a sandbox cannot mount the same volume twice") + ids.add(mount["id"]) + wire_mounts[mount["kind"]] = [ + {"volumeId": mount["id"], "path": mount["path"]} + ] + if mount["kind"] == "network": + dc = mount["dataCenterId"] + requested = body.get("dataCenterIds") + if requested is not None and dc not in requested: + raise ValueError( + f"network volume data center {dc!r} is not in data_center_ids" + ) + body["dataCenterIds"] = [dc] + body["mounts"] = wire_mounts + if self._registry_auth is not None: + body["registry"] = await resolve_registry_auth( + self._registry_auth, api=self._api + ) + try: + data = await self._api.create(body) + except SandboxCreationError as error: + self._sandbox_id = error.sandbox_id + self._owned = True + raise + sandbox_id = data["id"] + if not isinstance(sandbox_id, str) or not sandbox_id: + raise ValueError("Create response has no valid sandbox id") + self._sandbox_id = sandbox_id + self._owned = True + self._info = SandboxInfo.from_dict(data) + + task = asyncio.create_task(create_remote()) + try: + await asyncio.shield(task) + except BaseException as original: + + async def recover_and_release(original_error: BaseException) -> None: + recovery_error = None + try: + if not task.done(): + await asyncio.wait_for(task, self._request_timeout) + else: + task.result() + except BaseException as error: + # finish releasing the resource before propagating recovery errors. + if error is not original_error: + recovery_error = error + try: + if self._sandbox_id is not None: + await self.terminate() + else: + await self.close() + except BaseException as error: + if recovery_error is not None: + raise error from recovery_error + raise + if recovery_error is not None: + raise recovery_error + + await _cleanup( + recover_and_release(original), 2 * self._request_timeout, original + ) + raise + finally: + self._creating = False + + async def __aenter__(self) -> "AsyncioSandbox": + """Create if needed and enter a single, non-reentrant ownership scope.""" + if self._entering: + raise RuntimeError( + "Sandbox contexts cannot be nested or entered concurrently" + ) + self._entering = True + try: + await self._create() + return self + except BaseException: + self._entering = False + raise + + async def __aexit__( + self, + exc_type: Optional[type[BaseException]], + exc: Optional[BaseException], + traceback: Optional[TracebackType], + ) -> None: + """Release the owned resource, shielding bounded cleanup from cancellation.""" + try: + operation = self.terminate() if self._owned else self.close() + await _cleanup(operation, self._request_timeout, exc) + finally: + self._entering = False + + async def refresh(self) -> SandboxInfo: + """Fetch a fresh snapshot, including after local close or termination.""" + if self._sandbox_id is None: + raise RuntimeError("Sandbox has not been created or fetched") + self._info = SandboxInfo.from_dict(await self._api.get(self._sandbox_id)) + return self._info + + async def update( + self, + *, + idle_timeout_seconds: Optional[int] = None, + max_lifetime_seconds: Optional[int] = None, + ) -> SandboxInfo: + """update only the supplied lifetime fields and cache the returned snapshot.""" + body = {} + if idle_timeout_seconds is not None: + body["idleTimeoutSeconds"] = idle_timeout_seconds + if max_lifetime_seconds is not None: + body["maxLifetimeSeconds"] = max_lifetime_seconds + if not body: + raise ValueError("update requires at least one timeout") + async with self._lifetime_lock: + self._info = SandboxInfo.from_dict(await self._api.update(self.id, body)) + return self._info + + async def extend(self, *, seconds: int) -> SandboxInfo: + """add seconds to the current total lifetime, not to the current time. + + this refreshes before sending an absolute lifetime to the server. + updates through this handle are serialized, but extensions made by + different clients are not atomic and may race. + """ + if isinstance(seconds, bool) or not isinstance(seconds, int) or seconds <= 0: + raise ValueError("seconds must be a positive integer") + async with self._lifetime_lock: + info = await self.refresh() + self._info = SandboxInfo.from_dict( + await self._api.extend(self.id, info.max_lifetime_seconds + seconds) + ) + return self._info + + async def _ready_operation( + self, + operation: Callable[[], Awaitable[_T]], + startup_timeout: Optional[float], + *, + handshake: bool = False, + ) -> _T: + timeout = ( + self._startup_timeout + if startup_timeout is None + else _timeout(startup_timeout, "startup_timeout", allow_zero=True) + ) + sandbox_id = self.id + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while True: + try: + if handshake and timeout > 0: + try: + return await asyncio.wait_for( + operation(), deadline - loop.time() + ) + except asyncio.TimeoutError as error: + if loop.time() >= deadline: + raise SandboxStartupTimeout(sandbox_id, timeout) from error + raise + return await operation() + except QueryError as error: + if ( + not isinstance(error, SandboxConflictError) + or error.code != "sandbox_starting" + or timeout == 0 + ): + raise + remaining = deadline - loop.time() + if remaining <= 0: + raise SandboxStartupTimeout(sandbox_id, timeout) from error + try: + info = await asyncio.wait_for(self.refresh(), remaining) + except asyncio.TimeoutError as refresh_error: + if loop.time() >= deadline: + raise SandboxStartupTimeout( + sandbox_id, timeout + ) from refresh_error + raise + if info.state in ("FAILED", "TERMINATED"): + raise SandboxStateError(info) from error + remaining = deadline - loop.time() + if remaining <= 0: + raise SandboxStartupTimeout(sandbox_id, timeout) from error + await asyncio.sleep(min(0.25, remaining)) + if loop.time() >= deadline: + raise SandboxStartupTimeout(sandbox_id, timeout) from error + + async def exec( + self, + command: Sequence[str], + *, + check: bool = False, + startup_timeout: Optional[float] = None, + timeout_seconds: Optional[int] = None, + background: bool = False, + ) -> ExecResult: + """execute argv, retrying only explicit 409 startup rejections. + + the startup deadline limits retries, not execution of an accepted + command. each request retains its request timeout. transport errors, + timeouts, and other HTTP errors are never retried because the command + might already have executed. ``check=False`` returns command failures + in ``ExecResult.error``, matching ``subprocess.run``. use ``check=True`` + to raise ``SandboxExecutionError`` with partial output on failure. + + ``timeout_seconds`` (1–50) requests a command limit; the backend may + impose a shorter cap. ``request_timeout`` bounds the client http wait. + ``background=True`` returns after starting a detached command, not its + completion; redirect its output to a file and poll for results. + """ + if isinstance(command, (str, bytes)) or not isinstance(command, Sequence): + raise TypeError( + "command must be a nonempty sequence of strings, not a string" + ) + if not command or any(not isinstance(argument, str) for argument in command): + raise ValueError("command must be a nonempty sequence of strings") + if timeout_seconds is not None and ( + isinstance(timeout_seconds, bool) + or not isinstance(timeout_seconds, int) + or not 1 <= timeout_seconds <= 50 + ): + raise ValueError("timeout_seconds must be an integer from 1 to 50") + if not isinstance(background, bool): + raise TypeError("background must be a bool") + argv = tuple(command) + data = await self._ready_operation( + lambda: self._api.exec( + self.id, + argv, + timeout_seconds=timeout_seconds, + background=background, + ), + startup_timeout, + ) + result = ExecResult( + output=data["output"], + error=data.get("error"), + stdout=data.get("stdout"), + stderr=data.get("stderr"), + exit_code=data.get("exitCode"), + duration_ms=data.get("durationMs"), + truncated=data.get("truncated"), + ) + if check and (result.error is not None or result.exit_code not in (None, 0)): + raise SandboxExecutionError(self.id, result) + return result + + def logs( + self, + *, + source: Optional[LogSource] = None, + tail: Optional[int] = None, + since: Optional[str] = None, + last_event_id: Optional[str] = None, + startup_timeout: Optional[float] = None, + ) -> "AsyncioSandboxLogStream": + """Return a lazy, closeable stream of main-process or lifecycle logs. + + This does not include exec output. Only the initial HTTP handshake is + retried on 409; disconnects and errors after opening are never replayed. + Use an async context or ``aclose`` when stopping iteration early. + """ + sandbox_id = self.id + stream = AsyncioSandboxLogStream( + self, sandbox_id, source, tail, since, last_event_id, startup_timeout + ) + self._streams.add(stream) + return stream + + async def terminate(self) -> None: + """Delete remotely and always close locally, even if deletion fails. + + The readable remote record remains. Only successful deletion changes + the cached state; timestamps are left untouched until ``refresh``. + """ + try: + if self._sandbox_id is None: + raise RuntimeError("Sandbox has not been created or fetched") + await self._api.terminate(self._sandbox_id) + if self._info is not None: + self._info = replace(self._info, state="TERMINATED", compute=None) + except BaseException as error: + await _cleanup(self.close(), self._request_timeout, error) + raise + else: + await _cleanup(self.close(), self._request_timeout) + + async def close(self) -> None: + """Close streams and the local session without deleting; later reuse is allowed.""" + error = None + for stream in tuple(self._streams): + try: + await stream.aclose() + except BaseException as stream_error: + # close every stream before propagating cancellation or another failure. + if error is None: + error = stream_error + try: + await self._api.close() + self._lifetime_lock = asyncio.Lock() + except BaseException as close_error: + if error is not None: + raise error from close_error + raise + if error is not None: + raise error + + +class AsyncioSandboxLogStream: + """A handle-owned typed log stream with a separately awaitable handshake.""" + + def __init__( + self, + sandbox: AsyncioSandbox, + sandbox_id: str, + source: Optional[LogSource], + tail: Optional[int], + since: Optional[str], + last_event_id: Optional[str], + startup_timeout: Optional[float], + ) -> None: + self._sandbox = sandbox + self._sandbox_id = sandbox_id + self._options = dict( + source=source, tail=tail, since=since, last_event_id=last_event_id + ) + self._startup_timeout = startup_timeout + self._stream: Any = None + self._opened = False + self._closed = False + self._opening: Optional[asyncio.Task[None]] = None + + async def _open_once(self) -> None: + stream = self._sandbox._api.logs(self._sandbox_id, **self._options) + self._stream = stream + try: + await stream.open() + except BaseException as error: + self._stream = None + await _cleanup(stream.aclose(), self._sandbox._request_timeout, error) + raise + + async def open(self) -> "AsyncioSandboxLogStream": + """Complete the HTTP handshake without waiting for the first log event.""" + if self._closed: + raise RuntimeError("Log stream is closed") + if not self._opened: + if self._opening is None: + self._opening = asyncio.create_task( + self._sandbox._ready_operation( + self._open_once, self._startup_timeout, handshake=True + ) + ) + try: + await self._opening + if self._closed: + raise RuntimeError("Log stream was closed while opening") + self._opened = True + except BaseException as error: + await _cleanup(self.aclose(), self._sandbox._request_timeout, error) + raise + return self + + def __aiter__(self) -> "AsyncioSandboxLogStream": + return self + + async def __anext__(self) -> LogEvent: + if self._closed: + raise StopAsyncIteration + try: + await self.open() + return LogEvent.from_dict(await self._stream.__anext__()) + except StopAsyncIteration: + await self.aclose() + raise + except BaseException as error: + await _cleanup(self.aclose(), self._sandbox._request_timeout, error) + raise + + async def aclose(self) -> None: + """Release the response and detach from the sandbox; idempotent.""" + if self._closed: + return + self._closed = True + self._sandbox._streams.discard(self) + try: + if self._opening is not None and not self._opening.done(): + self._opening.cancel() + await asyncio.gather(self._opening, return_exceptions=True) + finally: + stream, self._stream = self._stream, None + if stream is not None: + await stream.aclose() + + async def __aenter__(self) -> "AsyncioSandboxLogStream": + return await self.open() + + async def __aexit__( + self, + exc_type: Optional[type[BaseException]], + exc: Optional[BaseException], + traceback: Optional[TracebackType], + ) -> None: + await _cleanup(self.aclose(), self._sandbox._request_timeout, exc) diff --git a/runpod/sandbox/models.py b/runpod/sandbox/models.py new file mode 100644 index 000000000..f65da1fed --- /dev/null +++ b/runpod/sandbox/models.py @@ -0,0 +1,175 @@ +"""Typed snapshots and results returned by the sandbox domain.""" + +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Literal, Mapping, Optional + +from runpod.error import RunPodError + +SandboxState = Literal["CREATING", "RUNNING", "TERMINATED", "FAILED"] +LogSource = Literal["container", "system"] + + +def _timestamp(value: str) -> datetime: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + + +@dataclass(frozen=True) +class SandboxCompute: + vcpu_count: float + memory_in_gb: float + container_disk_in_gb: int + cost_per_hr: float + + +@dataclass(frozen=True) +class SandboxInfo: + """A server snapshot. Reading fields does not fetch or refresh anything.""" + + id: str + name: str + state: SandboxState + idle_timeout_seconds: int + max_lifetime_seconds: int + last_activity_at: datetime + idle_expires_at: datetime + expires_at: datetime + created_at: datetime + updated_at: datetime + template_id: Optional[str] + image: Optional[str] + cpu_flavor_id: Optional[str] + termination_reason: Optional[str] + terminated_at: Optional[datetime] + labels: Optional[Mapping[str, str]] + compute: Optional[SandboxCompute] + data_center_id: Optional[str] + env: Optional[Mapping[str, str]] + registry: Optional[str] + started_at: Optional[datetime] + ssh: Optional[Mapping[str, Any]] + ports: Optional[list[Mapping[str, Any]]] + mounts: Optional[Mapping[str, list[Mapping[str, str]]]] + cmd: Optional[list[str]] + entrypoint: Optional[list[str]] + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> "SandboxInfo": + compute = data.get("compute") + terminated_at = data.get("terminatedAt") + started_at = data.get("startedAt") + return cls( + id=data["id"], + name=data["name"], + state=data["state"], + idle_timeout_seconds=data["idleTimeoutSeconds"], + max_lifetime_seconds=data["maxLifetimeSeconds"], + last_activity_at=_timestamp(data["lastActivityAt"]), + idle_expires_at=_timestamp(data["idleExpiresAt"]), + expires_at=_timestamp(data["expiresAt"]), + created_at=_timestamp(data["createdAt"]), + updated_at=_timestamp(data["updatedAt"]), + template_id=data.get("templateId"), + image=data.get("image"), + cpu_flavor_id=(data.get("cpu") or {}).get("id"), + termination_reason=data.get("terminationReason"), + terminated_at=_timestamp(terminated_at) if terminated_at else None, + labels=dict(data["labels"]) if data.get("labels") is not None else None, + data_center_id=data.get("dataCenterId"), + env=dict(data["env"]) if data.get("env") is not None else None, + registry=data.get("registry"), + started_at=_timestamp(started_at) if started_at is not None else None, + ssh=dict(data["ssh"]) if data.get("ssh") is not None else None, + ports=( + [dict(port) for port in data["ports"]] + if data.get("ports") is not None + else None + ), + mounts=( + { + kind: [dict(mount) for mount in mounts] + for kind, mounts in data["mounts"].items() + } + if data.get("mounts") is not None + else None + ), + cmd=list(data["cmd"]) if data.get("cmd") is not None else None, + entrypoint=( + list(data["entrypoint"]) if data.get("entrypoint") is not None else None + ), + compute=( + SandboxCompute( + vcpu_count=compute["vcpuCount"], + memory_in_gb=compute["memoryInGb"], + container_disk_in_gb=compute["containerDiskInGb"], + cost_per_hr=compute["costPerHr"], + ) + if compute is not None + else None + ), + ) + + +@dataclass(frozen=True) +class ExecResult: + """command output with nullable host-reported execution details.""" + + output: str + error: Optional[str] = None + stdout: Optional[str] = None + stderr: Optional[str] = None + exit_code: Optional[int] = None + duration_ms: Optional[int] = None + truncated: Optional[bool] = None + + +@dataclass(frozen=True) +class LogEvent: + """One log record; preserve the opaque SSE id for subsequent resume requests.""" + + source: LogSource + line: str + timestamp: datetime + id: Optional[str] = None + + @classmethod + def from_dict(cls, data: Mapping[str, Any]) -> "LogEvent": + return cls( + source=data["source"], + line=data["line"], + timestamp=_timestamp(data["ts"]), + id=data.get("id"), + ) + + +class SandboxExecutionError(RunPodError): + """A command failed with check=True; its partial output remains available.""" + + def __init__(self, sandbox_id: str, result: ExecResult): + super().__init__( + result.error or f"Sandbox command exited with status {result.exit_code}" + ) + self.sandbox_id = sandbox_id + self.result = result + + +class SandboxStateError(RunPodError): + """An operation cannot proceed because the sandbox reached a terminal state.""" + + def __init__(self, info: SandboxInfo): + super().__init__( + f"Sandbox {info.id} is {info.state}" + + (f": {info.termination_reason}" if info.termination_reason else "") + ) + self.info = info + + +class SandboxStartupTimeout(TimeoutError): + """The sandbox did not accept the operation within the startup deadline.""" + + def __init__(self, sandbox_id: str, timeout: float): + super().__init__( + f"Sandbox {sandbox_id} did not become ready within {timeout:g}s" + ) + self.sandbox_id = sandbox_id + self.timeout = timeout diff --git a/runpod/sandbox/sync.py b/runpod/sandbox/sync.py new file mode 100644 index 000000000..f01acf625 --- /dev/null +++ b/runpod/sandbox/sync.py @@ -0,0 +1,583 @@ +"""Blocking sandbox facade backed by one stable asyncio loop per active handle.""" + +import asyncio +import threading +from concurrent.futures import Future +from datetime import datetime +from typing import ( + Any, + Coroutine, + Iterator, + Literal, + Mapping, + Optional, + Sequence, + TypeVar, +) + +from runpod.apps.volume import Volume +from runpod.sandbox.asyncio import AsyncioSandbox, _cleanup +from .models import ( + ExecResult, + LogEvent, + LogSource, + SandboxCompute, + SandboxInfo, + SandboxState, +) + +_T = TypeVar("_T") + + +class _LoopRunner: + """Keep aiohttp sessions on their original loop, including across log reads.""" + + def __init__(self) -> None: + self._loop: Optional[asyncio.AbstractEventLoop] = None + self._thread: Optional[threading.Thread] = None + self._lock = threading.RLock() + self._stopping = False + self._stopped = threading.Event() + self._stopped.set() + self._loop_finished = threading.Event() + self._tasks: dict[asyncio.Task, bool] = {} + + @property + def started(self) -> bool: + return self._thread is not None + + def _start(self) -> asyncio.AbstractEventLoop: + with self._lock: + if self._loop is None: + ready = threading.Event() + self._loop_finished.clear() + + def serve() -> None: + loop = asyncio.new_event_loop() + self._loop = loop + ready.set() + try: + loop.run_forever() + finally: + try: + loop.run_until_complete(loop.shutdown_asyncgens()) + loop.run_until_complete(loop.shutdown_default_executor()) + loop.close() + finally: + self._loop_finished.set() + + self._thread = threading.Thread( + target=serve, name="runpod-sandbox", daemon=True + ) + self._thread.start() + ready.wait() + assert self._loop is not None + return self._loop + + def run( + self, coroutine: Coroutine[Any, Any, _T], *, cleanup: bool = False + ) -> _T: + if threading.current_thread() is self._thread: + coroutine.close() + raise RuntimeError("Cannot call the sync sandbox API from its own loop") + loop: asyncio.AbstractEventLoop + future: Future[_T] = Future() + finished = threading.Event() + task: Optional[asyncio.Task[tuple[Optional[_T], Optional[BaseException]]]] = ( + None + ) + invoked = False + + async def invoke() -> tuple[Optional[_T], Optional[BaseException]]: + nonlocal invoked + invoked = True + try: + return await coroutine, None + except BaseException as error: + # interrupts belong to the calling thread, not the background loop. + return None, error + + def complete(done: asyncio.Task) -> None: + self._tasks.pop(done, None) + try: + result, error = done.result() + if error is not None: + future.set_exception(error) + else: + future.set_result(result) + except BaseException as error: + # the waiting caller receives the task's cancellation or failure. + future.set_exception(error) + finally: + if not invoked: + coroutine.close() + finished.set() + + def submit() -> None: + nonlocal task + task = loop.create_task(invoke()) + self._tasks[task] = cleanup + task.add_done_callback(complete) + + def cancel() -> None: + if task is not None: + task.cancel() + + try: + while True: + with self._lock: + if not self._stopping: + loop = self._start() + loop.call_soon_threadsafe(submit) + break + if not cleanup: + raise RuntimeError("Sandbox loop is shutting down") + stopped = self._stopped + while not stopped.is_set(): + try: + stopped.wait() + except KeyboardInterrupt: + # owner cleanup must survive repeated interruptions. + continue + except BaseException: + coroutine.close() + raise + try: + return future.result() + except BaseException as original: + with self._lock: + if not cleanup and not finished.is_set() and not self._stopping: + loop.call_soon_threadsafe(cancel) + while not finished.is_set(): + try: + finished.wait() + except KeyboardInterrupt: + # async cleanup owns its child tasks and bounds their lifetime. + continue + error = future.exception() + if error is not None and error is not original: + raise original from error + raise + + def stop(self) -> None: + if threading.current_thread() is self._thread: + raise RuntimeError("Cannot stop the sync sandbox API from its own loop") + with self._lock: + if self._stopping: + stopped = self._stopped + shutdown = None + else: + loop, thread = self._loop, self._thread + if loop is None or thread is None: + return + self._stopping = True + stopped = self._stopped = threading.Event() + + async def drain() -> None: + # cancel invocations, not the protected cleanup tasks they own. + tasks = tuple(self._tasks) + for task in tasks: + if not self._tasks[task]: + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + + shutdown = asyncio.run_coroutine_threadsafe(drain(), loop) + interrupted = None + if shutdown is None: + while not stopped.is_set(): + try: + stopped.wait() + except KeyboardInterrupt as error: + interrupted = error + else: + try: + while True: + try: + shutdown.result() + break + except KeyboardInterrupt as error: + interrupted = error + finally: + loop.call_soon_threadsafe(loop.stop) + while not self._loop_finished.is_set(): + try: + self._loop_finished.wait() + except KeyboardInterrupt as error: + interrupted = error + with self._lock: + self._loop = None + self._thread = None + self._stopping = False + stopped.set() + if interrupted is not None: + raise interrupted + + +class SandboxLogs(Iterator[LogEvent]): + """Closeable log iterator. Use a with block when stopping consumption early.""" + + def __init__(self, stream: Any, runner: _LoopRunner) -> None: + self._stream = stream + self._runner = runner + self._closed = False + + def __enter__(self) -> "SandboxLogs": + if self._closed: + raise RuntimeError("Log stream is closed") + self._runner.run(self._stream.__aenter__()) + return self + + def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> bool: + self._closed = True + if self._runner.started: + self._runner.run(self._stream.__aexit__(exc_type, exc, traceback)) + return False + + def __iter__(self) -> "SandboxLogs": + return self + + def __next__(self) -> LogEvent: + if self._closed: + raise StopIteration + try: + return self._runner.run(self._stream.__anext__()) + except StopAsyncIteration: + self._closed = True + raise StopIteration from None + except BaseException: + self.close() + raise + + def close(self) -> None: + if not self._closed: + self._closed = True + if self._runner.started: + self._runner.run(self._stream.aclose()) + + +class Sandbox: + """An isolated CPU sandbox with blocking methods. + + ``with Sandbox(image=...)`` creates on entry and terminates on exit. + ``create`` leaves the lifetime under your control; call ``terminate`` when + finished, or ``close`` to release only local connections. Handles retrieved + with ``get`` or ``list`` never terminate compute on context exit. + + supply at most one of ``image`` or ``template_id``; omitting both uses the + server's default image. supply ``cpu_flavor_id`` and ``vcpu_count`` together, + or omit both for server defaults. memory is derived by the server from the + CPU flavor and vCPU count. + + A stable background event loop runs the async implementation. No thread or + connection is started by the constructor. Prefer AsyncioSandbox in async + applications so blocking calls do not pause the application's event loop. + """ + + def __init__( + self, + *, + image: Optional[str] = None, + template_id: Optional[str] = None, + name: Optional[str] = None, + cpu_flavor_id: Optional[str] = None, + vcpu_count: Optional[int] = None, + disk_gb: Optional[int] = None, + data_center_ids: Optional[Sequence[str]] = None, + mounts: Optional[Mapping[str, Volume]] = None, + ports: Optional[ + Mapping[ + int, + Literal["http", "tcp", "udp"] + | tuple[Literal["http", "tcp", "udp"], ...], + ] + ] = None, + cmd: Optional[Sequence[str]] = None, + entrypoint: Optional[Sequence[str]] = None, + start_ssh: Optional[bool] = None, + registry_auth: Optional[str] = None, + env: Optional[Mapping[str, str]] = None, + idle_timeout_seconds: Optional[int] = None, + max_lifetime_seconds: Optional[int] = None, + labels: Optional[Mapping[str, str]] = None, + api_key: Optional[str] = None, + base_url: Optional[str] = None, + request_timeout: float = 30, + startup_timeout: float = 60, + ) -> None: + self._runner = _LoopRunner() + self._entered = False + self._context_lock = threading.Lock() + self._sandbox = AsyncioSandbox( + image=image, + template_id=template_id, + name=name, + cpu_flavor_id=cpu_flavor_id, + vcpu_count=vcpu_count, + disk_gb=disk_gb, + data_center_ids=data_center_ids, + mounts=mounts, + ports=ports, + cmd=cmd, + entrypoint=entrypoint, + start_ssh=start_ssh, + registry_auth=registry_auth, + env=env, + idle_timeout_seconds=idle_timeout_seconds, + max_lifetime_seconds=max_lifetime_seconds, + labels=labels, + api_key=api_key, + base_url=base_url, + request_timeout=request_timeout, + startup_timeout=startup_timeout, + ) + + @classmethod + def _from_async( + cls, sandbox: AsyncioSandbox, runner: Optional[_LoopRunner] = None + ) -> "Sandbox": + instance = cls.__new__(cls) + instance._sandbox = sandbox + instance._runner = runner if runner is not None else _LoopRunner() + instance._entered = False + instance._context_lock = threading.Lock() + return instance + + @classmethod + def create(cls, **options: Any) -> "Sandbox": + """Create immediately; the returned snapshot may still be CREATING.""" + sandbox = cls(**options) + completed = False + + async def create() -> AsyncioSandbox: + nonlocal completed + await sandbox._sandbox._create() + completed = True + return sandbox._sandbox + + try: + sandbox._runner.run(create()) + return sandbox + except BaseException as original: + sandbox._abort_entry(original, completed) + raise + + @classmethod + def get( + cls, + sandbox_id: str, + *, + api_key: Optional[str] = None, + base_url: Optional[str] = None, + request_timeout: float = 30, + startup_timeout: float = 60, + ) -> "Sandbox": + """Fetch a borrowed handle; closing its context does not terminate it.""" + runner = _LoopRunner() + sandbox = None + + async def get() -> AsyncioSandbox: + nonlocal sandbox + sandbox = await AsyncioSandbox.get( + sandbox_id, + api_key=api_key, + base_url=base_url, + request_timeout=request_timeout, + startup_timeout=startup_timeout, + ) + return sandbox + + try: + result = runner.run(get()) + return cls._from_async(result, runner) + except BaseException as original: + if sandbox is not None: + cls._from_async(sandbox, runner)._abort_entry(original, True) + else: + try: + runner.stop() + except BaseException as error: + raise original from error + raise + + @classmethod + def list( + cls, + *, + state: Optional[SandboxState] = None, + labels: Optional[Mapping[str, str]] = None, + api_key: Optional[str] = None, + base_url: Optional[str] = None, + request_timeout: float = 30, + startup_timeout: float = 60, + ) -> list["Sandbox"]: + """List borrowed handles. Filters combine with AND on the server.""" + runner = _LoopRunner() + try: + sandboxes = runner.run( + AsyncioSandbox.list( + state=state, + labels=labels, + api_key=api_key, + base_url=base_url, + request_timeout=request_timeout, + startup_timeout=startup_timeout, + ) + ) + finally: + runner.stop() + return [cls._from_async(sandbox) for sandbox in sandboxes] + + @property + def info(self) -> SandboxInfo: + return self._sandbox.info + + @property + def id(self) -> str: + return self._sandbox.id + + @property + def state(self) -> SandboxState: + return self._sandbox.state + + @property + def compute(self) -> Optional[SandboxCompute]: + return self._sandbox.compute + + @property + def expires_at(self) -> datetime: + return self._sandbox.expires_at + + def refresh(self) -> SandboxInfo: + """Fetch current server metadata and replace the local snapshot.""" + return self._runner.run(self._sandbox.refresh()) + + def update( + self, + *, + idle_timeout_seconds: Optional[int] = None, + max_lifetime_seconds: Optional[int] = None, + ) -> SandboxInfo: + """update the supplied timeout fields and return the refreshed snapshot.""" + return self._runner.run( + self._sandbox.update( + idle_timeout_seconds=idle_timeout_seconds, + max_lifetime_seconds=max_lifetime_seconds, + ) + ) + + def extend(self, *, seconds: int) -> SandboxInfo: + """add to the current total lifetime; not atomic across separate clients.""" + return self._runner.run(self._sandbox.extend(seconds=seconds)) + + def exec( + self, + command: Sequence[str], + *, + check: bool = False, + startup_timeout: Optional[float] = None, + timeout_seconds: Optional[int] = None, + background: bool = False, + ) -> ExecResult: + """Execute argv, waiting only for explicit startup rejections.""" + return self._runner.run( + self._sandbox.exec( + command, + check=check, + startup_timeout=startup_timeout, + timeout_seconds=timeout_seconds, + background=background, + ) + ) + + def logs( + self, + *, + source: Optional[LogSource] = None, + tail: Optional[int] = None, + since: Optional[str] = None, + last_event_id: Optional[str] = None, + startup_timeout: Optional[float] = None, + ) -> SandboxLogs: + """Stream container/system logs, not exec output; close on early exit.""" + return SandboxLogs( + self._sandbox.logs( + source=source, + tail=tail, + since=since, + last_event_id=last_event_id, + startup_timeout=startup_timeout, + ), + self._runner, + ) + + def terminate(self) -> None: + """Release remote compute and local connections; safe to repeat.""" + try: + self._runner.run(self._sandbox.terminate()) + finally: + self._runner.stop() + + def close(self) -> None: + """Release local connections and the loop, without terminating compute.""" + if self._runner.started: + try: + self._runner.run(self._sandbox.close()) + finally: + self._runner.stop() + + def _abort_entry(self, original: BaseException, completed: bool) -> None: + async def release() -> None: + try: + # shutdown may have closed the session's loop before handoff. + await _cleanup(self._sandbox.close(), self._sandbox._request_timeout) + finally: + if completed: + await self._sandbox.__aexit__( + type(original), original, original.__traceback__ + ) + + try: + try: + self._runner.run(release(), cleanup=True) + finally: + self._runner.stop() + except BaseException as error: + if error is not original: + raise original from error + raise + + def __enter__(self) -> "Sandbox": + with self._context_lock: + if self._entered: + raise RuntimeError("Sandbox context is already entered") + self._entered = True + completed = False + + async def enter() -> AsyncioSandbox: + nonlocal completed + result = await self._sandbox.__aenter__() + completed = True + return result + + try: + self._runner.run(enter()) + return self + except BaseException as original: + try: + self._abort_entry(original, completed) + finally: + with self._context_lock: + self._entered = False + raise + + def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> bool: + try: + self._runner.run(self._sandbox.__aexit__(exc_type, exc, traceback)) + return False + finally: + try: + self._runner.stop() + finally: + with self._context_lock: + self._entered = False diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index 8bb8b577c..b0d861fca 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -8,8 +8,6 @@ import pytest import runpod -from tests.e2e.e2e_provisioner import load_test_cases, provision_endpoints - log = logging.getLogger(__name__) REQUEST_TIMEOUT = 300 # seconds per job request @@ -41,6 +39,8 @@ def require_api_key(): @pytest.fixture(scope="session") def test_cases(): """Load test cases from tests.json.""" + from tests.e2e.e2e_provisioner import load_test_cases + cases = load_test_cases() log.info("Loaded %d test cases: %s", len(cases), [c.get("id") for c in cases]) return cases @@ -52,6 +52,8 @@ def endpoints(require_api_key, test_cases): Endpoints deploy lazily on first .run()/.runsync() call. """ + from tests.e2e.e2e_provisioner import provision_endpoints + eps = provision_endpoints(test_cases) for key, ep in eps.items(): log.info("Endpoint ready: name=%s image=%s template.dockerArgs=%s", ep.name, ep.image, ep.template.dockerArgs if ep.template else "N/A") diff --git a/tests/e2e/test_sandbox.py b/tests/e2e/test_sandbox.py new file mode 100644 index 000000000..844b33598 --- /dev/null +++ b/tests/e2e/test_sandbox.py @@ -0,0 +1,100 @@ +import asyncio +import os +import time + +import pytest + +from runpod import AsyncioSandbox, Sandbox +from runpod.sandbox import SandboxExecutionError + + +async def _read_with_borrowed_async(sandbox_id, connection_options): + async with await AsyncioSandbox.get(sandbox_id, **connection_options) as borrowed: + result = await borrowed.exec( + ["python", "-c", "print('borrowed async')"], + check=True, + timeout_seconds=4, + ) + assert result.output.strip() == "borrowed async" + + +def test_sandbox_execution_and_borrowed_handles(require_api_key): + connection_options = { + "api_key": os.environ["RUNPOD_API_KEY"], + "request_timeout": 30, + "startup_timeout": 90, + } + with Sandbox( + image="python:3.12-slim", + cmd=["sleep", "infinity"], + idle_timeout_seconds=180, + max_lifetime_seconds=300, + **connection_options, + ) as sandbox: + result = sandbox.exec( + ["python", "-c", "print('foreground ready')"], + check=True, + timeout_seconds=4, + ) + assert result.output.strip() == "foreground ready" + + with pytest.raises(SandboxExecutionError) as failure: + sandbox.exec( + ["python", "-c", "print('expected failure', flush=True); raise SystemExit(17)"], + check=True, + timeout_seconds=4, + ) + assert failure.value.sandbox_id == sandbox.id + assert "expected failure" in failure.value.result.output + + sandbox.exec( + [ + "python", + "-c", + ( + "import pathlib, time; time.sleep(6); " + + "pathlib.Path('/tmp/runpod-sandbox-e2e').write_text('detached complete')" + ), + ], + check=True, + timeout_seconds=15, + background=True, + ) + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + result = sandbox.exec( + [ + "python", + "-c", + ( + "import pathlib; path = pathlib.Path('/tmp/runpod-sandbox-e2e'); " + + "print(path.read_text() if path.exists() else 'pending')" + ), + ], + check=True, + timeout_seconds=4, + startup_timeout=0, + ) + if result.output.strip() == "detached complete": + break + assert result.output.strip() == "pending", result + time.sleep(1) + else: + pytest.fail("detached command did not complete within 30 seconds") + + with Sandbox.get(sandbox.id, **connection_options) as borrowed: + result = borrowed.exec( + ["python", "-c", "print('borrowed sync')"], + check=True, + timeout_seconds=4, + ) + assert result.output.strip() == "borrowed sync" + + asyncio.run(_read_with_borrowed_async(sandbox.id, connection_options)) + result = sandbox.exec( + ["python", "-c", "print('owner still running')"], check=True + ) + assert result.output.strip() == "owner still running" + + with Sandbox.get(sandbox.id, **connection_options) as terminated: + assert terminated.info.state == "TERMINATED" diff --git a/tests/test_init.py b/tests/test_init.py index 694ec2894..ee1c6da80 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -87,6 +87,7 @@ def test_private_imports_not_exported(self): assert private_symbol not in all_symbols, f"Private symbol '{private_symbol}' should not be in __all__" + def test_no_duplicate_symbols_in_all(self): """Test that __all__ contains no duplicate symbols.""" all_symbols = runpod.__all__ diff --git a/tests/test_sandbox.py b/tests/test_sandbox.py new file mode 100644 index 000000000..af24e9523 --- /dev/null +++ b/tests/test_sandbox.py @@ -0,0 +1,1383 @@ +"""Sandbox lifecycle and streaming regressions over an actual HTTP connection.""" + +import asyncio +import subprocess +import sys +import threading +from collections import deque +from concurrent.futures import Future +from contextlib import contextmanager +from datetime import datetime, timedelta +from typing import get_type_hints + +import pytest +from aiohttp import web + +import runpod.sandbox as sandbox_domain +from runpod import AsyncioSandbox, Sandbox +from runpod.api.sandboxes import AsyncSandboxAPI +from runpod.apps.volume import GlobalVolume, NetworkVolume, VolumeError +from runpod.error import AuthenticationError, QueryError +from runpod.sandbox import ( + SandboxCreationError, + SandboxExecutionError, + SandboxStartupTimeout, + SandboxStateError, +) + + +class SandboxService: + """A controllable REST peer for lifecycle races, not a mocked SDK transport.""" + + def __init__(self): + self.records = {} + self.requests = [] + self.executed = [] + self.exec_statuses = deque() + self.exec_responses = deque() + self.log_statuses = deque() + self.conflict_code = "sandbox_starting" + self.network_volumes = [] + self.global_volumes = [] + self.global_error = None + self.volume_page_size = None + self.volume_datacenters = [ + {"id": "US-KS-2", "networkVolumeTypes": ["NETWORK_VOLUME"]} + ] + self.registries = [] + self.delete_status = 204 + self.create_status = 201 + self.create_error = "container provisioning failed" + self.job_started = threading.Event() + self.job_completed = threading.Event() + self.allow_job = threading.Event() + self.jobs = set() + self.create_started = threading.Event() + self.allow_create = threading.Event() + self.allow_create.set() + self.log_started = threading.Event() + self.allow_logs = threading.Event() + self.allow_logs.set() + self.log_disconnected = threading.Event() + self.hold_logs = False + self.stopping = threading.Event() + # Mix line endings, split a UTF-8 codepoint, preserve an opaque SSE id, + # and include both a multiline data event and a non-JSON timeout frame. + self.log_bytes = ( + ": heartbeat\r\n\r\nid: opaque/42\r\nevent: message\r\n" + 'data: {"source":"container",\r\n' + 'data: "line":"caf\u00e9","ts":"2026-09-14T12:00:00Z"}\r\n\r\n' + 'id: opaque/43\rdata: {"source":"system","line":"started",' + '"ts":"2026-09-14T12:00:01Z"}\r\r' + "event: timeout\ndata: max stream duration reached\n\n" + ).encode() + + def create_record(self, body): + sandbox_id = f"sandbox-{len(self.records) + 1}" + now = "2026-09-14T12:00:00Z" + record = { + "id": sandbox_id, + "name": body.get("name", sandbox_id), + "state": "RUNNING", + "imageName": body.get("imageName"), + "templateId": body.get("templateId"), + "idleTimeoutSeconds": body.get("idleTimeoutSeconds", 300), + "maxLifetimeSeconds": body.get("maxLifetimeSeconds", 900), + "lastActivityAt": now, + "idleExpiresAt": "2026-09-14T12:05:00Z", + "expiresAt": "2026-09-14T12:15:00Z", + "createdAt": now, + "updatedAt": now, + "labels": body.get("labels", {}), + "cpuFlavorId": body.get("cpuFlavorId"), + "dataCenterId": (body.get("dataCenterIds") or [None])[0], + "env": body.get("env", {}), + "registry": body.get("registry"), + "startedAt": now, + "ssh": {"proxy": {"command": "ssh sandbox@ssh.runpod.io"}}, + "ports": body.get("ports", []), + "mounts": body.get("mounts", {"network": [], "global": []}), + "cmd": body.get("cmd", ["tail", "-f", "/dev/null"]), + "entrypoint": body.get("entrypoint", []), + "compute": { + "vcpuCount": body.get("vcpuCount", 2), + "memoryInGb": body.get("memoryInGb", 4), + "containerDiskInGb": body.get("disk", 10), + "costPerHr": 0.026, + }, + } + self.records[sandbox_id] = record + return record + + +@contextmanager +def sandbox_peer(): + service = SandboxService() + ready, stopped = threading.Event(), asyncio.Event() + + def failure(status, detail, code=None): + payload = {"detail": detail} + if code is not None: + payload["code"] = code + return web.json_response( + payload, status=status, content_type="application/problem+json" + ) + + def volume_page(request, records, collection): + start = int(request.query.get("cursor", "0")) + end = start + (service.volume_page_size or len(records)) + pagination = {"hasNextPage": end < len(records)} + if pagination["hasNextPage"]: + pagination["nextCursor"] = str(end) + return web.json_response( + {collection: records[start:end], "pagination": pagination} + ) + + async def handle(request): + body = await request.json() if request.can_read_body else None + service.requests.append( + (request.method, request.path, body, dict(request.headers)) + ) + if request.headers.get("Authorization") != "Bearer sandbox-test-key": + return failure(401, "invalid key") + if request.path == "/v2/global-volumes": + if service.global_error: + return failure(403, service.global_error) + if request.method == "POST": + volume = { + "id": f"global-{len(service.global_volumes) + 1}", + "name": body["name"], + } + service.global_volumes.append(volume) + return web.json_response(volume, status=201) + return volume_page(request, service.global_volumes, "globalVolumes") + if request.path == "/v2/catalog/datacenters": + return web.json_response({"dataCenters": service.volume_datacenters}) + if request.path == "/v2/network-volumes": + if request.method == "POST": + volume = {"id": "created-volume", **body} + service.network_volumes.append(volume) + return web.json_response(volume, status=201) + return volume_page(request, service.network_volumes, "networkVolumes") + if request.path == "/v2/registries": + return web.json_response({"registries": service.registries}) + sandbox_id = request.match_info.get("id") + if sandbox_id is None: + if request.method == "POST": + record = service.create_record(body) + service.create_started.set() + await asyncio.to_thread(service.allow_create.wait, 5) + if service.create_status != 201: + return web.json_response( + {"detail": service.create_error, "sandboxId": record["id"]}, + status=service.create_status, + content_type="application/problem+json", + ) + return web.json_response(record, status=201) + state = request.query.get("state") + labels = [term.split("=", 1) for term in request.query.getall("labels", [])] + records = [ + record + for record in service.records.values() + if ( + record["state"] == state + if state + else record["state"] != "TERMINATED" + ) + and all(record["labels"].get(key) == value for key, value in labels) + ] + return web.json_response({"sandboxes": records}) + if sandbox_id not in service.records: + return failure(404, "missing sandbox") + record = service.records[sandbox_id] + operation = request.match_info.get("operation") + if request.method == "PATCH" or operation == "extend": + if ( + operation == "extend" + and body["maxLifetimeSeconds"] < record["maxLifetimeSeconds"] + ): + return failure(422, "extension cannot reduce lifetime") + for field, origin, deadline in ( + ("idleTimeoutSeconds", "lastActivityAt", "idleExpiresAt"), + ("maxLifetimeSeconds", "createdAt", "expiresAt"), + ): + if field in body: + record[field] = body[field] + start = datetime.fromisoformat( + record[origin].replace("Z", "+00:00") + ) + record[deadline] = ( + start + timedelta(seconds=body[field]) + ).isoformat() + return web.json_response(record) + if request.method == "DELETE": + if service.delete_status != 204: + return failure(service.delete_status, "cleanup unavailable") + record.update(state="TERMINATED", compute=None) + return web.Response(status=204) + if operation == "exec": + timeout = body.get("timeoutSeconds", 50) + background = body.get("background", False) + if ( + type(timeout) is not int + or not 1 <= timeout <= 50 + or type(background) is not bool + ): + return failure(422, "invalid execution controls") + status = service.exec_statuses.popleft() if service.exec_statuses else 200 + if status == 409 or record["state"] in ("FAILED", "TERMINATED"): + return failure(409, "container not started", service.conflict_code) + service.executed.append(body["command"]) + if status >= 400: + return failure(status, "response lost after command execution") + if service.exec_responses: + return service.exec_responses.popleft() + if body["command"] == ["wait-for-release"]: + async def run_job(): + service.job_started.set() + while not service.allow_job.is_set(): + if service.stopping.is_set(): + return + await asyncio.sleep(0.01) + service.job_completed.set() + + if background: + task = asyncio.create_task(run_job()) + service.jobs.add(task) + task.add_done_callback(service.jobs.discard) + return web.json_response({"output": "started", "exitCode": None}) + try: + await asyncio.wait_for(run_job(), timeout) + except asyncio.TimeoutError: + return web.json_response( + {"output": "started", "error": "execution timed out"} + ) + return web.json_response({"output": "finished", "exitCode": 0}) + if body["command"] == ["job-status"]: + return web.json_response( + { + "output": ( + "completed" if service.job_completed.is_set() else "pending" + ), + "exitCode": 0, + } + ) + result = ( + {"output": "partial output", "error": "command failed"} + if body["command"] == ["fail"] + else {"output": "completed"} + ) + return web.json_response(result) + if operation == "logs": + service.log_started.set() + await asyncio.to_thread(service.allow_logs.wait, 5) + status = service.log_statuses.popleft() if service.log_statuses else 200 + if status != 200: + return failure(status, "logs not ready", service.conflict_code) + response = web.StreamResponse(headers={"Content-Type": "text/event-stream"}) + try: + await response.prepare(request) + for byte in service.log_bytes: + await response.write(bytes([byte])) + await asyncio.sleep(0) + while service.hold_logs and not service.stopping.is_set(): + await response.write(b": heartbeat\n\n") + await asyncio.sleep(0.01) + except ConnectionResetError: + # clients may stop reading before the fixture finishes streaming. + pass + finally: + service.log_disconnected.set() + return response + return web.json_response(record) + + async def serve(): + service.loop = asyncio.get_running_loop() + app = web.Application() + for path in ( + "/v2/global-volumes", + "/v2/catalog/datacenters", + "/v2/sandboxes", + "/v2/network-volumes", + "/v2/registries", + "/v2/sandboxes/{id}", + "/v2/sandboxes/{id}/{operation}", + ): + app.router.add_route("*", path, handle) + runner = web.AppRunner(app, access_log=None, shutdown_timeout=1) + await runner.setup() + try: + await web.TCPSite(runner, "127.0.0.1", 0).start() + service.options = { + "api_key": "sandbox-test-key", + "base_url": f"http://127.0.0.1:{runner.addresses[0][1]}", + "request_timeout": 2, + "startup_timeout": 2, + } + ready.set() + await stopped.wait() + finally: + await runner.cleanup() + if service.jobs: + await asyncio.gather(*service.jobs) + + outcome = Future() + + def run_peer(): + try: + asyncio.run(serve()) + except ( + Exception, + asyncio.CancelledError, + KeyboardInterrupt, + SystemExit, + ) as error: + # cancellation and interrupts must reach the test thread. + outcome.set_exception(error) + else: + outcome.set_result(None) + + thread = threading.Thread(target=run_peer, daemon=True) + thread.start() + assert ready.wait(5), "local sandbox peer did not start" + try: + yield service + finally: + service.allow_create.set() + service.allow_logs.set() + service.allow_job.set() + service.stopping.set() + service.loop.call_soon_threadsafe(stopped.set) + thread.join(5) + assert not thread.is_alive(), "local sandbox peer did not stop" + outcome.result(timeout=0) + + +@pytest.fixture +def peer(): + with sandbox_peer() as service: + yield service + + +@pytest.mark.asyncio +async def test_transport_sessions_disable_connection_retries(): + api = AsyncSandboxAPI() + try: + session = api._get_session() + assert session._retry_connection is False + assert api._get_session() is session + await api.close() + assert session.closed + reopened = api._get_session() + assert reopened is not session + assert reopened._retry_connection is False + finally: + await api.close() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("vcpu_count", [0.5, 2.0, 2]) +async def test_compute_snapshot_preserves_numeric_vcpu_count(peer, vcpu_count): + assert get_type_hints(sandbox_domain.SandboxCompute)["vcpu_count"] is float + record = peer.create_record({}) + record["compute"]["vcpuCount"] = vcpu_count + sandbox = await AsyncioSandbox.get(record["id"], **peer.options) + try: + assert sandbox.compute.vcpu_count == vcpu_count + assert sandbox.info.compute is sandbox.compute + finally: + await sandbox.close() + + +@pytest.mark.asyncio +async def test_async_log_stream_matches_public_export(peer): + assert "AsyncioSandboxLogStream" in sandbox_domain.__all__ + assert ( + get_type_hints(AsyncioSandbox.logs)["return"] + is sandbox_domain.AsyncioSandboxLogStream + ) + async with AsyncioSandbox(image="image", **peer.options) as sandbox: + stream = sandbox.logs() + assert type(stream) is sandbox_domain.AsyncioSandboxLogStream + assert aiter(stream) is stream + opened_stream = await stream.open() + assert opened_stream is stream + async with stream as opened: + assert opened is stream + event = await anext(opened) + assert event.line == "café" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_api", [False, True], ids=["sync", "async"]) +async def test_execution_controls_bound_foreground_and_detach_background(peer, async_api): + api = AsyncioSandbox if async_api else Sandbox + + async def invoke(method, *args, **kwargs): + if async_api: + return await method(*args, **kwargs) + return await asyncio.to_thread(method, *args, **kwargs) + + sandbox = await invoke(api.create, image="image", **peer.options) + try: + for controls in ( + {"timeout_seconds": 0}, + {"timeout_seconds": 51}, + {"timeout_seconds": True}, + {"timeout_seconds": 1.5}, + {"background": 1}, + ): + with pytest.raises((TypeError, ValueError)): + await invoke(sandbox.exec, ["wait-for-release"], **controls) + assert peer.executed == [] + assert not any(path.endswith("/exec") for _, path, _, _ in peer.requests) + + with pytest.raises(SandboxExecutionError) as expired: + await invoke( + sandbox.exec, ["wait-for-release"], timeout_seconds=1, check=True + ) + assert expired.value.result.output == "started" + assert expired.value.result.error == "execution timed out" + assert not peer.job_completed.is_set() + + peer.allow_job.set() + completed = await invoke( + sandbox.exec, ["wait-for-release"], timeout_seconds=50, check=True + ) + assert completed.output == "finished" + assert completed.exit_code == 0 + assert peer.job_completed.is_set() + + peer.allow_job.clear() + peer.job_started.clear() + peer.job_completed.clear() + launched = await invoke( + sandbox.exec, + ["wait-for-release"], + timeout_seconds=1, + background=True, + check=True, + ) + assert launched.output == "started" + assert await asyncio.to_thread(peer.job_started.wait, 2) + assert not peer.job_completed.is_set() + assert (await invoke(sandbox.exec, ["job-status"])).output == "pending" + peer.allow_job.set() + assert await asyncio.to_thread(peer.job_completed.wait, 2) + assert (await invoke(sandbox.exec, ["job-status"])).output == "completed" + finally: + peer.allow_job.set() + await invoke(sandbox.terminate) + + +def test_sync_submission_racing_close_settles_and_allows_reuse(): + script = """ +import asyncio +import aiohttp +import threading +from concurrent.futures import Future +from runpod import Sandbox +from runpod.error import QueryError +from tests.test_sandbox import sandbox_peer + +with sandbox_peer() as peer: + sandbox = Sandbox.create(image="image", **peer.options) + loop = sandbox._runner._loop + schedule = loop.call_soon_threadsafe + submitting = threading.Event() + release_submission = threading.Event() + submitted = threading.Event() + stopping = threading.Event() + release_loop = threading.Event() + refreshed = Future() + closed = Future() + + def controlled_schedule(callback, *args, **kwargs): + if threading.current_thread().name == "racing-refresh": + submitting.set() + assert release_submission.wait(4), "submission was not released" + try: + return schedule(callback, *args, **kwargs) + finally: + submitted.set() + if callback == loop.stop: + schedule(release_loop.wait, 4) + result = schedule(callback, *args, **kwargs) + stopping.set() + return result + return schedule(callback, *args, **kwargs) + + def refresh(): + try: + refreshed.set_result(sandbox.refresh()) + except BaseException as error: + refreshed.set_exception(error) + + def close(): + try: + sandbox.close() + closed.set_result(None) + except BaseException as error: + closed.set_exception(error) + + loop.call_soon_threadsafe = controlled_schedule + reader = threading.Thread(target=refresh, name="racing-refresh", daemon=True) + closer = threading.Thread(target=close, daemon=True) + try: + reader.start() + assert submitting.wait(2), "refresh never reached submission" + closer.start() + # admission may exclude shutdown until the submission gate is released. + stopping.wait(1) + release_submission.set() + assert submitted.wait(2), "refresh submission did not resume" + release_loop.set() + try: + assert refreshed.result(timeout=3).state == "RUNNING" + except (RuntimeError, QueryError, asyncio.CancelledError, aiohttp.ClientConnectionError): + pass + closed.result(timeout=3) + reader.join(2) + closer.join(2) + assert not reader.is_alive(), "refresh was abandoned by shutdown" + assert not closer.is_alive(), "close did not finish" + assert sandbox.refresh().state == "RUNNING" + finally: + release_submission.set() + release_loop.set() + loop.call_soon_threadsafe = schedule + sandbox.terminate() + assert peer.records[sandbox.id]["state"] == "TERMINATED" +""" + process = subprocess.run( + [sys.executable, "-c", script], capture_output=True, text=True, timeout=15 + ) + assert process.returncode == 0, process.stdout + process.stderr + + +@pytest.mark.parametrize("delete_status", [204, 503]) +def test_sync_failed_creation_preserves_resource_id_and_cleans_up(peer, delete_status): + peer.create_status = 502 + peer.delete_status = delete_status + with pytest.raises(SandboxCreationError) as failed: + Sandbox.create(image="image", **peer.options) + error = failed.value + assert isinstance(error, QueryError) + assert error.sandbox_id == "sandbox-1" + assert error.status_code == 502 + assert peer.create_error in str(error) + assert [method for method, path, _, _ in peer.requests if path == "/v2/sandboxes"] == [ + "POST" + ] + assert [ + method + for method, path, _, _ in peer.requests + if path == "/v2/sandboxes/sandbox-1" + ] == ["DELETE"] + if delete_status == 204: + assert peer.records[error.sandbox_id]["state"] == "TERMINATED" + else: + assert isinstance(error.__cause__, QueryError) + assert error.__cause__.status_code == 503 + assert peer.records[error.sandbox_id]["state"] == "RUNNING" + peer.delete_status = 204 + Sandbox.get(error.sandbox_id, **peer.options).terminate() + assert peer.records[error.sandbox_id]["state"] == "TERMINATED" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("delete_status", [204, 503]) +async def test_async_failed_creation_preserves_resource_id_and_cleans_up( + peer, delete_status +): + peer.create_status = 502 + peer.delete_status = delete_status + with pytest.raises(SandboxCreationError) as failed: + await AsyncioSandbox.create(image="image", **peer.options) + error = failed.value + assert error.sandbox_id == "sandbox-1" + assert error.status_code == 502 + assert peer.create_error in str(error) + assert [method for method, path, _, _ in peer.requests if path == "/v2/sandboxes"] == [ + "POST" + ] + assert [ + method + for method, path, _, _ in peer.requests + if path == "/v2/sandboxes/sandbox-1" + ] == ["DELETE"] + if delete_status == 204: + assert peer.records[error.sandbox_id]["state"] == "TERMINATED" + else: + assert isinstance(error.__cause__, QueryError) + assert error.__cause__.status_code == 503 + peer.delete_status = 204 + recovery = await AsyncioSandbox.get(error.sandbox_id, **peer.options) + await recovery.terminate() + assert peer.records[error.sandbox_id]["state"] == "TERMINATED" + + +@pytest.mark.parametrize( + "operation, delete_status", + [ + ("create", 204), + ("enter", 204), + ("get", 204), + ("borrowed_enter", 204), + ("create", 503), + ("enter", 503), + ], +) +def test_sync_interrupt_at_successful_handle_handoff( + peer, monkeypatch, operation, delete_status +): + peer.delete_status = delete_status + borrowed = operation in ("get", "borrowed_enter") + handle = None + if borrowed: + peer.create_record({"imageName": "image"}) + if operation == "borrowed_enter": + handle = Sandbox.get("sandbox-1", **peer.options) + elif operation == "enter": + handle = Sandbox(image="image", **peer.options) + original_result = Future.result + caller = threading.get_ident() + interrupt = KeyboardInterrupt("interrupted after remote success") + armed = True + + def interrupt_result(future, *args, **kwargs): + nonlocal armed + result = original_result(future, *args, **kwargs) + if armed and threading.get_ident() == caller: + armed = False + raise interrupt + return result + + monkeypatch.setattr(Future, "result", interrupt_result) + try: + with pytest.raises(KeyboardInterrupt) as failed: + if operation == "create": + Sandbox.create(image="image", **peer.options) + elif operation == "get": + Sandbox.get("sandbox-1", **peer.options) + else: + handle.__enter__() + assert failed.value is interrupt + assert peer.records["sandbox-1"]["state"] == ( + "RUNNING" if borrowed or delete_status != 204 else "TERMINATED" + ) + deletes = [path for method, path, _, _ in peer.requests if method == "DELETE"] + assert deletes == ([] if borrowed else ["/v2/sandboxes/sandbox-1"]) + if delete_status != 204: + assert isinstance(failed.value.__cause__, QueryError) + assert failed.value.__cause__.status_code == 503 + peer.delete_status = 204 + Sandbox.get("sandbox-1", **peer.options).terminate() + assert peer.records["sandbox-1"]["state"] == "TERMINATED" + if operation == "borrowed_enter": + with handle: + assert handle.refresh().state == "RUNNING" + finally: + armed = False + if handle is not None: + handle.close() + + +def test_overlapping_sync_entries_preserve_the_first_owner(peer): + sandbox = Sandbox(image="image", **peer.options) + peer.allow_create.clear() + leave = threading.Event() + entered = threading.Event() + outcome = Future() + + def own_context(): + try: + with sandbox: + entered.set() + assert leave.wait(4), "test did not release the first context" + assert sandbox.refresh().state == "RUNNING" + outcome.set_result(None) + except ( + Exception, + asyncio.CancelledError, + KeyboardInterrupt, + SystemExit, + ) as error: + # cancellation and interrupts must reach the test thread. + outcome.set_exception(error) + + worker = threading.Thread(target=own_context, daemon=True) + worker.start() + try: + assert peer.create_started.wait(2) + with pytest.raises(RuntimeError): + sandbox.__enter__() + peer.allow_create.set() + assert entered.wait(3), "first context was disrupted by a rejected entry" + finally: + peer.allow_create.set() + leave.set() + worker.join(5) + assert not worker.is_alive(), "first context did not finish" + try: + outcome.result(timeout=0) + finally: + sandbox.close() + assert peer.records["sandbox-1"]["state"] == "TERMINATED" + assert [method for method, _, _, _ in peer.requests].count("POST") == 1 + + +def test_sync_startup_conflict_and_borrowed_context_ownership(peer): + peer.exec_statuses.extend([409, 200]) + with Sandbox(image="python:3.12-slim", **peer.options) as owner: + with Sandbox.get(owner.id, **peer.options) as borrowed: + assert borrowed.exec(["work"]).output == "completed" + assert peer.records[owner.id]["state"] == "RUNNING" + # A rejected nested entry must not destroy the outer context's client. + with pytest.raises(RuntimeError): + with owner: + pytest.fail("nested context entry succeeded") + assert owner.refresh().state == "RUNNING" + assert peer.records[owner.id]["state"] == "TERMINATED" + assert peer.executed == [["work"]] + + +@pytest.mark.asyncio +async def test_async_startup_conflict_and_borrowed_context_ownership(peer): + peer.exec_statuses.extend([409, 200]) + async with AsyncioSandbox(image="python:3.12-slim", **peer.options) as owner: + async with await AsyncioSandbox.get(owner.id, **peer.options) as borrowed: + assert (await borrowed.exec(["work"], check=True)).output == "completed" + assert peer.records[owner.id]["state"] == "RUNNING" + with pytest.raises(RuntimeError): + async with owner: + pytest.fail("nested context entry succeeded") + assert (await owner.refresh()).state == "RUNNING" + assert peer.records[owner.id]["state"] == "TERMINATED" + assert peer.executed == [["work"]] + + +@pytest.mark.asyncio +async def test_cancellation_during_creation_recovers_id_and_terminates(peer): + peer.allow_create.clear() + + async def create(): + async with AsyncioSandbox(image="python:3.12-slim", **peer.options): + pytest.fail("cancelled context was entered") + + task = asyncio.create_task(create()) + assert await asyncio.to_thread(peer.create_started.wait, 2) + task.cancel() + peer.allow_create.set() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(task, 4) + assert peer.records["sandbox-1"]["state"] == "TERMINATED" + + +def test_sync_interrupt_during_creation_waits_for_resource_cleanup(): + script = """ +import os +import signal +import threading +from runpod import Sandbox +from tests.test_sandbox import sandbox_peer + +with sandbox_peer() as peer: + peer.allow_create.clear() + def interrupt(): + assert peer.create_started.wait(2) + os.kill(os.getpid(), signal.SIGINT) + threading.Timer(0.1, peer.allow_create.set).start() + worker = threading.Thread(target=interrupt) + worker.start() + try: + Sandbox.create(image="python:3.12-slim", **peer.options) + raise AssertionError("SIGINT was swallowed") + except KeyboardInterrupt: + assert peer.records["sandbox-1"]["state"] == "TERMINATED" + finally: + worker.join() +""" + process = subprocess.run( + [sys.executable, "-c", script], capture_output=True, text=True, timeout=10 + ) + assert process.returncode == 0, process.stdout + process.stderr + + +@pytest.mark.parametrize("delete_status", [204, 503]) +@pytest.mark.parametrize("failure_type", [ValueError, KeyboardInterrupt, SystemExit]) +def test_sync_body_exception_remains_primary_during_cleanup( + peer, delete_status, failure_type +): + peer.delete_status = delete_status + failure = failure_type("application failed") + caught = None + try: + with Sandbox(image="python:3.12-slim", **peer.options): + raise failure + except failure_type as error: + caught = error + assert caught is failure + if delete_status == 204: + assert peer.records["sandbox-1"]["state"] == "TERMINATED" + else: + assert isinstance(caught.__cause__, QueryError) + assert caught.__cause__.status_code == 503 + assert peer.records["sandbox-1"]["state"] == "RUNNING" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("delete_status", [204, 503]) +async def test_async_body_exception_remains_primary_during_cleanup(peer, delete_status): + peer.delete_status = delete_status + failure = ValueError("application failed") + caught = None + try: + async with AsyncioSandbox(image="python:3.12-slim", **peer.options): + raise failure + except ValueError as error: + caught = error + assert caught is failure + if delete_status == 204: + assert peer.records["sandbox-1"]["state"] == "TERMINATED" + else: + assert isinstance(caught.__cause__, QueryError) + assert caught.__cause__.status_code == 503 + + +@pytest.mark.asyncio +async def test_exec_never_replays_ambiguous_failure_and_preserves_partial_output(peer): + async with AsyncioSandbox(image="python:3.12-slim", **peer.options) as sandbox: + peer.exec_statuses.append(500) + with pytest.raises(QueryError) as raised: + await sandbox.exec(["side-effect"]) + assert raised.value.status_code == 500 + assert peer.executed == [["side-effect"]] + result = await sandbox.exec(["fail"]) + assert result.output == "partial output" + assert result.error == "command failed" + with pytest.raises(SandboxExecutionError) as failure: + await sandbox.exec(["fail"], check=True) + assert failure.value.result == result + assert failure.value.sandbox_id == sandbox.id + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "body, content_type", + [ + ("{}", "application/json"), + ('{"output": null}', "application/json"), + ("proxy error", "text/html"), + ("[]", "application/json"), + ], +) +async def test_malformed_exec_response_is_a_query_error_without_replay( + peer, body, content_type +): + async with AsyncioSandbox(image="python:3.12-slim", **peer.options) as sandbox: + peer.exec_responses.append(web.Response(text=body, content_type=content_type)) + with pytest.raises(QueryError) as failure: + await sandbox.exec(["side-effect"]) + assert failure.value.query == f"POST /v2/sandboxes/{sandbox.id}/exec" + assert peer.executed == [["side-effect"]] + + +@pytest.mark.asyncio +async def test_startup_deadline_and_terminal_state_do_not_execute_commands(peer): + async with AsyncioSandbox(image="python:3.12-slim", **peer.options) as sandbox: + peer.exec_statuses.extend([409] * 20) + with pytest.raises(QueryError) as rejected: + await sandbox.exec(["work"], startup_timeout=0) + assert rejected.value.status_code == 409 + with pytest.raises(SandboxStartupTimeout) as expired: + await sandbox.exec(["work"], startup_timeout=0.03) + assert expired.value.sandbox_id == sandbox.id + peer.records[sandbox.id]["state"] = "FAILED" + with pytest.raises(SandboxStateError) as failed: + await sandbox.exec(["work"]) + assert failed.value.info.state == "FAILED" + assert peer.executed == [] + + +def test_sync_sse_decodes_fragments_and_retains_resume_cursor(peer): + with Sandbox(image="python:3.12-slim", **peer.options) as sandbox: + with sandbox.logs(source="container", tail=0) as logs: + events = list(logs) + assert [(event.source, event.line, event.id) for event in events] == [ + ("container", "caf\u00e9", "opaque/42"), + ("system", "started", "opaque/43"), + ] + assert events[0].timestamp.isoformat() == "2026-09-14T12:00:00+00:00" + with sandbox.logs(last_event_id=events[-1].id) as logs: + assert next(logs).line == "caf\u00e9" + log_requests = [ + request for request in peer.requests if request[1].endswith("/logs") + ] + assert log_requests[-1][3]["Last-Event-ID"] == "opaque/43" + + +@pytest.mark.asyncio +async def test_async_log_early_close_releases_live_connection(peer): + peer.hold_logs = True + # Exclude the timeout frame so this stays open until the client closes it. + peer.log_bytes = peer.log_bytes.split(b"event: timeout")[0] + peer.log_statuses.extend([409, 200]) + async with AsyncioSandbox(image="python:3.12-slim", **peer.options) as sandbox: + async with sandbox.logs() as logs: + event = await logs.__anext__() + assert event.line == "caf\u00e9" + assert await asyncio.to_thread(peer.log_disconnected.wait, 2) + + +@pytest.mark.asyncio +async def test_closing_sandbox_cancels_pending_log_handshake(peer): + peer.allow_logs.clear() + sandbox = await AsyncioSandbox.create(image="python:3.12-slim", **peer.options) + logs = sandbox.logs() + opening = asyncio.create_task(logs.open()) + try: + assert await asyncio.to_thread(peer.log_started.wait, 2) + await asyncio.wait_for(sandbox.close(), 3) + await asyncio.gather(opening, return_exceptions=True) + assert opening.cancelled() + assert peer.records[sandbox.id]["state"] == "RUNNING" + finally: + peer.allow_logs.set() + await sandbox.terminate() + + +@pytest.mark.asyncio +async def test_list_filters_and_independent_handles_do_not_delete_resources(peer): + first = peer.create_record( + {"imageName": "image", "labels": {"team": "a", "job": "42"}} + ) + second = peer.create_record( + {"imageName": "image", "labels": {"team": "a", "job": "43"}} + ) + peer.create_record({"imageName": "image", "labels": {"team": "b", "job": "42"}}) + matches = await AsyncioSandbox.list( + labels={"team": "a", "job": "42"}, **peer.options + ) + assert [sandbox.id for sandbox in matches] == [first["id"]] + await matches[0].close() + handles = await AsyncioSandbox.list(labels={"team": "a"}, **peer.options) + await handles[0].close() + async with handles[1] as borrowed: + assert (await borrowed.exec(["work"])).output == "completed" + assert borrowed.id == second["id"] + assert all(record["state"] == "RUNNING" for record in peer.records.values()) + + +@pytest.mark.asyncio +async def test_list_authentication_failure_propagates_after_cleanup(peer): + options = {**peer.options, "api_key": "invalid-key"} + with pytest.raises(AuthenticationError): + await AsyncioSandbox.list(**options) + + +def test_mixed_mounts_are_lazy_remote_bindings_with_explicit_transport(peer): + network = NetworkVolume("datasets", create=False) + global_volume = GlobalVolume("global-models") + peer.network_volumes = [ + {"id": "network-data", "name": "datasets", "dataCenter": "US-KS-2"} + ] + peer.global_volumes = [{"id": "gv-models", "name": "global-models"}] + peer.registries = [{"id": "registry-id", "name": "private-images"}] + sandbox = Sandbox( + image="private/image", + mounts={"/datasets/": network, "/models": global_volume}, + data_center_ids=["US-KS-2", "US-TX-3"], + registry_auth="private-images", + ports={8080: "http", 53: ("tcp", "udp")}, + disk_gb=20, + cmd=["python", "-m", "http.server"], + entrypoint=[], + start_ssh=True, + env={}, + **peer.options, + ) + assert peer.requests == [] + with sandbox: + with pytest.raises(VolumeError): + _ = network.path + with pytest.raises(VolumeError): + _ = global_volume.path + assert sandbox.info.mounts == { + "network": [{"volumeId": "network-data", "path": "/datasets"}], + "global": [{"volumeId": "gv-models", "path": "/models"}], + } + assert sandbox.info.data_center_id == "US-KS-2" + assert sandbox.info.registry == "registry-id" + assert sandbox.info.compute.container_disk_in_gb == 20 + assert sandbox.info.cmd == ["python", "-m", "http.server"] + assert sandbox.info.entrypoint == [] + assert sandbox.info.ports == [ + {"port": 8080, "protocol": "http"}, + {"port": 53, "protocol": "tcp"}, + {"port": 53, "protocol": "udp"}, + ] + create_body = next( + body + for method, path, body, _ in peer.requests + if path == "/v2/sandboxes" and method == "POST" + ) + assert create_body["dataCenterIds"] == ["US-KS-2"] + assert create_body["startSsh"] is True + assert "RUNPOD_MOUNTS" not in create_body["env"] + assert all( + headers["Authorization"] == "Bearer sandbox-test-key" + for _, _, _, headers in peer.requests + ) + assert not any(path.startswith("/v2/catalog") for _, path, _, _ in peer.requests) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("data_center_ids", [None, ["US-TX-3", "US-KS-2"]]) +async def test_global_mount_does_not_lookup_or_pin_network_placement( + peer, data_center_ids +): + peer.global_volumes = [{"id": "global-models", "name": "shared-models"}] + async with AsyncioSandbox( + image="image", + mounts={"/models": GlobalVolume("global-models")}, + data_center_ids=data_center_ids, + **peer.options, + ) as sandbox: + assert sandbox.info.mounts == { + "global": [{"volumeId": "global-models", "path": "/models"}] + } + create_body = next( + body + for method, path, body, _ in peer.requests + if path == "/v2/sandboxes" and method == "POST" + ) + if data_center_ids is None: + assert "dataCenterIds" not in create_body + else: + assert create_body["dataCenterIds"] == data_center_ids + assert not any( + path.startswith(("/v2/catalog", "/v2/network-volumes")) + for _, path, _, _ in peer.requests + ) + + +@pytest.mark.asyncio +async def test_global_creation_reuses_storage_by_name_and_id(peer): + volume = GlobalVolume("shared-models") + sandbox = AsyncioSandbox(image="image", mounts={"/models": volume}, **peer.options) + assert peer.requests == [] + async with sandbox: + assert sandbox.info.mounts == { + "global": [{"volumeId": "global-1", "path": "/models"}] + } + for reference in ("shared-models", "global-1"): + async with AsyncioSandbox( + image="image", + mounts={"/data": GlobalVolume(reference, create=False)}, + **peer.options, + ) as reused: + assert reused.info.mounts == { + "global": [{"volumeId": "global-1", "path": "/data"}] + } + assert peer.global_volumes == [{"id": "global-1", "name": "shared-models"}] + + +@pytest.mark.asyncio +async def test_missing_global_with_creation_disabled_prevents_provisioning(peer): + with pytest.raises(VolumeError): + await AsyncioSandbox.create( + image="image", + mounts={"/models": GlobalVolume("missing-models", create=False)}, + **peer.options, + ) + assert peer.global_volumes == [] + assert peer.records == {} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("volume_type", [GlobalVolume, NetworkVolume]) +async def test_sandbox_volume_lookup_reads_every_page(peer, volume_type): + records = [ + {"id": "first", "name": "first", "dataCenter": "US-KS-2"}, + {"id": "second", "name": "second", "dataCenter": "US-KS-2"}, + ] + if volume_type is GlobalVolume: + peer.global_volumes = records + path = "/v2/global-volumes" + else: + peer.network_volumes = records + path = "/v2/network-volumes" + peer.volume_page_size = 1 + async with AsyncioSandbox( + image="image", + mounts={"/data": volume_type("second", create=False)}, + **peer.options, + ) as sandbox: + assert sandbox.info.mounts[volume_type.kind] == [ + {"volumeId": "second", "path": "/data"} + ] + assert sum(request_path == path for _, request_path, _, _ in peer.requests) == 2 + assert all( + headers["Authorization"] == "Bearer sandbox-test-key" + for _, _, _, headers in peer.requests + ) + + +@pytest.mark.asyncio +async def test_sandbox_network_creation_requires_catalog_support(peer): + peer.volume_datacenters = [{"id": "US-KS-2", "networkVolumeTypes": []}] + with pytest.raises(VolumeError, match="does not support network volumes"): + await AsyncioSandbox.create( + image="image", + mounts={"/data": NetworkVolume("new-data", datacenter="US-KS-2")}, + **peer.options, + ) + assert peer.network_volumes == [] + assert peer.records == {} + + +@pytest.mark.asyncio +async def test_graphql_global_lookup_failure_prevents_provisioning(peer): + peer.global_error = "global store access denied" + with pytest.raises(QueryError): + await AsyncioSandbox.create( + image="image", + mounts={"/models": GlobalVolume("shared-models")}, + **peer.options, + ) + assert peer.global_volumes == [] + assert peer.records == {} + + +@pytest.mark.asyncio +async def test_omitted_and_empty_mounts_preserve_template_override_semantics(peer): + async with AsyncioSandbox(template_id="template", **peer.options): + assert "mounts" not in peer.requests[0][2] + async with AsyncioSandbox( + template_id="template", mounts={}, ports={}, cmd=[], **peer.options + ): + body = next( + body + for method, path, body, _ in reversed(peer.requests) + if method == "POST" + ) + assert body["mounts"] == {} + assert body["ports"] == [] + assert body["cmd"] == [] + assert "entrypoint" not in body + + +@pytest.mark.asyncio +async def test_network_creation_requires_explicit_placement_and_checks_requested_dc( + peer, +): + missing = NetworkVolume("new-dataset") + with pytest.raises(VolumeError): + await AsyncioSandbox.create( + image="image", mounts={"/data": missing}, **peer.options + ) + assert peer.records == {} + assert peer.network_volumes == [] + volume = NetworkVolume("new-dataset", size=25, datacenter="US-KS-2") + async with AsyncioSandbox( + image="image", mounts={"/data": volume}, **peer.options + ) as sandbox: + assert sandbox.info.data_center_id == "US-KS-2" + assert sandbox.info.mounts["network"] == [ + {"volumeId": "created-volume", "path": "/data"} + ] + count = len(peer.records) + with pytest.raises(ValueError): + await AsyncioSandbox.create( + image="image", + mounts={"/data": NetworkVolume("created-volume", create=False)}, + data_center_ids=["US-TX-3"], + **peer.options, + ) + assert len(peer.records) == count + + +def test_mount_and_port_collisions_are_rejected_before_provisioning(peer): + for mounts in ( + {"/data": NetworkVolume("first"), "/other": NetworkVolume("second")}, + {"/data": NetworkVolume("first"), "/data/models": GlobalVolume("second")}, + {"/etc": GlobalVolume("second")}, + {"/data": "untyped-volume"}, + ): + with pytest.raises((TypeError, ValueError, VolumeError)): + Sandbox(image="image", mounts=mounts, **peer.options) + with pytest.raises(ValueError): + Sandbox(image="image", ports={8080: ("http", "tcp")}, **peer.options) + assert peer.requests == [] + + +@pytest.mark.asyncio +async def test_snapshots_preserve_unavailable_versus_empty_configuration(peer): + async with AsyncioSandbox(image="image", **peer.options) as sandbox: + row = peer.records[sandbox.id] + row.update( + env=None, + ports=None, + mounts=None, + cmd=None, + entrypoint=None, + ssh=None, + startedAt=None, + ) + unavailable = await sandbox.refresh() + assert unavailable.env is None + assert unavailable.ports is None + assert unavailable.mounts is None + assert unavailable.cmd is None + assert unavailable.entrypoint is None + assert unavailable.ssh is None + assert unavailable.started_at is None + row.update( + env={}, + ports=[], + mounts={"network": [], "global": []}, + cmd=[], + entrypoint=[], + ssh={}, + ) + empty = await sandbox.refresh() + assert empty.env == {} + assert empty.ports == [] + assert empty.mounts == {"network": [], "global": []} + assert empty.cmd == [] + assert empty.entrypoint == [] + assert empty.ssh == {} + + +@pytest.mark.asyncio +async def test_rich_exec_results_preserve_nulls_and_check_exit_status(peer): + async with AsyncioSandbox(image="image", **peer.options) as sandbox: + peer.exec_responses.append( + web.json_response( + { + "output": "partial", + "stdout": None, + "stderr": "", + "exitCode": 3, + "durationMs": 0, + "truncated": False, + "error": None, + } + ) + ) + with pytest.raises(SandboxExecutionError) as failed: + await sandbox.exec(["fail-status"], check=True) + result = failed.value.result + assert result.output == "partial" + assert result.stdout is None + assert result.stderr == "" + assert result.exit_code == 3 + assert result.duration_ms == 0 + assert result.truncated is False + peer.exec_responses.append( + web.json_response( + { + "output": "partial", + "stdout": "partial", + "stderr": None, + "exitCode": None, + "durationMs": 4000, + "truncated": True, + "error": "execution timed out", + } + ) + ) + timed_out = await sandbox.exec(["long-command"]) + assert timed_out.exit_code is None + assert timed_out.stderr is None + assert timed_out.duration_ms == 4000 + assert timed_out.truncated is True + assert timed_out.error == "execution timed out" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("code", ["container_exited", "sandbox_terminated", None]) +async def test_non_startup_conflicts_are_not_retried_even_with_running_snapshot( + peer, code +): + async with AsyncioSandbox(image="image", **peer.options) as sandbox: + peer.conflict_code = code + peer.exec_statuses.append(409) + with pytest.raises(QueryError) as conflict: + await sandbox.exec(["work"]) + assert conflict.value.status_code == 409 + assert peer.executed == [] + assert [path for _, path, _, _ in peer.requests] == [ + "/v2/sandboxes", + f"/v2/sandboxes/{sandbox.id}/exec", + ] + + +def test_sync_extend_uses_fresh_total_lifetime_and_update_preserves_omissions(peer): + with Sandbox(image="image", **peer.options) as sandbox: + peer.records[sandbox.id]["maxLifetimeSeconds"] = 1200 + extended = sandbox.extend(seconds=300) + assert extended.max_lifetime_seconds == 1500 + assert extended.expires_at == extended.created_at + timedelta(seconds=1500) + updated = sandbox.update(idle_timeout_seconds=600) + assert updated.max_lifetime_seconds == 1500 + assert updated.idle_expires_at == updated.last_activity_at + timedelta( + seconds=600 + ) + assert updated.expires_at == extended.expires_at + + +@pytest.mark.asyncio +async def test_async_extensions_serialize_on_one_handle(peer): + async with AsyncioSandbox(image="image", **peer.options) as sandbox: + await sandbox.update(max_lifetime_seconds=1200) + await asyncio.gather(sandbox.extend(seconds=100), sandbox.extend(seconds=200)) + assert sandbox.info.max_lifetime_seconds == 1500 + assert sandbox.info.expires_at == sandbox.info.created_at + timedelta( + seconds=1500 + ) + + +def test_lifetime_updates_survive_close_and_event_loop_reuse(peer): + async def first_session(): + sandbox = await AsyncioSandbox.create(image="image", **peer.options) + try: + await sandbox.update(max_lifetime_seconds=1200) + await asyncio.gather( + sandbox.extend(seconds=100), sandbox.extend(seconds=200) + ) + finally: + await sandbox.close() + return sandbox + + sandbox = asyncio.run(first_session()) + + async def second_session(): + try: + await asyncio.gather( + sandbox.extend(seconds=100), sandbox.extend(seconds=200) + ) + finally: + await sandbox.terminate() + + asyncio.run(second_session()) + assert sandbox.info.max_lifetime_seconds == 1800 + + +@pytest.mark.asyncio +async def test_same_resolved_volume_id_cannot_be_mounted_as_two_kinds(peer): + peer.network_volumes = [ + {"id": "same-id", "name": "dataset", "dataCenter": "US-KS-2"} + ] + peer.global_volumes = [{"id": "same-id", "name": "shared-models"}] + with pytest.raises(ValueError): + await AsyncioSandbox.create( + image="image", + mounts={ + "/data": NetworkVolume("dataset", create=False), + "/models": GlobalVolume("same-id"), + }, + **peer.options, + ) + assert peer.records == {} diff --git a/tests/test_sandbox_contract.py b/tests/test_sandbox_contract.py new file mode 100644 index 000000000..34b26bd68 --- /dev/null +++ b/tests/test_sandbox_contract.py @@ -0,0 +1,114 @@ +"""Sandbox requests and snapshots use the current REST contract.""" + +import asyncio + +import pytest + +from runpod import AsyncioSandbox, Sandbox +from runpod.sandbox import SandboxInfo +from tests.test_sandbox import SandboxService, sandbox_peer + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_api", [False, True], ids=["sync", "async"]) +@pytest.mark.parametrize( + "options, expected_body", + [ + ({}, {}), + ({"image": "python:3.12-slim"}, {"image": "python:3.12-slim"}), + ({"template_id": "template"}, {"templateId": "template"}), + ( + {"cpu_flavor_id": "cpu3c", "vcpu_count": 2}, + {"cpu": {"id": "cpu3c", "vcpuCount": 2}}, + ), + ( + {"image": "image", "cpu_flavor_id": "cpu3c", "vcpu_count": 2}, + {"image": "image", "cpu": {"id": "cpu3c", "vcpuCount": 2}}, + ), + ], +) +async def test_create_wire_contract(async_api, options, expected_body, monkeypatch): + with sandbox_peer() as peer: + create_record = peer.create_record + + def modern_record(body): + record = create_record(body) + record.pop("imageName") + record.pop("cpuFlavorId") + record["image"] = body.get("image", "default-image") + record["cpu"] = { + "id": "cpu3c", + "vcpuCount": 2, + "memory": 4, + } + return record + + monkeypatch.setattr(peer, "create_record", modern_record) + if async_api: + async with AsyncioSandbox(**options, **peer.options) as sandbox: + info = sandbox.info + else: + + def create_sync(): + with Sandbox(**options, **peer.options) as sandbox: + return sandbox.info + + info = await asyncio.to_thread(create_sync) + + assert peer.requests[0][:3] == ("POST", "/v2/sandboxes", expected_body) + assert info.image == options.get("image", "default-image") + assert info.cpu_flavor_id == "cpu3c" + assert peer.records[info.id]["state"] == "TERMINATED" + + +@pytest.mark.parametrize("api", [Sandbox, AsyncioSandbox]) +def test_image_and_template_remain_mutually_exclusive(api): + with pytest.raises(ValueError, match="at most one"): + api(image="image", template_id="template") + + +@pytest.mark.parametrize("api", [Sandbox, AsyncioSandbox]) +@pytest.mark.parametrize("options", [{"cpu_flavor_id": "cpu3c"}, {"vcpu_count": 2}]) +def test_partial_cpu_selection_is_rejected_before_any_request(api, options): + with sandbox_peer() as peer: + with pytest.raises(ValueError, match="cpu_flavor_id and vcpu_count together"): + api(**options, **peer.options) + assert peer.requests == [] + + +@pytest.mark.parametrize("api", [Sandbox, AsyncioSandbox]) +def test_explicit_memory_constraint_is_not_a_create_option(api): + with sandbox_peer() as peer: + with pytest.raises(TypeError, match="memory_in_gb"): + api(memory_in_gb=4, **peer.options) + assert peer.requests == [] + + +@pytest.mark.parametrize( + "fields, expected_image, expected_cpu", + [ + ({"image": "image", "cpu": {"id": "cpu3c"}}, "image", "cpu3c"), + ({"imageName": "legacy", "cpuFlavorId": "cpu5c"}, None, None), + ( + { + "image": "image", + "imageName": "legacy", + "cpu": {"id": "cpu3c"}, + "cpuFlavorId": "cpu5c", + }, + "image", + "cpu3c", + ), + ({"image": None, "imageName": "legacy"}, None, None), + ({"cpu": None, "cpuFlavorId": "cpu5c"}, None, None), + ({}, None, None), + ], +) +def test_snapshot_reads_only_current_fields(fields, expected_image, expected_cpu): + record = SandboxService().create_record({}) + record.pop("imageName") + record.pop("cpuFlavorId") + record.update(fields) + info = SandboxInfo.from_dict(record) + assert info.image == expected_image + assert info.cpu_flavor_id == expected_cpu