diff --git a/examples/run-code-env-inheritance/Dockerfile b/examples/run-code-env-inheritance/Dockerfile new file mode 100644 index 000000000..ce505b55b --- /dev/null +++ b/examples/run-code-env-inheritance/Dockerfile @@ -0,0 +1,6 @@ +# syntax=docker/dockerfile:1.7 + +ARG SANDBOX_CODE_IMAGE=cube-sandbox-cn.tencentcloudcr.com/cube-sandbox/sandbox-code:latest +FROM ${SANDBOX_CODE_IMAGE} + +COPY lightweight-code-interpreter/server.py /opt/lightweight-code-interpreter/server.py diff --git a/examples/run-code-env-inheritance/README.md b/examples/run-code-env-inheritance/README.md new file mode 100644 index 000000000..bc8c04967 --- /dev/null +++ b/examples/run-code-env-inheritance/README.md @@ -0,0 +1,118 @@ +# Derived `sandbox-code` image: make `run_code` inherit sandbox-level environment variables + +[中文](README_zh.md) + +This example provides a derived image based on the official `sandbox-code` image. It replaces only the lightweight code interpreter so `run_code` can read sandbox-level environment variables injected through `Sandbox.create(envs=...)`, while preserving temporary per-call environment overrides. + +## Background + +Environment variables passed to `Sandbox.create(envs=...)` are available to `commands.run`, but `run_code` in the official `sandbox-code` image cannot read them because its lightweight code interpreter does not inject the sandbox environment from envd into the Jupyter kernel. + +This derived image aligns the environments inherited by `run_code` and `commands.run` without modifying the default `sandbox-code` image or any existing templates. + +## Environment semantics + +Before the first user execution in each Jupyter kernel, the lightweight code interpreter reads and caches the sandbox-level environment variables from `http://127.0.0.1:49983/envs`, then injects them into the kernel. Each `run_code` call applies its `env` or `env_vars` values afterward, with the following precedence: + +```text +Sandbox.create(envs=...) < per-call run_code environment variables +``` + +Sandbox-level values and per-call overrides are applied in a separate background kernel execution before the user code, so a compilation failure in the first user cell does not prevent environment initialization. Before applying per-call values, the interpreter snapshots the previous kernel value of each affected key. Afterward, a background cleanup restores the environment: + +- Keys that existed before the call are restored to their previous kernel values. +- Keys that did not exist before the call are removed. + +The next execution waits for pending cleanup so per-call values cannot leak into later executions. This lifecycle follows E2B code-interpreter. Sandbox-level environment variables are loaded only once per Jupyter kernel; this example does not provide a general runtime refresh mechanism. + +This derived interpreter requires envd's `/envs` endpoint. If envd is unavailable, the endpoint is unsupported, the request exceeds `ENVD_TIMEOUT` (2 seconds by default), or the payload is invalid, `run_code` fails with HTTP 502 instead of falling back to the stock interpreter behavior. Set `ENVD_TIMEOUT` in the image environment to adjust the fetch timeout. + +## Build and test the image + +From the repository root: + +```bash +docker build \ + -t sandbox-code-env-inheritance:latest \ + examples/run-code-env-inheritance +``` + +Override `SANDBOX_CODE_IMAGE` when using the international registry or a pinned base image: + +```bash +docker build \ + --build-arg SANDBOX_CODE_IMAGE=cube-sandbox-int.tencentcloudcr.com/cube-sandbox/sandbox-code:latest \ + -t sandbox-code-env-inheritance:latest \ + examples/run-code-env-inheritance +``` + +After building the image, run the focused unit tests inside it so they use the same Python dependencies as the runtime: + +```bash +docker run --rm \ + --entrypoint python \ + -v "$PWD/examples/run-code-env-inheritance:/work:ro" \ + -e PYTHONPATH=/work/lightweight-code-interpreter \ + sandbox-code-env-inheritance:latest \ + -m unittest discover -s /work/tests -v +``` + +## Create a CubeSandbox template + +Push the image to a registry reachable by CubeMaster, then create the template: + +```bash +cubemastercli tpl create-from-image \ + --image /sandbox-code-env-inheritance:latest \ + --writable-layer-size 1G \ + --expose-port 49983 \ + --expose-port 49999 \ + --probe 49999 \ + --probe-path /health +``` + +Watch the asynchronous build with the `job_id` returned by the create command: + +```bash +cubemastercli tpl watch --job-id +``` + +### Use a local image in one-click deployments + +If CubeMaster cannot access an image registry, build or load `sandbox-code-env-inheritance:latest` into Docker on the one-click control-plane host. Add the following setting to `/usr/local/services/cubetoolbox/.one-click.env`: + +```bash +CUBEMASTER_NATIVE_ROOTFS_EXPORT_ENABLED=false +``` + +Restart CubeMaster to apply the setting: + +```bash +systemctl restart cube-sandbox-cubemaster.service +``` + +This disables the default native rootfs exporter and falls back to the Docker-based exporter, which can consume the local Docker image. Create the template with the local image tag: + +```bash +cubemastercli tpl create-from-image \ + --image sandbox-code-env-inheritance:latest \ + --writable-layer-size 1G \ + --expose-port 49983 \ + --expose-port 49999 \ + --probe 49999 \ + --probe-path /health +``` + +## Run the SDK compatibility E2E test + +After creating the template, run the shared CubeSandbox/E2B compatibility case to verify sandbox-level inheritance and per-call merge, override, and cleanup behavior: + +```bash +export CUBE_TEMPLATE_ID= +cd tests/e2e/sdk_compat +SDK_E2E_RUN_CODE_ENV_INHERITANCE=true \ +SDK_E2E_BACKENDS=e2b,cubesandbox pytest --run-e2e \ + cases/run_code/test_python.py::test_run_code_merges_create_and_per_call_envs -q +``` + +`CUBE_TEMPLATE_ID` reuses the SDK compatibility suite's existing template selection mechanism; `--cube-template-id ` is also supported. The environment-inheritance case is skipped by default before creating a sandbox and runs only when `SDK_E2E_RUN_CODE_ENV_INHERITANCE=true` is explicitly set for a compatible template. diff --git a/examples/run-code-env-inheritance/README_zh.md b/examples/run-code-env-inheritance/README_zh.md new file mode 100644 index 000000000..935ccaeb4 --- /dev/null +++ b/examples/run-code-env-inheritance/README_zh.md @@ -0,0 +1,118 @@ +# `sandbox-code` 派生镜像：让 `run_code` 继承沙箱级环境变量 + +[English](README.md) + +本示例提供一个基于官方 `sandbox-code` 的派生镜像，仅替换其中的 lightweight code interpreter，让 `run_code` 读取通过 `Sandbox.create(envs=...)` 注入的沙箱环境变量，并保留单次调用环境变量的临时覆盖语义。 + +## 背景 + +创建沙箱时，可以通过 `Sandbox.create(envs=...)` 注入环境变量。这些变量可以被 `commands.run` 读取，但官方 `sandbox-code` 镜像中的 `run_code` 默认无法读取，因为其 lightweight code interpreter 不会将 envd 中的沙箱环境变量注入 Jupyter kernel。 + +本派生镜像补齐该行为，使 `run_code` 和 `commands.run` 继承相同的沙箱环境变量，同时不修改默认 `sandbox-code` 镜像或已有模板。 + +## 环境变量语义 + +每个 Jupyter kernel 第一次执行用户代码前，lightweight code interpreter 会从 `http://127.0.0.1:49983/envs` 读取并缓存当前沙箱的环境变量，然后将其注入 kernel。每次 `run_code` 调用再叠加本次传入的 `env` 或 `env_vars`，优先级如下： + +```text +Sandbox.create(envs=...) < run_code 单次调用环境变量 +``` + +沙箱级值和单次调用覆盖会在用户代码执行前，通过单独的后台 kernel execution 注入，因此首个用户 cell 编译失败不会阻止环境初始化。如果两层环境中存在同名变量，本次执行使用单次调用传入的值。在应用单次调用值前，解释器会记录每个受影响变量在 kernel 中的原有状态。执行结束后，后台清理任务会恢复环境： + +- 调用前已存在的变量会恢复为原有 kernel 值。 +- 调用前不存在的变量会被删除。 + +下一次执行会先等待上一轮清理完成，避免单次调用环境变量泄漏到后续执行中。该生命周期与 E2B code-interpreter 保持一致。沙箱环境变量在每个 Jupyter kernel 中只读取一次，本示例不提供通用的运行时刷新机制。 + +本派生解释器依赖 envd 的 `/envs` 接口。如果 envd 不可用、接口不受支持、请求超过 `ENVD_TIMEOUT`（默认 2 秒），或返回的数据无效，`run_code` 会返回 HTTP 502，不会回退到官方解释器的原有行为。可以通过镜像环境变量 `ENVD_TIMEOUT` 调整读取超时时间。 + +## 构建并测试镜像 + +在仓库根目录执行： + +```bash +docker build \ + -t sandbox-code-env-inheritance:latest \ + examples/run-code-env-inheritance +``` + +如果需要使用国际站镜像仓库，或指定固定版本的基础镜像，可以覆盖 `SANDBOX_CODE_IMAGE`： + +```bash +docker build \ + --build-arg SANDBOX_CODE_IMAGE=cube-sandbox-int.tencentcloudcr.com/cube-sandbox/sandbox-code:latest \ + -t sandbox-code-env-inheritance:latest \ + examples/run-code-env-inheritance +``` + +构建完成后，可以直接在镜像内运行聚焦单元测试，以复用实际运行时中的 Python 依赖： + +```bash +docker run --rm \ + --entrypoint python \ + -v "$PWD/examples/run-code-env-inheritance:/work:ro" \ + -e PYTHONPATH=/work/lightweight-code-interpreter \ + sandbox-code-env-inheritance:latest \ + -m unittest discover -s /work/tests -v +``` + +## 创建 CubeSandbox 模板 + +将镜像推送到 CubeMaster 可以访问的镜像仓库，然后创建模板： + +```bash +cubemastercli tpl create-from-image \ + --image /sandbox-code-env-inheritance:latest \ + --writable-layer-size 1G \ + --expose-port 49983 \ + --expose-port 49999 \ + --probe 49999 \ + --probe-path /health +``` + +创建命令会返回 `job_id`。使用该 ID 等待异步构建完成： + +```bash +cubemastercli tpl watch --job-id +``` + +### one-click 部署中使用本地镜像 + +如果 CubeMaster 无法访问镜像仓库，可以使用 one-click 控制面节点上的本地 Docker 镜像。先在该节点构建或加载 `sandbox-code-env-inheritance:latest`，再将以下配置加入 `/usr/local/services/cubetoolbox/.one-click.env`： + +```bash +CUBEMASTER_NATIVE_ROOTFS_EXPORT_ENABLED=false +``` + +重启 CubeMaster，使配置生效： + +```bash +systemctl restart cube-sandbox-cubemaster.service +``` + +该配置会关闭默认的 native rootfs exporter，并回退到能够读取本地 Docker 镜像的 Docker-based exporter。之后可以直接使用本地镜像标签创建模板： + +```bash +cubemastercli tpl create-from-image \ + --image sandbox-code-env-inheritance:latest \ + --writable-layer-size 1G \ + --expose-port 49983 \ + --expose-port 49999 \ + --probe 49999 \ + --probe-path /health +``` + +## 运行 SDK 兼容性 E2E + +创建模板后，可以运行 CubeSandbox/E2B 共用的 SDK 兼容性用例，验证沙箱环境变量与单次调用环境变量的合并、覆盖和清理行为： + +```bash +export CUBE_TEMPLATE_ID= +cd tests/e2e/sdk_compat +SDK_E2E_RUN_CODE_ENV_INHERITANCE=true \ +SDK_E2E_BACKENDS=e2b,cubesandbox pytest --run-e2e \ + cases/run_code/test_python.py::test_run_code_merges_create_and_per_call_envs -q +``` + +`CUBE_TEMPLATE_ID` 复用 SDK 兼容性测试套件已有的模板选择机制，也可以改用 `--cube-template-id `。环境变量继承用例默认在创建沙箱前跳过，只有确认所选模板支持该行为，并显式设置 `SDK_E2E_RUN_CODE_ENV_INHERITANCE=true` 时才会运行。 diff --git a/examples/run-code-env-inheritance/lightweight-code-interpreter/server.py b/examples/run-code-env-inheritance/lightweight-code-interpreter/server.py new file mode 100644 index 000000000..42eb7a28f --- /dev/null +++ b/examples/run-code-env-inheritance/lightweight-code-interpreter/server.py @@ -0,0 +1,576 @@ +from __future__ import annotations + +import asyncio +import datetime as dt +import json +import logging +import os +import uuid +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from pathlib import Path +from typing import Any + +import httpx +from fastapi import FastAPI, HTTPException +from fastapi.middleware.cors import CORSMiddleware +from pydantic import BaseModel, Field +from starlette.responses import StreamingResponse +from websockets.client import WebSocketClientProtocol, connect + + +JUPYTER_BASE_URL = os.getenv("JUPYTER_BASE_URL", "http://127.0.0.1:8888") +JUPYTER_WS_URL = JUPYTER_BASE_URL.replace("http://", "ws://").replace("https://", "wss://") +ENVD_ENVS_URL = f"http://127.0.0.1:{os.getenv('ENVD_PORT', '49983')}/envs" +ENVD_TIMEOUT = float(os.getenv("ENVD_TIMEOUT", "2")) +WORKDIR = Path(os.getenv("CODE_INTERPRETER_WORKDIR", "/workspace")).resolve() +DEFAULT_LANGUAGE = "python" +PING_TIMEOUT = 30 + +logging.basicConfig(level=os.getenv("CODE_INTERPRETER_LOG_LEVEL", "INFO")) +logger = logging.getLogger("lightweight-code-interpreter") + + +class ExecuteRequest(BaseModel): + code: str = Field(..., min_length=1) + context_id: str | None = None + language: str | None = None + timeout: float | None = None + cwd: str | None = None + env: dict[str, str] | None = None + env_vars: dict[str, str] | None = None + + +class CreateContext(BaseModel): + cwd: str | None = None + language: str | None = DEFAULT_LANGUAGE + + +class Context(BaseModel): + id: str + language: str + cwd: str + + +def _event(event_type: str, **payload: Any) -> str: + return json.dumps({"type": event_type, **payload}, ensure_ascii=False) + "\n" + + +def _normalize_language(language: str | None) -> str: + value = (language or DEFAULT_LANGUAGE).lower().strip() + if value in {"python", "python3", "py"}: + return DEFAULT_LANGUAGE + raise HTTPException(status_code=400, detail="only python language is supported") + + +def _resolve_cwd(cwd: str | None) -> str: + path = Path(cwd).resolve() if cwd else WORKDIR + try: + path.relative_to(WORKDIR) + except ValueError as exc: + raise HTTPException(status_code=400, detail="cwd must be under CODE_INTERPRETER_WORKDIR") from exc + path.mkdir(parents=True, exist_ok=True) + return str(path) + + +async def _fetch_sandbox_envs( + http_client: httpx.AsyncClient | None = None, +) -> dict[str, str]: + owns_client = http_client is None + if http_client is None: + http_client = httpx.AsyncClient() + + try: + response = await http_client.get(ENVD_ENVS_URL, timeout=ENVD_TIMEOUT) + response.raise_for_status() + except httpx.HTTPError as exc: + raise HTTPException( + status_code=502, + detail="failed to read sandbox environment from envd", + ) from exc + finally: + if owns_client: + await http_client.aclose() + + try: + payload = response.json() + except ValueError as exc: + raise HTTPException( + status_code=502, + detail="envd returned an invalid environment payload", + ) from exc + + if not isinstance(payload, dict) or not all( + isinstance(key, str) and isinstance(value, str) + for key, value in payload.items() + ): + raise HTTPException( + status_code=502, + detail="envd returned an invalid environment payload", + ) + return payload + + +def _format_result(data: dict[str, Any]) -> dict[str, Any]: + text = data.pop("text/plain", None) + if text and ( + (text.startswith("'") and text.endswith("'")) + or (text.startswith('"') and text.endswith('"')) + ): + text = text[1:-1] + return { + "text": text, + "html": data.pop("text/html", None), + "markdown": data.pop("text/markdown", None), + "svg": data.pop("image/svg+xml", None), + "png": data.pop("image/png", None), + "jpeg": data.pop("image/jpeg", None), + "pdf": data.pop("application/pdf", None), + "latex": data.pop("text/latex", None), + "json": data.pop("application/json", None), + "javascript": data.pop("application/javascript", None), + "data": data.pop("e2b/data", None), + "chart": data.pop("e2b/chart", None), + "extra": data or None, + } + + +class KernelContext: + def __init__(self, context_id: str, session_id: str, cwd: str): + self.context_id = context_id + self.session_id = session_id + self.cwd = cwd + self.language = DEFAULT_LANGUAGE + self.ws_url = f"{JUPYTER_WS_URL}/api/kernels/{context_id}/channels" + self.ws: WebSocketClientProtocol | None = None + self.receive_task: asyncio.Task | None = None + self.executions: dict[str, asyncio.Queue[dict[str, Any]]] = {} + self.lock = asyncio.Lock() + self.env_lock = asyncio.Lock() + self.sandbox_envs: dict[str, str] | None = None + self.sandbox_envs_applied = False + self.cleanup_task: asyncio.Task[None] | None = None + + async def connect(self) -> None: + ws_logger = logging.getLogger("websockets.client") + ws_logger.setLevel(logging.ERROR) + self.ws = await connect( + self.ws_url, + ping_timeout=PING_TIMEOUT, + max_size=None, + max_queue=None, + logger=ws_logger, + ) + self.receive_task = asyncio.create_task(self._receive_messages()) + await self._run_background(f"%cd {self.cwd}") + + async def _run_background(self, code: str) -> None: + if self.ws is None: + raise RuntimeError("kernel websocket is not connected") + + msg_id = str(uuid.uuid4()) + queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue() + self.executions[msg_id] = queue + + try: + await self.ws.send(self._execute_request(msg_id, code, silent=True)) + while True: + item = await queue.get() + if item["type"] == "end_of_execution": + break + if item["type"] == "unexpected_end_of_execution": + raise RuntimeError("background execution ended unexpectedly") + if item["type"] == "error": + raise RuntimeError(f"background execution failed: {item}") + finally: + self.executions.pop(msg_id, None) + + async def load_sandbox_envs(self) -> None: + async with self.env_lock: + if self.sandbox_envs is not None: + return + self.sandbox_envs = await _fetch_sandbox_envs() + + @staticmethod + def _build_env_assignments(env_vars: dict[str, str]) -> list[str]: + return [ + f"import os; os.environ[{json.dumps(key)}] = {json.dumps(value)}" + for key, value in env_vars.items() + ] + + def _build_env_setup_code( + self, + per_call_envs: dict[str, str] | None, + snapshot_name: str | None, + ) -> str: + lines: list[str] = [] + if not self.sandbox_envs_applied: + lines.extend(self._build_env_assignments(self.sandbox_envs or {})) + if per_call_envs: + if snapshot_name is None: + raise ValueError("snapshot_name is required for per-call envs") + keys = json.dumps(list(per_call_envs)) + lines.append( + f"import os; {snapshot_name} = " + f"{{key: (key in os.environ, os.environ.get(key)) for key in {keys}}}" + ) + lines.extend(self._build_env_assignments(per_call_envs or {})) + return "\n".join(lines) + + @staticmethod + def _build_env_cleanup_code(snapshot_name: str) -> str: + restore_name = f"{snapshot_name}_restore" + return "\n".join( + [ + f"def {restore_name}():", + " import os", + f" snapshot = globals().pop({json.dumps(snapshot_name)}, {{}})", + " for key, (existed, value) in snapshot.items():", + " if existed:", + " os.environ[key] = value", + " else:", + " os.environ.pop(key, None)", + f"{restore_name}()", + f"del {restore_name}", + ] + ) + + async def _cleanup_env_vars(self, snapshot_name: str) -> None: + await self._run_background(self._build_env_cleanup_code(snapshot_name)) + + async def _rollback_env_setup(self, snapshot_name: str) -> None: + try: + await self._cleanup_env_vars(snapshot_name) + except Exception as exc: + logger.warning("per-call environment setup rollback failed: %s", exc) + + async def _wait_for_cleanup(self) -> None: + if self.cleanup_task is None: + return + try: + await self.cleanup_task + except Exception as exc: + logger.warning("per-call environment cleanup failed: %s", exc) + finally: + self.cleanup_task = None + + def _execute_request(self, msg_id: str, code: str, *, silent: bool = False) -> str: + return json.dumps( + { + "header": { + "msg_id": msg_id, + "username": "e2b", + "session": self.session_id, + "msg_type": "execute_request", + "version": "5.3", + "date": dt.datetime.now(dt.timezone.utc).isoformat(), + }, + "parent_header": {}, + "metadata": { + "trusted": True, + "deletedCells": [], + "recordTiming": False, + "cellId": str(uuid.uuid4()), + }, + "content": { + "code": code, + "silent": silent, + "store_history": not silent, + "user_expressions": {}, + "stop_on_error": True, + "allow_stdin": False, + }, + } + ) + + async def execute( + self, + code: str, + *, + env_vars: dict[str, str] | None, + ) -> AsyncIterator[dict[str, Any]]: + if self.ws is None: + await self.connect() + + async with self.lock: + await self._wait_for_cleanup() + assert self.ws is not None + apply_sandbox_envs = not self.sandbox_envs_applied + snapshot_name = ( + f"_cube_lci_env_snapshot_{uuid.uuid4().hex}" + if env_vars + else None + ) + env_code = self._build_env_setup_code(env_vars, snapshot_name) + try: + if env_code: + await self._run_background(env_code) + except BaseException: + if snapshot_name is not None: + await self._rollback_env_setup(snapshot_name) + raise + if apply_sandbox_envs: + self.sandbox_envs_applied = True + + msg_id = str(uuid.uuid4()) + queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue() + self.executions[msg_id] = queue + + try: + await self.ws.send(self._execute_request(msg_id, code)) + + while True: + item = await queue.get() + if item["type"] == "end_of_execution": + break + if item["type"] == "unexpected_end_of_execution": + yield { + "type": "error", + "name": "UnexpectedEndOfExecution", + "value": "Connection to the execution was closed before the execution finished", + "traceback": "", + } + break + yield item + finally: + self.executions.pop(msg_id, None) + if snapshot_name is not None: + self.cleanup_task = asyncio.create_task( + self._cleanup_env_vars(snapshot_name) + ) + + async def _receive_messages(self) -> None: + if self.ws is None: + return + try: + async for raw in self.ws: + await self._process_message(json.loads(raw)) + except Exception as exc: + logger.warning("kernel websocket closed: %s", exc) + finally: + for queue in self.executions.values(): + await queue.put({"type": "unexpected_end_of_execution"}) + + async def _process_message(self, data: dict[str, Any]) -> None: + parent_msg_id = data.get("parent_header", {}).get("msg_id") + if not parent_msg_id: + return + + queue = self.executions.get(parent_msg_id) + if queue is None: + return + + msg_type = data.get("msg_type") + content = data.get("content", {}) + + if msg_type == "stream": + stream_type = "stdout" if content.get("name") == "stdout" else "stderr" + await queue.put( + { + "type": stream_type, + "text": content.get("text", ""), + "timestamp": data.get("header", {}).get("date"), + } + ) + elif msg_type in {"display_data", "execute_result"}: + await queue.put( + { + "type": "result", + "is_main_result": msg_type == "execute_result", + **_format_result(dict(content.get("data", {}))), + } + ) + elif msg_type == "error": + await queue.put( + { + "type": "error", + "name": content.get("ename", ""), + "value": content.get("evalue", ""), + "traceback": "".join(content.get("traceback", [])), + } + ) + elif msg_type == "execute_input": + await queue.put( + { + "type": "number_of_executions", + "execution_count": content.get("execution_count", 0), + } + ) + elif msg_type == "execute_reply" and content.get("status") == "abort": + await queue.put( + { + "type": "error", + "name": "ExecutionAborted", + "value": "Execution was aborted", + "traceback": "", + } + ) + await queue.put({"type": "end_of_execution"}) + elif msg_type == "status" and content.get("execution_state") == "idle": + await queue.put({"type": "end_of_execution"}) + + async def close(self) -> None: + if self.cleanup_task is not None: + if not self.cleanup_task.done(): + self.cleanup_task.cancel() + try: + await self.cleanup_task + except asyncio.CancelledError: + pass + except Exception as exc: + logger.warning("per-call environment cleanup failed during close: %s", exc) + self.cleanup_task = None + if self.ws is not None: + await self.ws.close() + if self.receive_task is not None: + self.receive_task.cancel() + + +async def _wait_for_jupyter() -> None: + async with httpx.AsyncClient(timeout=2) as http_client: + for _ in range(120): + try: + response = await http_client.get(f"{JUPYTER_BASE_URL}/api/status") + if response.is_success: + return + except Exception: + pass + await asyncio.sleep(0.5) + raise RuntimeError("Jupyter Server did not become ready") + + +contexts: dict[str, KernelContext] = {} +default_context_id: str | None = None +client: httpx.AsyncClient | None = None + + +async def create_context(language: str | None = None, cwd: str | None = None) -> Context: + _normalize_language(language) + resolved_cwd = _resolve_cwd(cwd) + assert client is not None + + response = await client.post( + f"{JUPYTER_BASE_URL}/api/sessions", + json={ + "path": str(uuid.uuid4()), + "kernel": {"name": "python3"}, + "type": "notebook", + "name": str(uuid.uuid4()), + }, + ) + if not response.is_success: + raise HTTPException(status_code=500, detail=f"failed to create context: {response.text}") + + session = response.json() + context_id = session["kernel"]["id"] + kernel = KernelContext(context_id, session["id"], resolved_cwd) + await kernel.connect() + contexts[context_id] = kernel + return Context(id=context_id, language=DEFAULT_LANGUAGE, cwd=resolved_cwd) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + global client, default_context_id + await _wait_for_jupyter() + client = httpx.AsyncClient() + default = await create_context(DEFAULT_LANGUAGE, str(WORKDIR)) + default_context_id = default.id + try: + yield + finally: + for context in list(contexts.values()): + await context.close() + contexts.clear() + if client is not None: + await client.aclose() + + +app = FastAPI(title="Lightweight Python Code Interpreter", version="0.2.0", lifespan=lifespan) +app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) + + +@app.get("/") +async def root() -> dict[str, Any]: + return {"ok": True, "service": "lightweight-python-code-interpreter"} + + +@app.get("/health") +async def health() -> str: + return "OK" + + +@app.post("/execute") +async def execute(exec_request: ExecuteRequest) -> StreamingResponse: + if exec_request.context_id and exec_request.language: + raise HTTPException(status_code=400, detail="only one of context_id or language can be provided") + + context_id = exec_request.context_id or default_context_id + if exec_request.language: + _normalize_language(exec_request.language) + if exec_request.cwd and not exec_request.context_id: + context = await create_context(DEFAULT_LANGUAGE, exec_request.cwd) + context_id = context.id + + if context_id is None or context_id not in contexts: + raise HTTPException(status_code=404, detail=f"context {exec_request.context_id} not found") + + env_vars = {} + if exec_request.env: + env_vars.update(exec_request.env) + if exec_request.env_vars: + env_vars.update(exec_request.env_vars) + + context = contexts[context_id] + await context.load_sandbox_envs() + + async def stream() -> AsyncIterator[str]: + async for item in context.execute(exec_request.code, env_vars=env_vars or None): + yield json.dumps(item, ensure_ascii=False) + "\n" + yield _event("end_of_execution") + + return StreamingResponse(stream(), media_type="application/x-ndjson") + + +@app.post("/contexts") +async def post_contexts(request: CreateContext) -> Context: + return await create_context(request.language, request.cwd) + + +@app.get("/contexts") +async def get_contexts() -> list[Context]: + return [ + Context(id=context.context_id, language=context.language, cwd=context.cwd) + for context in contexts.values() + ] + + +@app.post("/contexts/{context_id}/restart") +async def restart_context(context_id: str) -> None: + context = contexts.get(context_id) + if context is None or client is None: + raise HTTPException(status_code=404, detail=f"context {context_id} not found") + + await context.close() + response = await client.post(f"{JUPYTER_BASE_URL}/api/kernels/{context_id}/restart") + if not response.is_success: + raise HTTPException(status_code=500, detail=f"failed to restart context {context_id}") + + restarted = KernelContext(context.context_id, context.session_id, context.cwd) + await restarted.connect() + contexts[context_id] = restarted + + +@app.delete("/contexts/{context_id}") +async def remove_context(context_id: str) -> None: + context = contexts.get(context_id) + if context is None or client is None: + raise HTTPException(status_code=404, detail=f"context {context_id} not found") + if context_id == default_context_id: + raise HTTPException(status_code=400, detail="default context cannot be deleted") + + await context.close() + response = await client.delete(f"{JUPYTER_BASE_URL}/api/kernels/{context_id}") + if not response.is_success: + raise HTTPException(status_code=500, detail=f"failed to remove context {context_id}") + contexts.pop(context_id, None) diff --git a/examples/run-code-env-inheritance/tests/test_server.py b/examples/run-code-env-inheritance/tests/test_server.py new file mode 100644 index 000000000..76723d099 --- /dev/null +++ b/examples/run-code-env-inheritance/tests/test_server.py @@ -0,0 +1,388 @@ +from __future__ import annotations + +import asyncio +import json +import os +import unittest +from unittest.mock import AsyncMock, patch + +import httpx +from fastapi import HTTPException + +import server + + +class FetchSandboxEnvsTest(unittest.IsolatedAsyncioTestCase): + async def test_uses_configured_envd_timeout(self) -> None: + response = httpx.Response( + 200, + json={}, + request=httpx.Request("GET", server.ENVD_ENVS_URL), + ) + client = AsyncMock() + client.get.return_value = response + + with patch.object(server, "ENVD_TIMEOUT", 7.5): + self.assertEqual(await server._fetch_sandbox_envs(client), {}) + + client.get.assert_awaited_once_with(server.ENVD_ENVS_URL, timeout=7.5) + + async def test_returns_envd_environment(self) -> None: + def handler(request: httpx.Request) -> httpx.Response: + self.assertEqual(request.url.path, "/envs") + return httpx.Response(200, json={"BASE": "sandbox", "JSON": '{"n":1}'}) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client: + self.assertEqual( + await server._fetch_sandbox_envs(client), + {"BASE": "sandbox", "JSON": '{"n":1}'}, + ) + + async def test_rejects_non_string_envd_values(self) -> None: + transport = httpx.MockTransport( + lambda request: httpx.Response(200, json={"BASE": 1}) + ) + + async with httpx.AsyncClient(transport=transport) as client: + with self.assertRaisesRegex(HTTPException, "invalid environment payload"): + await server._fetch_sandbox_envs(client) + + async def test_maps_envd_http_failure_to_bad_gateway(self) -> None: + transport = httpx.MockTransport( + lambda request: httpx.Response(503, text="not ready") + ) + + async with httpx.AsyncClient(transport=transport) as client: + with self.assertRaises(HTTPException) as raised: + await server._fetch_sandbox_envs(client) + + self.assertEqual(raised.exception.status_code, 502) + + +class KernelContextSandboxEnvsTest(unittest.IsolatedAsyncioTestCase): + async def test_loads_sandbox_environment_once(self) -> None: + context = server.KernelContext("kernel", "session", "/workspace") + fetch = AsyncMock(return_value={"BASE": "sandbox", "OVERRIDE": "sandbox"}) + + with patch.object(server, "_fetch_sandbox_envs", fetch): + await context.load_sandbox_envs() + await context.load_sandbox_envs() + + fetch.assert_awaited_once() + self.assertEqual( + context.sandbox_envs, + {"BASE": "sandbox", "OVERRIDE": "sandbox"}, + ) + + async def test_caches_empty_envd_environment(self) -> None: + context = server.KernelContext("kernel", "session", "/workspace") + fetch = AsyncMock(return_value={}) + + with patch.object(server, "_fetch_sandbox_envs", fetch): + await context.load_sandbox_envs() + await context.load_sandbox_envs() + + fetch.assert_awaited_once() + self.assertEqual(context.sandbox_envs, {}) + + async def test_retries_after_envd_failure(self) -> None: + context = server.KernelContext("kernel", "session", "/workspace") + fetch = AsyncMock( + side_effect=[ + HTTPException(status_code=502, detail="envd unavailable"), + {"BASE": "sandbox"}, + ] + ) + with patch.object(server, "_fetch_sandbox_envs", fetch): + with self.assertRaises(HTTPException): + await context.load_sandbox_envs() + await context.load_sandbox_envs() + + self.assertEqual(fetch.await_count, 2) + self.assertEqual(context.sandbox_envs, {"BASE": "sandbox"}) + + async def test_applies_sandbox_environment_only_on_first_execution(self) -> None: + context = server.KernelContext("kernel", "session", "/workspace") + context.sandbox_envs = { + "BASE": "sandbox", + "OVERRIDE": "sandbox", + } + + first = context._build_env_setup_code( + { + "OVERRIDE": "request", + "REQUEST_ONLY": "value", + }, + "_snapshot", + ) + context.sandbox_envs_applied = True + second = context._build_env_setup_code(None, None) + + self.assertEqual( + first.splitlines(), + [ + 'import os; os.environ["BASE"] = "sandbox"', + 'import os; os.environ["OVERRIDE"] = "sandbox"', + 'import os; _snapshot = {key: (key in os.environ, os.environ.get(key)) for key in ["OVERRIDE", "REQUEST_ONLY"]}', + 'import os; os.environ["OVERRIDE"] = "request"', + 'import os; os.environ["REQUEST_ONLY"] = "value"', + ], + ) + self.assertEqual(second, "") + + async def test_restores_or_deletes_per_call_environment_after_execution( + self, + ) -> None: + cleanup = server.KernelContext._build_env_cleanup_code("_snapshot") + + self.assertEqual( + cleanup.splitlines(), + [ + "def _snapshot_restore():", + " import os", + ' snapshot = globals().pop("_snapshot", {})', + " for key, (existed, value) in snapshot.items():", + " if existed:", + " os.environ[key] = value", + " else:", + " os.environ.pop(key, None)", + "_snapshot_restore()", + "del _snapshot_restore", + ], + ) + + async def test_executes_cleanup_after_per_call_environment(self) -> None: + context = server.KernelContext("kernel", "session", "/workspace") + context.sandbox_envs = {"OVERRIDE": "sandbox"} + context.ws = AsyncMock() + requests: list[dict] = [] + + async def send(request: str) -> None: + payload = json.loads(request) + requests.append(payload) + msg_id = payload["header"]["msg_id"] + await context.executions[msg_id].put({"type": "end_of_execution"}) + + context.ws.send.side_effect = send + + results = [ + item + async for item in context.execute( + "from __future__ import annotations\nprint('ok')", + env_vars={ + "OVERRIDE": "request", + "REQUEST_ONLY": "value", + }, + ) + ] + self.assertEqual(results, []) + self.assertIsNotNone(context.cleanup_task) + await context.cleanup_task + + self.assertEqual(len(requests), 3) + setup_lines = requests[0]["content"]["code"].splitlines() + snapshot_name = setup_lines[1].split()[2] + self.assertTrue(snapshot_name.startswith("_cube_lci_env_snapshot_")) + self.assertEqual( + setup_lines, + [ + 'import os; os.environ["OVERRIDE"] = "sandbox"', + f'import os; {snapshot_name} = {{key: (key in os.environ, os.environ.get(key)) for key in ["OVERRIDE", "REQUEST_ONLY"]}}', + 'import os; os.environ["OVERRIDE"] = "request"', + 'import os; os.environ["REQUEST_ONLY"] = "value"', + ], + ) + self.assertTrue(requests[0]["content"]["silent"]) + self.assertFalse(requests[0]["content"]["store_history"]) + self.assertEqual( + requests[1]["content"]["code"], + "from __future__ import annotations\nprint('ok')", + ) + self.assertFalse(requests[1]["content"]["silent"]) + self.assertTrue(requests[1]["content"]["store_history"]) + self.assertEqual( + requests[2]["content"]["code"].splitlines(), + [ + f"def {snapshot_name}_restore():", + " import os", + f' snapshot = globals().pop("{snapshot_name}", {{}})', + " for key, (existed, value) in snapshot.items():", + " if existed:", + " os.environ[key] = value", + " else:", + " os.environ.pop(key, None)", + f"{snapshot_name}_restore()", + f"del {snapshot_name}_restore", + ], + ) + self.assertTrue(requests[2]["content"]["silent"]) + self.assertFalse(requests[2]["content"]["store_history"]) + + async def test_rolls_back_per_call_environment_when_setup_fails(self) -> None: + context = server.KernelContext("kernel", "session", "/workspace") + context.sandbox_envs = {"SANDBOX_BASE": "sandbox-value"} + context.ws = AsyncMock() + namespace: dict[str, object] = {} + requests: list[dict] = [] + + async def send(request: str) -> None: + payload = json.loads(request) + requests.append(payload) + msg_id = payload["header"]["msg_id"] + try: + exec(payload["content"]["code"], namespace) + except ValueError as exc: + await context.executions[msg_id].put( + { + "type": "error", + "name": "ValueError", + "value": str(exc), + "traceback": "", + } + ) + await context.executions[msg_id].put({"type": "end_of_execution"}) + + context.ws.send.side_effect = send + + with patch.dict( + os.environ, + {"KERNEL_EXISTING": "kernel-value"}, + clear=False, + ): + with self.assertRaisesRegex(RuntimeError, "background execution failed"): + _ = [ + item + async for item in context.execute( + "pass", + env_vars={ + "KERNEL_EXISTING": "per-call-value", + "PER_CALL_ONLY": "per-call-only", + "INVALID_VALUE": "\x00", + }, + ) + ] + + self.assertEqual(os.environ["KERNEL_EXISTING"], "kernel-value") + self.assertNotIn("PER_CALL_ONLY", os.environ) + self.assertNotIn("INVALID_VALUE", os.environ) + self.assertFalse(context.sandbox_envs_applied) + + retry_results = [ + item + async for item in context.execute( + "pass", + env_vars=None, + ) + ] + self.assertEqual(retry_results, []) + self.assertTrue(context.sandbox_envs_applied) + self.assertEqual(len(requests), 4) + + async def test_waits_for_cleanup_before_the_next_execution(self) -> None: + context = server.KernelContext("kernel", "session", "/workspace") + context.sandbox_envs = {} + context.sandbox_envs_applied = True + context.ws = AsyncMock() + release_cleanup = asyncio.Event() + request_sent = asyncio.Event() + + async def cleanup() -> None: + await release_cleanup.wait() + + async def send(request: str) -> None: + payload = json.loads(request) + msg_id = payload["header"]["msg_id"] + request_sent.set() + await context.executions[msg_id].put({"type": "end_of_execution"}) + + context.cleanup_task = asyncio.create_task(cleanup()) + context.ws.send.side_effect = send + + execution = asyncio.create_task( + anext(context.execute("print('next')", env_vars=None), None) + ) + await asyncio.sleep(0) + self.assertFalse(request_sent.is_set()) + + release_cleanup.set() + await execution + self.assertTrue(request_sent.is_set()) + self.assertIsNone(context.cleanup_task) + + async def test_restores_kernel_environment_after_per_call_override(self) -> None: + context = server.KernelContext("kernel", "session", "/workspace") + context.sandbox_envs = {} + context.ws = AsyncMock() + namespace: dict[str, object] = {} + + async def send(request: str) -> None: + payload = json.loads(request) + msg_id = payload["header"]["msg_id"] + exec(payload["content"]["code"], namespace) + await context.executions[msg_id].put({"type": "end_of_execution"}) + + context.ws.send.side_effect = send + + with patch.dict( + os.environ, + {"KERNEL_EXISTING": "kernel-value"}, + clear=False, + ): + results = [ + item + async for item in context.execute( + "pass", + env_vars={ + "KERNEL_EXISTING": "per-call-value", + "PER_CALL_ONLY": "per-call-only", + }, + ) + ] + self.assertEqual(results, []) + self.assertIsNotNone(context.cleanup_task) + await context.cleanup_task + + self.assertEqual(os.environ["KERNEL_EXISTING"], "kernel-value") + self.assertNotIn("PER_CALL_ONLY", os.environ) + + async def test_applies_sandbox_environment_before_invalid_first_code(self) -> None: + context = server.KernelContext("kernel", "session", "/workspace") + context.sandbox_envs = {"SANDBOX_BASE": "sandbox-value"} + context.ws = AsyncMock() + namespace: dict[str, object] = {} + + async def send(request: str) -> None: + payload = json.loads(request) + msg_id = payload["header"]["msg_id"] + try: + exec(payload["content"]["code"], namespace) + except SyntaxError: + await context.executions[msg_id].put( + { + "type": "error", + "name": "SyntaxError", + "value": "invalid syntax", + "traceback": "", + } + ) + await context.executions[msg_id].put({"type": "end_of_execution"}) + + context.ws.send.side_effect = send + + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("SANDBOX_BASE", None) + results = [ + item + async for item in context.execute( + "if", + env_vars=None, + ) + ] + + self.assertEqual(results[0]["name"], "SyntaxError") + self.assertEqual(os.environ["SANDBOX_BASE"], "sandbox-value") + self.assertTrue(context.sandbox_envs_applied) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/e2e/sdk_compat/adapters/base.py b/tests/e2e/sdk_compat/adapters/base.py index fa8963512..de0673bc5 100644 --- a/tests/e2e/sdk_compat/adapters/base.py +++ b/tests/e2e/sdk_compat/adapters/base.py @@ -47,7 +47,13 @@ def read_file(self, path: str, *, user: str = "root") -> str: raise NotImplementedError @abstractmethod - def run_code(self, code: str, *, timeout: int = 60) -> CodeResult: + def run_code( + self, + code: str, + *, + env_vars: dict[str, str] | None = None, + timeout: int = 60, + ) -> CodeResult: raise NotImplementedError def get_host(self, port: int) -> str: diff --git a/tests/e2e/sdk_compat/adapters/cubesandbox_adapter.py b/tests/e2e/sdk_compat/adapters/cubesandbox_adapter.py index a80da043e..f0e21ee77 100644 --- a/tests/e2e/sdk_compat/adapters/cubesandbox_adapter.py +++ b/tests/e2e/sdk_compat/adapters/cubesandbox_adapter.py @@ -107,8 +107,14 @@ def write_file(self, path: str, content: str, *, user: str = "root") -> None: def read_file(self, path: str, *, user: str = "root") -> str: return self._sandbox.files.read(path, user=user) - def run_code(self, code: str, *, timeout: int = 60) -> CodeResult: - execution = self._sandbox.run_code(code, timeout=timeout) + def run_code( + self, + code: str, + *, + env_vars: dict[str, str] | None = None, + timeout: int = 60, + ) -> CodeResult: + execution = self._sandbox.run_code(code, envs=env_vars, timeout=timeout) stdout = _normalize_log_lines(execution.logs.stdout) if execution.logs else [] stderr = _normalize_log_lines(execution.logs.stderr) if execution.logs else [] return CodeResult( diff --git a/tests/e2e/sdk_compat/adapters/e2b_adapter.py b/tests/e2e/sdk_compat/adapters/e2b_adapter.py index c9825a1c1..a1d7b4019 100644 --- a/tests/e2e/sdk_compat/adapters/e2b_adapter.py +++ b/tests/e2e/sdk_compat/adapters/e2b_adapter.py @@ -286,8 +286,14 @@ def read_file(self, path: str, *, user: str = "root") -> str: raise RuntimeError("E2B files object does not expose read/read_file") return str(reader(path)) - def run_code(self, code: str, *, timeout: int = 60) -> CodeResult: - result = self._sandbox.run_code(code, timeout=timeout) + def run_code( + self, + code: str, + *, + env_vars: dict[str, str] | None = None, + timeout: int = 60, + ) -> CodeResult: + result = self._sandbox.run_code(code, envs=env_vars, timeout=timeout) logs = getattr(result, "logs", None) stdout = list(getattr(logs, "stdout", []) or []) stderr = list(getattr(logs, "stderr", []) or []) diff --git a/tests/e2e/sdk_compat/adapters/tracing_adapter.py b/tests/e2e/sdk_compat/adapters/tracing_adapter.py index b660c9456..9dafb17b2 100644 --- a/tests/e2e/sdk_compat/adapters/tracing_adapter.py +++ b/tests/e2e/sdk_compat/adapters/tracing_adapter.py @@ -80,16 +80,27 @@ def read_file(self, path: str, *, user: str = "root") -> str: output=_content_summary, ) - def run_code(self, code: str, *, timeout: int = 60) -> CodeResult: + def run_code( + self, + code: str, + *, + env_vars: dict[str, str] | None = None, + timeout: int = 60, + ) -> CodeResult: return self._trace.capture( "run_code", { "backend": self.backend, "sandbox_id": self.sandbox_id, "code": code, + "env_vars": env_vars, "timeout": timeout, }, - lambda: self._wrapped.run_code(code, timeout=timeout), + lambda: self._wrapped.run_code( + code, + env_vars=env_vars, + timeout=timeout, + ), ) def get_host(self, port: int) -> str: diff --git a/tests/e2e/sdk_compat/cases/run_code/test_python.py b/tests/e2e/sdk_compat/cases/run_code/test_python.py index 09636a16c..3d200c3af 100644 --- a/tests/e2e/sdk_compat/cases/run_code/test_python.py +++ b/tests/e2e/sdk_compat/cases/run_code/test_python.py @@ -3,6 +3,8 @@ from __future__ import annotations +import json + import pytest from framework.assertions import assert_code_ok @@ -45,6 +47,109 @@ def test_run_code_captures_stderr(sdk_sandbox, sdk_e2e_config): assert any(line.strip() == "hello stderr" for line in result.stderr) +@pytest.mark.sandbox_create_options( + env_vars={ + "SDK_COMPAT_RUN_CODE_BASE": "sandbox-base", + "SDK_COMPAT_RUN_CODE_OVERRIDE": "sandbox-value", + } +) +@pytest.mark.requires_run_code_env_inheritance +def test_run_code_merges_create_and_per_call_envs(sdk_sandbox, sdk_e2e_config): + code = ( + "import json, os\n" + "if os.environ.get('SDK_COMPAT_RUN_CODE_DELETE_DURING_CALL'):\n" + " del os.environ['SDK_COMPAT_RUN_CODE_DELETE_DURING_CALL']\n" + "print(json.dumps({" + "'base': os.environ.get('SDK_COMPAT_RUN_CODE_BASE'), " + "'override': os.environ.get('SDK_COMPAT_RUN_CODE_OVERRIDE'), " + "'per_call_only': os.environ.get('SDK_COMPAT_RUN_CODE_PER_CALL_ONLY'), " + "'kernel_existing': os.environ.get('SDK_COMPAT_RUN_CODE_KERNEL_EXISTING'), " + "'delete_during_call': os.environ.get('SDK_COMPAT_RUN_CODE_DELETE_DURING_CALL'), " + "'leak_candidate': os.environ.get('SDK_COMPAT_RUN_CODE_LEAK_CANDIDATE')" + "}, sort_keys=True))" + ) + + inherited = sdk_sandbox.run_code( + code, + timeout=sdk_e2e_config.run_code_timeout, + ) + assert_code_ok(inherited) + inherited_env = json.loads("".join(inherited.stdout)) + assert inherited_env == { + "base": "sandbox-base", + "delete_during_call": None, + "kernel_existing": None, + "leak_candidate": None, + "override": "sandbox-value", + "per_call_only": None, + }, f"unexpected inherited run_code env: {inherited_env!r}" + + seeded = sdk_sandbox.run_code( + "import os; os.environ['SDK_COMPAT_RUN_CODE_KERNEL_EXISTING'] = 'kernel-value'", + timeout=sdk_e2e_config.run_code_timeout, + ) + assert_code_ok(seeded) + + overridden = sdk_sandbox.run_code( + code, + env_vars={ + "SDK_COMPAT_RUN_CODE_DELETE_DURING_CALL": "delete-me", + "SDK_COMPAT_RUN_CODE_LEAK_CANDIDATE": "must-not-leak", + "SDK_COMPAT_RUN_CODE_OVERRIDE": "per-call-value", + "SDK_COMPAT_RUN_CODE_PER_CALL_ONLY": "per-call-only", + "SDK_COMPAT_RUN_CODE_KERNEL_EXISTING": "per-call-kernel-value", + }, + timeout=sdk_e2e_config.run_code_timeout, + ) + assert_code_ok(overridden) + overridden_env = json.loads("".join(overridden.stdout)) + assert overridden_env == { + "base": "sandbox-base", + "delete_during_call": None, + "kernel_existing": "per-call-kernel-value", + "leak_candidate": "must-not-leak", + "override": "per-call-value", + "per_call_only": "per-call-only", + }, f"unexpected per-call run_code env: {overridden_env!r}" + + restored = sdk_sandbox.run_code( + code, + timeout=sdk_e2e_config.run_code_timeout, + ) + assert_code_ok(restored) + restored_env = json.loads("".join(restored.stdout)) + assert restored_env == { + "base": "sandbox-base", + "delete_during_call": None, + "kernel_existing": "kernel-value", + "leak_candidate": None, + "override": "sandbox-value", + "per_call_only": None, + }, f"unexpected restored run_code env: {restored_env!r}" + + +@pytest.mark.sandbox_create_options( + env_vars={"SDK_COMPAT_RUN_CODE_FIRST_EXECUTION": "sandbox-value"} +) +@pytest.mark.requires_run_code_env_inheritance +def test_run_code_applies_create_env_before_invalid_first_code( + sdk_sandbox, + sdk_e2e_config, +): + invalid = sdk_sandbox.run_code( + "if", + timeout=sdk_e2e_config.run_code_timeout, + ) + assert invalid.error is not None, "expected the first execution to fail syntax" + + inherited = sdk_sandbox.run_code( + "import os; print(os.environ.get('SDK_COMPAT_RUN_CODE_FIRST_EXECUTION'))", + timeout=sdk_e2e_config.run_code_timeout, + ) + assert_code_ok(inherited) + assert "".join(inherited.stdout).strip() == "sandbox-value" + + @pytest.mark.p1 def test_run_code_preserves_kernel_state(sdk_sandbox, sdk_e2e_config): first = sdk_sandbox.run_code( diff --git a/tests/e2e/sdk_compat/conftest.py b/tests/e2e/sdk_compat/conftest.py index 15be04e17..cf8d73f3c 100644 --- a/tests/e2e/sdk_compat/conftest.py +++ b/tests/e2e/sdk_compat/conftest.py @@ -95,6 +95,7 @@ def pytest_configure(config: pytest.Config) -> None: "sandbox_create_options(**kwargs): SDK sandbox create options for this test", "sandbox_template_id(template_id): override template ID for this test or module", "requires_code_interpreter: test requires a stateful Code Interpreter kernel", + "requires_run_code_env_inheritance: test requires an opt-in template whose run_code inherits create-time envs", "requires_internet: test requires public internet access from the sandbox", "requires_cubeproxy: test requires CubeProxy routing to the sandbox", "auth: CUBE_API_KEY simple-key authentication control-plane tests", @@ -272,6 +273,15 @@ def sdk_sandbox( f"backend {sdk_backend!r} does not support stateful Code Interpreter" ) + if ( + request.node.get_closest_marker("requires_run_code_env_inheritance") + and not sdk_e2e_config.run_code_env_inheritance_enabled + ): + pytest.skip( + "run_code env inheritance requires " + "SDK_E2E_RUN_CODE_ENV_INHERITANCE=true and a compatible template" + ) + if request.node.get_closest_marker("requires_internet") and _env_true( "SDK_E2E_SKIP_INTERNET_TESTS" ): @@ -454,6 +464,9 @@ def _log_effective_environment(cfg: SdkE2EConfig) -> None: "CUBE_PROXY_NODE_IP": cfg.cube_proxy_node_ip, "CUBE_PROXY_PORT_HTTP": str(cfg.cube_proxy_port_http), "CUBE_SANDBOX_DOMAIN": cfg.cube_sandbox_domain, + "SDK_E2E_RUN_CODE_ENV_INHERITANCE": str( + cfg.run_code_env_inheritance_enabled + ).lower(), "SDK_E2E_PLATFORM_LIFECYCLE": str(cfg.platform_lifecycle_enabled).lower(), "SDK_E2E_VOLUME_PLUGIN": str(cfg.volume_plugin_enabled).lower(), "SDK_E2E_VOLUME_DRIVER": cfg.volume_driver, diff --git a/tests/e2e/sdk_compat/docs/test-coverage.md b/tests/e2e/sdk_compat/docs/test-coverage.md index 8d230276e..59bb1f3c2 100644 --- a/tests/e2e/sdk_compat/docs/test-coverage.md +++ b/tests/e2e/sdk_compat/docs/test-coverage.md @@ -83,10 +83,10 @@ round-trips, atomic replace and concurrent file access are not covered yet. - expression result text; - stdout and stderr capture; - Python errors and syntax errors; +- create-time env inheritance and temporary per-call env overrides, enabled with `SDK_E2E_RUN_CODE_ENV_INHERITANCE=true` for compatible templates; - stateful kernel variable preservation. -These scenarios require Code Interpreter support and validate normalized -`CodeResult` values rather than SDK-private response objects. +These scenarios require Code Interpreter support and validate normalized `CodeResult` values rather than SDK-private response objects. The environment-inheritance case is skipped unless explicitly enabled because the default template may not provide that behavior. ### 2.5 Network diff --git a/tests/e2e/sdk_compat/docs/zh/test-coverage.md b/tests/e2e/sdk_compat/docs/zh/test-coverage.md index dd4d0651d..82f6fb0d0 100644 --- a/tests/e2e/sdk_compat/docs/zh/test-coverage.md +++ b/tests/e2e/sdk_compat/docs/zh/test-coverage.md @@ -79,10 +79,10 @@ pytest --run-e2e -m "lifecycle and slow" - 表达式结果文本； - stdout 与 stderr 捕获； - Python 错误和语法错误； +- 创建时环境变量继承与临时 per-call env 覆盖，仅在兼容 template 上设置 `SDK_E2E_RUN_CODE_ENV_INHERITANCE=true` 时启用； - stateful kernel 变量保留。 -这些场景要求 Code Interpreter 能力。它们验证的是框架归一化后的 `CodeResult`， -而非单个 SDK 的内部响应格式。 +这些场景要求 Code Interpreter 能力。它们验证的是框架归一化后的 `CodeResult`，而非单个 SDK 的内部响应格式。环境变量继承用例默认跳过，因为默认 template 不一定提供该行为，必须显式开启。 ### 2.5 Network diff --git a/tests/e2e/sdk_compat/env.example b/tests/e2e/sdk_compat/env.example index a219b8289..ef6afef6b 100644 --- a/tests/e2e/sdk_compat/env.example +++ b/tests/e2e/sdk_compat/env.example @@ -54,6 +54,10 @@ SDK_E2E_CREATE_CAPACITY_BACKOFF_MAX=30 SDK_E2E_CREATE_CAPACITY_BUDGET=90 SDK_E2E_COMMAND_TIMEOUT=30 SDK_E2E_RUN_CODE_TIMEOUT=60 +# Opt in only when the selected template makes run_code inherit sandbox +# create-time environments. The default sandbox-code template may not support +# this behavior. +SDK_E2E_RUN_CODE_ENV_INHERITANCE=false SDK_E2E_NETWORK_PROBE_TIMEOUT=5 # Public TCP endpoints used by network policy tests. Configure targets that are # reachable and stable from the test environment. diff --git a/tests/e2e/sdk_compat/framework/config.py b/tests/e2e/sdk_compat/framework/config.py index 3f3a08dc8..844c824a0 100644 --- a/tests/e2e/sdk_compat/framework/config.py +++ b/tests/e2e/sdk_compat/framework/config.py @@ -32,6 +32,7 @@ class SdkE2EConfig: create_timeout: int command_timeout: int run_code_timeout: int + run_code_env_inheritance_enabled: bool network_probe_timeout: int e2b_validate_api_key: bool keep_sandbox_on_failure: bool @@ -76,6 +77,9 @@ def from_env( create_timeout=int(os.environ.get("SDK_E2E_CREATE_TIMEOUT", "120")), command_timeout=int(os.environ.get("SDK_E2E_COMMAND_TIMEOUT", "30")), run_code_timeout=int(os.environ.get("SDK_E2E_RUN_CODE_TIMEOUT", "60")), + run_code_env_inheritance_enabled=_bool_env( + "SDK_E2E_RUN_CODE_ENV_INHERITANCE" + ), network_probe_timeout=int(os.environ.get("SDK_E2E_NETWORK_PROBE_TIMEOUT", "5")), e2b_validate_api_key=_bool_env("SDK_E2E_E2B_VALIDATE_API_KEY"), keep_sandbox_on_failure=_bool_env("SDK_E2E_KEEP_SANDBOX_ON_FAILURE"),