diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0f482d3..d3d3596 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -70,6 +70,10 @@ jobs: - name: Build, verify, and install the wheel and sdist run: mise run build + e2e: + name: e2e + uses: ./.github/workflows/e2e.yml + # The one required status check. Its name matches ai-openshell/sdk-go, so # rulesets can require `ci` in every SDK repository regardless of the jobs # each one runs. diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml new file mode 100644 index 0000000..1955ab2 --- /dev/null +++ b/.github/workflows/e2e.yml @@ -0,0 +1,65 @@ +# SPDX-License-Identifier: Apache-2.0 + +# Runs the E2E suite against the published gateway of a core version, started +# from core's deploy/docker files (scripts/e2e/start-gateway.sh). Sandboxes +# run the image from tests/e2e/Dockerfile.workload, which has the same Python +# as the client, so exec_python can unpickle what the client pickled. +name: e2e + +on: + workflow_call: + inputs: + core-version: + description: Core version (default the pin, OPENSHELL_CORE_REF in mise.toml) + type: string + default: "" + workflow_dispatch: + inputs: + core-version: + description: Core version, for example v0.1.7 (default the pin, OPENSHELL_CORE_REF in mise.toml) + type: string + default: "" + +permissions: + contents: read + +jobs: + e2e: + name: e2e + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + persist-credentials: false + + - uses: jdx/mise-action@e6a8b3978addb5a52f2b4cd9d91eafa7f0ab959d # v4.2.0 + + - name: Build the sandbox image + run: mise run e2e:image + + - name: Start the gateway + env: + CORE_VERSION: ${{ inputs.core-version }} + run: | + set -euo pipefail + version=${CORE_VERSION:-$(mise run -q core:version)} + echo "Core version: $version" >>"$GITHUB_STEP_SUMMARY" + scripts/e2e/start-gateway.sh "$version" --default-image openshell/e2e-python:dev + + - name: Run the E2E suite + env: + OPENSHELL_GATEWAY: e2e + run: mise run e2e + + - name: Stop the gateway + if: always() + run: scripts/e2e/stop-gateway.sh + + - name: Upload the gateway log + if: failure() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: gateway-log + path: ${{ runner.temp }}/openshell-e2e/gateway.log + if-no-files-found: ignore diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index be20d04..1f99646 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -20,6 +20,7 @@ virtual environment. `mise tasks` lists every task. The ones you need most: | `mise run format` | Format the code | | `mise run test` | Unit tests; `UV_PYTHON=3.11` picks the interpreter | | `mise run build` | Build the wheel and sdist, verify them, and install them on Python 3.11 | +| `mise run e2e` | E2E tests against a running gateway; see the README for setup | | `mise run proto:gen` | Regenerate the bindings from core at the pinned ref | | `mise run core:version` | Print the pinned core version | diff --git a/README.md b/README.md index 3920b91..93ff9f9 100644 --- a/README.md +++ b/README.md @@ -40,7 +40,29 @@ reads them, and `grpcio-tools` generates the bindings: - A change of the pin shows the effect on the generated API in its diff. - `mise run proto:check` fails when the committed bindings differ from the pinned protos, and `mise run conformance:check` does the same for the - vendored test fixtures. + vendored test fixtures and `tests/e2e/Dockerfile.workload`. + +## E2E tests + +`tests/e2e` runs against a real gateway; `mise run test` skips it. CI runs it +against the published gateway of the pinned core version, with the Docker +compute driver. To run it locally, you need Docker with the compose plugin: + +```shell +mise run e2e:image +scripts/e2e/start-gateway.sh "$(mise run -q core:version)" --default-image openshell/e2e-python:dev +OPENSHELL_GATEWAY=e2e mise run e2e +scripts/e2e/stop-gateway.sh +``` + +`e2e:image` builds the sandbox image from `tests/e2e/Dockerfile.workload`, a +copy of core's, with the Python version from `.python-version` and +`cloudpickle`, so `Sandbox.exec_python` can run the client's callables. + +`start-gateway.sh` is meant for throwaway hosts. It registers `e2e` as the +active gateway, and `stop-gateway.sh` removes every sandbox container labelled +with core's default namespace. Set `XDG_CONFIG_HOME` to keep your own gateway +registrations apart. ## History diff --git a/mise.toml b/mise.toml index 2597641..0a99ef1 100644 --- a/mise.toml +++ b/mise.toml @@ -65,16 +65,17 @@ fi """ [tasks."conformance:check"] -description = "Fail if vendored fixtures differ from core at the pinned ref" +description = "Fail if vendored fixtures or the E2E workload Dockerfile differ from core at the pinned ref" run = """ #!/usr/bin/env bash set -euo pipefail core=$(mktemp -d) trap 'rm -rf "$core"' EXIT -scripts/fetch-core.sh "$OPENSHELL_CORE_REF" "$core" sdk/conformance +scripts/fetch-core.sh "$OPENSHELL_CORE_REF" "$core" sdk/conformance e2e/python/Dockerfile.workload for fixture in oauth-client-credentials.json oauth-client-credentials.json.license; do diff -u "$core/sdk/conformance/$fixture" "tests/fixtures/$fixture" done +diff -u "$core/e2e/python/Dockerfile.workload" tests/e2e/Dockerfile.workload """ [tasks.lint] @@ -100,7 +101,18 @@ run = "uv run --frozen ty check {{vars.python_paths}}" [tasks.test] description = "Run unit tests (set UV_PYTHON to pick the interpreter)" depends = ["sync", "proto:fetch"] -run = "uv run --frozen pytest -n auto" +run = "uv run --frozen pytest -n auto --ignore=tests/e2e" + +[tasks."e2e:image"] +description = "Build the E2E sandbox image: Python from .python-version plus cloudpickle, for exec_python" +run = """ +docker build --file tests/e2e/Dockerfile.workload --build-arg "PYTHON_VERSION=$(cat .python-version)" --tag openshell/e2e-python:dev tests/e2e +""" + +[tasks.e2e] +description = "Run the E2E suite against the gateway named in OPENSHELL_GATEWAY (see scripts/e2e/start-gateway.sh)" +depends = ["sync"] +run = "uv run --frozen pytest tests/e2e -n ${E2E_PARALLEL:-5}" [tasks.check] description = "Run formatting, lint, type, and drift checks" diff --git a/scripts/e2e/start-gateway.sh b/scripts/e2e/start-gateway.sh new file mode 100755 index 0000000..0118e34 --- /dev/null +++ b/scripts/e2e/start-gateway.sh @@ -0,0 +1,132 @@ +#!/usr/bin/env bash +# SPDX-License-Identifier: Apache-2.0 + +# start-gateway.sh: run the published OpenShell gateway for a core version +# with the Docker compute driver, for the E2E lane. +# +# Usage: start-gateway.sh [--state-dir ] [--default-image ] +# +# Uses core's own deployment files, deploy/docker/docker-compose.yml and +# gateway.toml, at the core tag: an init container generates the supervisor +# JWT keys, and the gateway serves plaintext gRPC on 127.0.0.1:8080 and +# /readyz on 127.0.0.1:8081. The supervisor and sandbox images are pinned to +# the same version instead of :latest. --default-image replaces the image that +# sandboxes run when they don't name one (core's default is plain Ubuntu); the +# gateway pulls it only when it isn't present locally. Then registers the +# gateway as "e2e" for the SDK in $XDG_CONFIG_HOME/openshell. Waits up to 15 +# minutes for the gateway image of a fresh tag to be published. +# +# Needs docker with the compose plugin and a host path /var/lib/openshell +# (core's compose file bind-mounts it at the same path). Meant for throwaway +# CI hosts: it registers the gateway as the active one, and stop-gateway.sh +# removes every container labelled with core's default sandbox namespace, +# including sandboxes of another local gateway that uses the same namespace. +set -euo pipefail + +core_repo=${CORE_REPO:-NVIDIA/OpenShell} +registry=ghcr.io/nvidia/openshell +name=e2e +port=8080 +health_port=8081 + +tag= +default_image= +state=${OPENSHELL_E2E_STATE:-${RUNNER_TEMP:-/tmp}/openshell-e2e} +while [ $# -gt 0 ]; do + case $1 in + --state-dir) state=${2:?--state-dir needs a path}; shift 2 ;; + --default-image) default_image=${2:?--default-image needs an image}; shift 2 ;; + -*) echo "usage: $0 [--state-dir ] [--default-image ]" >&2; exit 2 ;; + *) tag=$1; shift ;; + esac +done +[[ "$tag" =~ ^v[0-9]+\.[0-9]+\.[0-9]+(-pre\.[0-9]+)?$ ]] || { + echo "usage: $0 (vX.Y.Z or vX.Y.Z-pre.N)" >&2 + exit 2 +} +semver=${tag#v} +mkdir -p "$state" +echo "$tag" >"$state/core-version" + +# Core pushes images after tagging; a proposal can be faster than that. +manifest_exists() { + local token + token=$(curl -fsS "https://ghcr.io/token?scope=repository:nvidia/openshell/gateway:pull" | jq -r .token) || return 1 + curl -fsS -o /dev/null -H "Authorization: Bearer $token" \ + -H 'Accept: application/vnd.oci.image.index.v1+json,application/vnd.docker.distribution.manifest.list.v2+json,application/vnd.oci.image.manifest.v1+json' \ + "https://ghcr.io/v2/nvidia/openshell/gateway/manifests/$semver" +} +for i in $(seq 1 15); do + manifest_exists && break + if [ "$i" -eq 15 ]; then + echo "ERROR: gateway image for $semver not published yet, re-run later" >&2 + exit 1 + fi + echo "Waiting for $registry/gateway:$semver ($i/15) ..." + sleep 60 +done + +# Core's deployment files at the tag. +base="https://raw.githubusercontent.com/$core_repo/$tag/deploy/docker" +curl -fsSL "$base/docker-compose.yml" -o "$state/docker-compose.yml" +curl -fsSL "$base/gateway.toml" -o "$state/gateway.toml" +sed -i.bak \ + -e "s#\"$registry/supervisor:latest\"#\"$registry/supervisor:$semver\"#" \ + -e "s#\"$registry/sandbox:latest\"#\"$registry/sandbox:$semver\"#" \ + "$state/gateway.toml" && rm -f "$state/gateway.toml.bak" +for img in supervisor sandbox; do + grep -q "\"$registry/$img:$semver\"" "$state/gateway.toml" || { + echo "ERROR: could not pin the $img image in core's gateway.toml at $tag" >&2 + exit 1 + } +done +# Core's file binds the health server to loopback inside the container, where +# the published port can't reach it. Bind it to all container interfaces; the +# compose file publishes the port on the host's 127.0.0.1 only. +sed -i.bak 's#^health_bind_address *= *"127\.0\.0\.1:8081"#health_bind_address = "0.0.0.0:8081"#' \ + "$state/gateway.toml" && rm -f "$state/gateway.toml.bak" +grep -q '^health_bind_address = "0.0.0.0:8081"' "$state/gateway.toml" || { + echo "ERROR: could not rebind the health server in core's gateway.toml at $tag" >&2 + exit 1 +} + +if [ -n "$default_image" ]; then + sed -i.bak "s#^default_image *= *\".*\"#default_image = \"$default_image\"#" \ + "$state/gateway.toml" && rm -f "$state/gateway.toml.bak" + grep -qF "default_image = \"$default_image\"" "$state/gateway.toml" || { + echo "ERROR: could not set default_image in core's gateway.toml at $tag" >&2 + exit 1 + } +fi + +compose() { IMAGE_TAG=$semver docker compose -p openshell-e2e -f "$state/docker-compose.yml" "$@"; } +echo "Starting gateway $semver ..." +compose up -d --quiet-pull + +for i in $(seq 1 90); do + if curl -fsS -o /dev/null "http://127.0.0.1:$health_port/readyz"; then + break + fi + if [ "$i" -eq 90 ]; then + echo "ERROR: gateway not ready after 180 s" >&2 + compose logs --no-color gateway | tail -n 80 >&2 || true + exit 1 + fi + sleep 2 +done +echo "Gateway $semver ready on 127.0.0.1:$port." + +config_home=${XDG_CONFIG_HOME:-$HOME/.config} +dir="$config_home/openshell/gateways/$name" +mkdir -p "$dir" +cat >"$dir/metadata.json" <"$config_home/openshell/active_gateway" +echo "Registered gateway '$name' in $config_home/openshell; run the suite with OPENSHELL_GATEWAY=$name." diff --git a/scripts/e2e/stop-gateway.sh b/scripts/e2e/stop-gateway.sh new file mode 100755 index 0000000..e30e577 --- /dev/null +++ b/scripts/e2e/stop-gateway.sh @@ -0,0 +1,53 @@ +#!/usr/bin/env bash +# SPDX-License-Identifier: Apache-2.0 + +# stop-gateway.sh: stop the E2E gateway started by start-gateway.sh, save its +# log, remove the sandbox containers, and remove the "e2e" gateway +# registration (and the active-gateway pointer when it still names "e2e"). +# +# Usage: stop-gateway.sh [--state-dir ] +# +# Writes /gateway.log. Safe to run when nothing is running. Runs +# every cleanup step even when one fails, then exits nonzero if any failed. +set -uo pipefail + +state=${OPENSHELL_E2E_STATE:-${RUNNER_TEMP:-/tmp}/openshell-e2e} +[ "${1:-}" != --state-dir ] || state=${2:?--state-dir needs a path} +[ -f "$state/docker-compose.yml" ] || { echo "No E2E gateway state in $state."; exit 0; } + +semver=$(sed 's/^v//' "$state/core-version" 2>/dev/null || echo latest) +compose() { IMAGE_TAG=$semver docker compose -p openshell-e2e -f "$state/docker-compose.yml" "$@"; } + +failed=0 +fail() { echo "stop-gateway: $*" >&2; failed=1; } + +if compose logs --no-color --timestamps >"$state/gateway.log" 2>&1; then + echo "Gateway log: $state/gateway.log" +else + fail "could not save the gateway log; see $state/gateway.log" +fi + +# The Docker driver labels sandbox and supervisor containers with core's +# openshell.ai/managed-by and openshell.ai/sandbox-namespace labels. Compose +# doesn't know about them, so remove them before taking the project down. +if ids=$(docker ps -aq \ + --filter label=openshell.ai/managed-by=openshell \ + --filter label=openshell.ai/sandbox-namespace=openshell); then + # shellcheck disable=SC2086 # one container ID per word + [ -z "$ids" ] || docker rm -f $ids >/dev/null || fail "could not remove sandbox containers" +else + fail "could not list sandbox containers" +fi +compose down --volumes --remove-orphans >/dev/null || fail "docker compose down failed" + +config_home=${XDG_CONFIG_HOME:-$HOME/.config} +rm -rf "$config_home/openshell/gateways/e2e" || fail "could not remove the e2e gateway registration" +if [ "$(cat "$config_home/openshell/active_gateway" 2>/dev/null)" = e2e ]; then + rm -f "$config_home/openshell/active_gateway" || fail "could not remove the active-gateway pointer" +fi + +if [ "$failed" -ne 0 ]; then + echo "Gateway cleanup failed; the gateway or its sandboxes may still be running." >&2 + exit 1 +fi +echo "Gateway stopped." diff --git a/tests/e2e/Dockerfile.workload b/tests/e2e/Dockerfile.workload new file mode 100644 index 0000000..f25c05d --- /dev/null +++ b/tests/e2e/Dockerfile.workload @@ -0,0 +1,29 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +ARG WORKLOAD_BASE=nvcr.io/nvidia/base/ubuntu:24.04 +FROM ghcr.io/astral-sh/uv:0.12.17@sha256:10787c682e4184e4f290de1171fd4703dc63de99221f10fe1c99002ce7fa9acc AS uv +FROM ${WORKLOAD_BASE} + +COPY --from=uv /uv /uvx /usr/local/bin/ +# Supplied from .python-version by the build task for cloudpickle compatibility. +ARG PYTHON_VERSION +# curl and openssl also support the Rust Docker tests sharing this fixture. +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + bash ca-certificates curl git openssl python3 python3-pip python3-venv python-is-python3 \ + && rm -rf /var/lib/apt/lists/* \ + && if getent passwd ubuntu >/dev/null; then userdel ubuntu; fi \ + && if getent group ubuntu >/dev/null; then groupdel ubuntu; fi \ + && groupadd --gid 1000 sandbox \ + && useradd --uid 1000 --gid sandbox --create-home --shell /bin/bash sandbox \ + && UV_PYTHON_INSTALL_DIR=/sandbox/.uv/python uv python install "${PYTHON_VERSION}" \ + && UV_PYTHON_INSTALL_DIR=/sandbox/.uv/python uv venv --python "${PYTHON_VERSION}" --seed /sandbox/.venv \ + && uv pip install --python /sandbox/.venv/bin/python cloudpickle==3.1.2 google-auth==2.40.3 requests==2.32.5 \ + && chown -R sandbox:sandbox /sandbox \ + && uv cache clean + +ENV VIRTUAL_ENV=/sandbox/.venv +ENV PATH="/sandbox/.venv/bin:${PATH}" +WORKDIR /sandbox +USER sandbox:sandbox diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py new file mode 100644 index 0000000..cc7a70a --- /dev/null +++ b/tests/e2e/conftest.py @@ -0,0 +1,86 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Fixtures for the E2E suite. + +The suite runs against the gateway named in ``OPENSHELL_GATEWAY``, for example +the one ``scripts/e2e/start-gateway.sh`` registers as ``e2e``. Without it, +every test is skipped. See ``mise run e2e``. +""" + +from __future__ import annotations + +import os +import time +from typing import TYPE_CHECKING + +import grpc +import pytest + +from openshell import Sandbox, SandboxClient, WorkspaceClient + +if TYPE_CHECKING: + from collections.abc import Callable, Iterator + + from openshell._proto import openshell_pb2 + + +@pytest.fixture(scope="session") +def cluster_name() -> str: + name = os.environ.get("OPENSHELL_GATEWAY") + if not name: + pytest.skip("OPENSHELL_GATEWAY not set") + return name + + +@pytest.fixture(scope="session") +def sandbox_client(cluster_name: str) -> Iterator[SandboxClient]: + with SandboxClient.from_active_cluster(cluster=cluster_name) as client: + yield client + + +@pytest.fixture(scope="session", autouse=True) +def ensure_sandbox_persistence_ready(sandbox_client: SandboxClient) -> None: + for _ in range(60): + try: + sandbox_client.list_ids(workspace="default", page_size=1) + return + except grpc.RpcError as exc: + details = exc.details() or "" + if exc.code() == grpc.StatusCode.UNAVAILABLE: + time.sleep(2) + continue + if ( + exc.code() == grpc.StatusCode.INTERNAL + and "no such table: objects" in details + ): + time.sleep(1) + continue + raise + + pytest.fail( + "openshell-server persistence is not initialized (missing sqlite objects table); " + "restart the gateway and rerun the E2E suite" + ) + + +@pytest.fixture +def sandbox(cluster_name: str) -> Callable[..., Sandbox]: + def _create( + *, spec: openshell_pb2.SandboxSpec | None = None, delete_on_exit: bool = True + ) -> Sandbox: + return Sandbox( + workspace="default", + cluster=cluster_name, + spec=spec, + delete_on_exit=delete_on_exit, + # Allow time to pull an explicitly supplied workload fixture. + ready_timeout_seconds=300.0, + ) + + return _create + + +@pytest.fixture(scope="session") +def workspace_client(sandbox_client: SandboxClient) -> WorkspaceClient: + return WorkspaceClient.from_sandbox_client(sandbox_client) diff --git a/tests/e2e/exec_admission_test.py b/tests/e2e/exec_admission_test.py new file mode 100644 index 0000000..dd5c8a9 --- /dev/null +++ b/tests/e2e/exec_admission_test.py @@ -0,0 +1,157 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import threading +import time +import uuid +from typing import TYPE_CHECKING + +import grpc +import pytest +from google.protobuf import duration_pb2 + +from openshell._proto import datamodel_pb2, openshell_pb2 +from openshell.errors import from_grpc_error + +if TYPE_CHECKING: + from collections.abc import Callable + + from openshell import Sandbox, SandboxClient + + +def assert_blocked(stream, reason: str) -> None: + with pytest.raises(grpc.RpcError) as caught: + list(stream) + error = from_grpc_error(caught.value) + assert error.code() == grpc.StatusCode.FAILED_PRECONDITION + assert error.error_info is not None + assert error.error_info.reason == reason + + +@pytest.mark.parametrize("interactive", [False, True]) +def test_exec_request_id_never_relaunches_or_replays_output( + sandbox: Callable[..., Sandbox], sandbox_client: SandboxClient, interactive: bool +) -> None: + with sandbox(delete_on_exit=True) as sb: + path = f"/sandbox/exec-admission-{uuid.uuid4().hex}" + request = openshell_pb2.ExecSandboxRequest( + sandbox=sb.sandbox.name, + workspace_scope=datamodel_pb2.WorkspaceSelector(workspace="default"), + command=[ + "/bin/sh", + "-c", + f"printf x >> {path}; printf started; " + f"while [ ! -e {path}.release ]; do sleep 0.1; done; " + "printf finished; exit 7", + ], + request_id=str(uuid.uuid4()), + ) + done = threading.Event() + + def invoke(value): + if not interactive: + return sandbox_client._stub.ExecSandbox(value, timeout=30) + + def inputs(): + yield openshell_pb2.ExecSandboxInput(start=value) + done.wait(timeout=30) + + return sandbox_client._stub.ExecSandboxInteractive(inputs(), timeout=30) + + stream = invoke(request) + try: + first = next(stream) + assert first.WhichOneof("payload") == "stdout" + assert b"started" in first.stdout.data + assert_blocked(invoke(request), "REQUEST_OUTCOME_UNCERTAIN") + changed = openshell_pb2.ExecSandboxRequest() + changed.CopyFrom(request) + changed.command.append("changed-payload") + assert_blocked(invoke(changed), "REQUEST_ID_PAYLOAD_MISMATCH") + assert sb.exec(["touch", f"{path}.release"]).exit_code == 0 + events = list(stream) + assert events[-1].WhichOneof("payload") == "exit" + assert events[-1].exit.exit_code == 7 + assert_blocked(invoke(request), "REQUEST_STREAM_UNAVAILABLE") + result = sb.exec(["cat", path]) + assert result.exit_code == 0 + assert result.stdout == "x" + finally: + done.set() + stream.cancel() + + +@pytest.mark.parametrize("interactive", [False, True]) +def test_exec_request_timeout_keeps_launch_unresolved( + sandbox: Callable[..., Sandbox], sandbox_client: SandboxClient, interactive: bool +) -> None: + with sandbox(delete_on_exit=True) as sb: + path = f"/sandbox/exec-timeout-{uuid.uuid4().hex}" + request = openshell_pb2.ExecSandboxRequest( + sandbox=sb.sandbox.name, + workspace_scope=datamodel_pb2.WorkspaceSelector(workspace="default"), + command=["/bin/sh", "-c", f"printf x >> {path}; sleep 5"], + execution_timeout=duration_pb2.Duration(seconds=1), + request_id=str(uuid.uuid4()), + ) + done = threading.Event() + + def invoke(): + if not interactive: + return sandbox_client._stub.ExecSandbox(request, timeout=30) + + def inputs(): + yield openshell_pb2.ExecSandboxInput(start=request) + done.wait(timeout=30) + + return sandbox_client._stub.ExecSandboxInteractive(inputs(), timeout=30) + + stream = invoke() + try: + events = [] + for event in stream: + events.append(event) + if event.WhichOneof("payload") == "exit": + # A synthetic timeout leaves the interactive input task + # alive until the client closes its side of the stream. + done.set() + assert events[-1].exit.exit_code == 124 + assert_blocked(invoke(), "REQUEST_OUTCOME_UNCERTAIN") + assert sb.exec(["cat", path]).stdout == "x" + finally: + done.set() + stream.cancel() + + +def test_exec_client_cancellation_does_not_clear_launch_admission( + sandbox: Callable[..., Sandbox], sandbox_client: SandboxClient +) -> None: + with sandbox(delete_on_exit=True) as sb: + path = f"/sandbox/exec-cancel-{uuid.uuid4().hex}" + request = openshell_pb2.ExecSandboxRequest( + sandbox=sb.sandbox.name, + workspace_scope=datamodel_pb2.WorkspaceSelector(workspace="default"), + command=[ + "/bin/sh", + "-c", + f"printf x >> {path}; printf started; sleep 1; printf finished", + ], + request_id=str(uuid.uuid4()), + ) + stream = sandbox_client._stub.ExecSandbox(request, timeout=30) + assert b"started" in next(stream).stdout.data + stream.cancel() + deadline = time.monotonic() + 15 + while True: + with pytest.raises(grpc.RpcError) as caught: + list(sandbox_client._stub.ExecSandbox(request, timeout=30)) + error = from_grpc_error(caught.value) + assert error.error_info is not None + if error.error_info.reason == "REQUEST_STREAM_UNAVAILABLE": + break + assert error.error_info.reason == "REQUEST_OUTCOME_UNCERTAIN" + assert time.monotonic() < deadline + time.sleep(0.1) + assert sb.exec(["cat", path]).stdout == "x" diff --git a/tests/e2e/sandbox_api_test.py b/tests/e2e/sandbox_api_test.py new file mode 100644 index 0000000..cb7eccf --- /dev/null +++ b/tests/e2e/sandbox_api_test.py @@ -0,0 +1,387 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import contextlib +import threading +import uuid +from typing import TYPE_CHECKING + +from google.protobuf import duration_pb2 + +from openshell._proto import datamodel_pb2, openshell_pb2, sandbox_pb2 + +if TYPE_CHECKING: + from collections.abc import Callable + + from openshell import Sandbox, SandboxClient, WorkspaceClient + + +def test_mutation_replay_preserves_sandbox_lifecycle_and_replacement( + sandbox_client: SandboxClient, +) -> None: + name = f"replay-{uuid.uuid4().hex[:8]}" + scope = "default" + stub = sandbox_client._stub + create = openshell_pb2.CreateSandboxRequest( + name=name, + spec=openshell_pb2.SandboxSpec(), + workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=scope), + request_id=str(uuid.uuid4()), + ) + + def replay(method, request): + result, call = method.with_call(request, timeout=60) + assert dict(call.initial_metadata())["openshell-replayed"] == "true" + return result + + try: + original = stub.CreateSandbox(create, timeout=60).sandbox.metadata.id + sandbox_client.wait_ready(name, workspace="default", timeout_seconds=300) + assert replay(stub.CreateSandbox, create).sandbox.metadata.id == original + stop = openshell_pb2.StopSandboxRequest( + name=name, + workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=scope), + request_id=str(uuid.uuid4()), + ) + stub.StopSandbox(stop, timeout=60) + sandbox_client.wait_stopped(name, workspace="default", timeout_seconds=120) + assert replay(stub.StopSandbox, stop).sandbox.metadata.id == original + start = openshell_pb2.StartSandboxRequest( + name=name, + workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=scope), + request_id=str(uuid.uuid4()), + ) + stub.StartSandbox(start, timeout=60) + sandbox_client.wait_ready(name, workspace="default", timeout_seconds=300) + assert replay(stub.StartSandbox, start).sandbox.metadata.id == original + update = openshell_pb2.UpdateConfigRequest( + sandbox=name, + workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=scope), + setting_key="ocsf_json_enabled", + setting_value=sandbox_pb2.SettingValue(bool_value=True), + request_id=str(uuid.uuid4()), + ) + updated = stub.UpdateConfig(update, timeout=30) + assert replay(stub.UpdateConfig, update) == updated + delete = openshell_pb2.DeleteSandboxRequest( + name=name, + workspace_scope=datamodel_pb2.WorkspaceSelector(workspace=scope), + request_id=str(uuid.uuid4()), + ) + deleted = stub.DeleteSandbox(delete, timeout=60) + assert deleted.sandbox_id == original + sandbox_client.wait_deleted(name, workspace="default", timeout_seconds=120) + replacement = sandbox_client.create(workspace="default", name=name) + assert replacement.id != original + assert replay(stub.DeleteSandbox, delete) == deleted + assert sandbox_client.get(name, workspace="default").id == replacement.id + finally: + with contextlib.suppress(Exception): + sandbox_client.delete(name, workspace="default", allow_missing=True) + sandbox_client.wait_deleted(name, workspace="default", timeout_seconds=120) + + +def test_sandbox_api_crud_and_exec( + sandbox: Callable[..., Sandbox], + sandbox_client: SandboxClient, +) -> None: + class _FileOps: + def write(self, path: str, content: str) -> None: + from pathlib import Path + + Path(path).write_text(content) + + def read(self, path: str) -> str: + from pathlib import Path + + return Path(path).read_text() + + with sandbox(delete_on_exit=True) as sb: + assert sb.id + # Server auto-generates a petname (e.g. "feasible-retriever") + assert sb.sandbox.name + parts = sb.sandbox.name.split("-") + assert len(parts) == 2, ( + f"expected petname with 2 parts, got {sb.sandbox.name!r}" + ) + assert all(p.isalpha() and p.islower() for p in parts) + + fetched = sandbox_client.get(sb.sandbox.name, workspace="default") + assert fetched.id == sb.id + + ids = set(sandbox_client.list_ids(workspace="default", page_size=100)) + assert sb.id in ids + + result = sb.exec(["python", "-c", "print('sandbox-ok')"]) + assert result.exit_code == 0 + assert "sandbox-ok" in result.stdout + + file_ops = _FileOps() + create_file = sb.exec_python( + file_ops.write, + args=("/sandbox/exec-persistence.txt", "ok"), + ) + assert create_file.exit_code == 0 + + verify_file = sb.exec_python( + file_ops.read, args=("/sandbox/exec-persistence.txt",) + ) + assert verify_file.exit_code == 0 + assert verify_file.stdout.strip() == "ok" + + +def test_sandbox_interactive_exec_honors_tty( + sandbox: Callable[..., Sandbox], + sandbox_client: SandboxClient, +) -> None: + ready_marker = b"tty-ready\n" + stdin_sentinel = b"streamed-stdin-sentinel" + stdout_sentinel = b"stdout-sentinel" + stderr_sentinel = b"stderr-sentinel" + + def exec_interactive(sandbox_name: str, *, tty: bool) -> tuple[bytes, bytes]: + request = openshell_pb2.ExecSandboxInput( + start=openshell_pb2.ExecSandboxRequest( + sandbox=sandbox_name, + workspace_scope=datamodel_pb2.WorkspaceSelector(workspace="default"), + command=[ + "/bin/sh", + "-c", + "[ -t 0 ] && printf T || printf N; " + "[ -t 1 ] && printf T || printf N; " + "[ -t 2 ] && printf T || printf N; " + f"printf '\\n%s' '{ready_marker.decode()}'; " + "IFS= read -r stdin_value; " + "printf 'stdin:%s\\n' \"$stdin_value\"; " + "printf 'stdout-sentinel\\n'; " + "printf 'stderr-sentinel\\n' >&2", + ], + tty=tty, + execution_timeout=duration_pb2.Duration(seconds=20), + ) + ) + + ready = threading.Event() + done = threading.Event() + + def requests(): + yield request + # PTY input echo can split the separate TTY flag writes. Wait for + # the complete marker, including its newline, before sending input. + if not ready.wait(timeout=20) or done.is_set(): + return + yield openshell_pb2.ExecSandboxInput(stdin=stdin_sentinel + b"\n") + done.wait(timeout=30) + + stdout = bytearray() + stderr = bytearray() + exit_code: int | None = None + try: + events = sandbox_client._stub.ExecSandboxInteractive(requests(), timeout=30) + for event in events: + payload = event.WhichOneof("payload") + if payload == "stdout": + stdout.extend(event.stdout.data) + if ready_marker in stdout.replace(b"\r\n", b"\n"): + ready.set() + elif payload == "stderr": + stderr.extend(event.stderr.data) + elif payload == "exit": + exit_code = int(event.exit.exit_code) + finally: + done.set() + # Unblock a request iterator waiting for readiness on early exit + # or RPC failure without sending input after the call has ended. + ready.set() + + assert exit_code == 0 + return bytes(stdout), bytes(stderr) + + with sandbox(delete_on_exit=True) as sb: + stdout, stderr = exec_interactive(sb.sandbox.name, tty=False) + assert b"NNN" in stdout.splitlines() + assert b"stdin:" + stdin_sentinel in stdout + assert stdout_sentinel in stdout + assert stdout_sentinel not in stderr + assert stderr_sentinel in stderr + assert stderr_sentinel not in stdout + + stdout, stderr = exec_interactive(sb.sandbox.name, tty=True) + terminal_output = stdout + stderr + assert b"TTT" in terminal_output.splitlines() + assert b"stdin:" + stdin_sentinel in terminal_output + assert stdout_sentinel in terminal_output + assert stderr_sentinel in terminal_output + + +def test_interactive_exec_drains_output_after_request_eof( + sandbox: Callable[..., Sandbox], + sandbox_client: SandboxClient, +) -> None: + with sandbox(delete_on_exit=True) as sb: + + def requests(): + yield openshell_pb2.ExecSandboxInput( + start=openshell_pb2.ExecSandboxRequest( + sandbox=sb.sandbox.name, + workspace_scope=datamodel_pb2.WorkspaceSelector( + workspace="default" + ), + command=[ + "/bin/sh", + "-c", + "input=$(cat); sleep 0.2; " + "printf '%s' \"$input\"; printf 'drained-stderr' >&2; exit 7", + ], + tty=False, + execution_timeout=duration_pb2.Duration(seconds=20), + ) + ) + yield openshell_pb2.ExecSandboxInput(stdin=b"drained-stdout") + # End requests before the command emits output. The receive + # direction must survive long enough to drain both output streams. + + stdout = bytearray() + stderr = bytearray() + exit_codes: list[int] = [] + for event in sandbox_client._stub.ExecSandboxInteractive( + requests(), timeout=30 + ): + assert not exit_codes, "received another event after terminal exit" + payload = event.WhichOneof("payload") + if payload == "stdout": + stdout.extend(event.stdout.data) + elif payload == "stderr": + stderr.extend(event.stderr.data) + elif payload == "exit": + exit_codes.append(event.exit.exit_code) + assert stdout == b"drained-stdout" + assert stderr == b"drained-stderr" + assert exit_codes == [7] + + +def test_list_scoped_and_for_all_workspaces( + sandbox_client: SandboxClient, + workspace_client: WorkspaceClient, +) -> None: + import contextlib + import uuid + + suffix = uuid.uuid4().hex[:8] + other_ws = f"list-ws-{suffix}" + created_default: list[str] = [] + created_other: list[str] = [] + + try: + workspace_client.create(other_ws) + + ref_default = sandbox_client.create( + workspace="default", name=f"ls-def-{suffix}" + ) + created_default.append(ref_default.name) + + ref_other = sandbox_client.create(workspace=other_ws, name=f"ls-oth-{suffix}") + created_other.append(ref_other.name) + + default_ids = set(sandbox_client.list_ids(workspace="default")) + assert ref_default.id in default_ids + assert ref_other.id not in default_ids + + other_ids = set(sandbox_client.list_ids(workspace=other_ws)) + assert ref_other.id in other_ids + assert ref_default.id not in other_ids + + all_ids = set(sandbox_client.list_ids_for_all_workspaces()) + assert ref_default.id in all_ids + assert ref_other.id in all_ids + finally: + for name in created_default: + with contextlib.suppress(Exception): + sandbox_client.delete(name, workspace="default") + sandbox_client.wait_deleted(name, workspace="default") + for name in created_other: + with contextlib.suppress(Exception): + sandbox_client.delete(name, workspace=other_ws) + sandbox_client.wait_deleted(name, workspace=other_ws) + with contextlib.suppress(Exception): + workspace_client.delete(other_ws) + + +def test_sandbox_labels_and_selectors(sandbox_client: SandboxClient) -> None: + import contextlib + import uuid + + suffix = uuid.uuid4().hex[:8] + job_a = f"lbl-a-{suffix}" + job_b = f"lbl-b-{suffix}" + group_selector = f"aiq-test={suffix}" + primary_selector = f"aiq-test={suffix},role=primary" + + created: list[str] = [] + try: + ref_a = sandbox_client.create( + workspace="default", + name=job_a, + labels={"aiq-test": suffix, "role": "primary"}, + ) + created.append(ref_a.name) + ref_b = sandbox_client.create( + workspace="default", + name=job_b, + labels={"aiq-test": suffix, "role": "secondary"}, + ) + created.append(ref_b.name) + + # Labels round-trip through create and get. + assert ref_a.labels["role"] == "primary" + assert ( + dict(sandbox_client.get(job_a, workspace="default").labels)["role"] + == "primary" + ) + assert ( + dict(sandbox_client.get(job_b, workspace="default").labels)["role"] + == "secondary" + ) + + # A specific selector filters to exactly the primary sandbox. + assert { + s.name + for s in sandbox_client.list_all( + workspace="default", label_selector=primary_selector + ) + } == {job_a} + # The shared group label returns both. + assert { + s.name + for s in sandbox_client.list_all( + workspace="default", label_selector=group_selector + ) + } == { + job_a, + job_b, + } + + # Deleting one removes only it from selector results. + assert sandbox_client.delete(job_a, workspace="default") + sandbox_client.wait_deleted(job_a, workspace="default") + created.remove(job_a) + assert { + s.name + for s in sandbox_client.list_all( + workspace="default", label_selector=group_selector + ) + } == {job_b} + + # Final deletion leaves no matching sandboxes. + assert sandbox_client.delete(job_b, workspace="default") + sandbox_client.wait_deleted(job_b, workspace="default") + created.remove(job_b) + assert not sandbox_client.list_all( + workspace="default", label_selector=group_selector + ) + finally: + for name in created: + with contextlib.suppress(Exception): + sandbox_client.delete(name, workspace="default") diff --git a/tests/e2e/sandbox_exec_python_test.py b/tests/e2e/sandbox_exec_python_test.py new file mode 100644 index 0000000..4c61d7d --- /dev/null +++ b/tests/e2e/sandbox_exec_python_test.py @@ -0,0 +1,42 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Callable + + from openshell import Sandbox + + +def test_exec_python_accepts_bound_methods( + sandbox: Callable[..., Sandbox], +) -> None: + class _Calculator: + def multiply(self, left: int, right: int) -> int: + return left * right + + with sandbox(delete_on_exit=True) as sb: + calculator = _Calculator() + result = sb.exec_python(calculator.multiply, args=(6, 7), timeout_seconds=20) + + assert result.exit_code == 0 + assert result.stdout.strip() == "42" + + +def test_exec_python_surfaces_callable_errors( + sandbox: Callable[..., Sandbox], +) -> None: + class _Boom: + def fail(self) -> None: + raise RuntimeError("expected-failure") + + with sandbox(delete_on_exit=True) as sb: + result = sb.exec_python(_Boom().fail, timeout_seconds=20) + + assert result.exit_code != 0 + output = f"{result.stdout}\n{result.stderr}" + assert "expected-failure" in output + assert "RuntimeError" in output diff --git a/tests/e2e/workspace_api_test.py b/tests/e2e/workspace_api_test.py new file mode 100644 index 0000000..a60597e --- /dev/null +++ b/tests/e2e/workspace_api_test.py @@ -0,0 +1,173 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import contextlib +import uuid +from typing import TYPE_CHECKING + +import grpc +import pytest + +from openshell._proto import openshell_pb2 + +if TYPE_CHECKING: + from openshell import WorkspaceClient + + +def test_workspace_crud(workspace_client: WorkspaceClient) -> None: + name = f"ws-crud-{uuid.uuid4().hex[:8]}" + + try: + ws = workspace_client.create(name) + assert ws.name == name + assert ws.phase == "WORKSPACE_PHASE_ACTIVE" + + fetched = workspace_client.get(name) + assert fetched.name == name + assert fetched.phase == "WORKSPACE_PHASE_ACTIVE" + finally: + with contextlib.suppress(Exception): + workspace_client.delete(name) + + +def test_workspace_create_with_labels(workspace_client: WorkspaceClient) -> None: + name = f"ws-lbl-{uuid.uuid4().hex[:8]}" + + try: + ws = workspace_client.create(name, labels={"env": "test", "team": "infra"}) + assert ws.labels["env"] == "test" + assert ws.labels["team"] == "infra" + + fetched = workspace_client.get(name) + assert fetched.labels["env"] == "test" + assert fetched.labels["team"] == "infra" + finally: + with contextlib.suppress(Exception): + workspace_client.delete(name) + + +def test_workspace_list_includes_created(workspace_client: WorkspaceClient) -> None: + name = f"ws-list-{uuid.uuid4().hex[:8]}" + + try: + workspace_client.create(name) + + names = {ws.name for ws in workspace_client.list_all()} + assert name in names + assert "default" in names + finally: + with contextlib.suppress(Exception): + workspace_client.delete(name) + + +def test_workspace_delete_nonexistent_raises_not_found( + workspace_client: WorkspaceClient, +) -> None: + with pytest.raises(grpc.RpcError) as exc_info: + workspace_client.delete(f"no-such-ws-{uuid.uuid4().hex[:8]}") + assert exc_info.value.code() == grpc.StatusCode.NOT_FOUND + + +def test_workspace_get_nonexistent_raises_not_found( + workspace_client: WorkspaceClient, +) -> None: + with pytest.raises(grpc.RpcError) as exc_info: + workspace_client.get(f"no-such-ws-{uuid.uuid4().hex[:8]}") + assert exc_info.value.code() == grpc.StatusCode.NOT_FOUND + + +def test_workspace_request_ids_are_scoped_to_each_target( + workspace_client: WorkspaceClient, +) -> None: + names = [f"ws-scope-{uuid.uuid4().hex[:8]}" for _ in range(2)] + create_id = str(uuid.uuid4()) + delete_id = str(uuid.uuid4()) + stub = workspace_client._stub + originals = [] + try: + for name in names: + request = openshell_pb2.CreateWorkspaceRequest( + name=name, request_id=create_id + ) + response, call = stub.CreateWorkspace.with_call(request, timeout=20) + assert "openshell-replayed" not in dict(call.initial_metadata()) + assert response.workspace.metadata.name == name + fetched = stub.GetWorkspace( + openshell_pb2.GetWorkspaceRequest(name=name), timeout=20 + ) + assert fetched.workspace.metadata.id == response.workspace.metadata.id + originals.append(response) + assert originals[0].workspace.metadata.id != originals[1].workspace.metadata.id + + for name, original in zip(names, originals, strict=True): + replay, call = stub.CreateWorkspace.with_call( + openshell_pb2.CreateWorkspaceRequest(name=name, request_id=create_id), + timeout=20, + ) + assert replay == original + assert dict(call.initial_metadata())["openshell-replayed"] == "true" + + for name in names: + request = openshell_pb2.DeleteWorkspaceRequest( + name=name, request_id=delete_id + ) + response, call = stub.DeleteWorkspace.with_call(request, timeout=20) + assert "openshell-replayed" not in dict(call.initial_metadata()) + assert response.outcome == openshell_pb2.DELETION_OUTCOME_COMPLETED + with pytest.raises(grpc.RpcError) as exc_info: + workspace_client.get(name) + assert exc_info.value.code() == grpc.StatusCode.NOT_FOUND + + for name in names: + replay, call = stub.DeleteWorkspace.with_call( + openshell_pb2.DeleteWorkspaceRequest(name=name, request_id=delete_id), + timeout=20, + ) + assert replay.outcome == openshell_pb2.DELETION_OUTCOME_COMPLETED + assert dict(call.initial_metadata())["openshell-replayed"] == "true" + finally: + for name in names: + with contextlib.suppress(Exception): + workspace_client.delete(name, allow_missing=True) + + +def test_workspace_request_id_replays_without_deleting_replacement( + workspace_client: WorkspaceClient, +) -> None: + name = f"ws-replay-{uuid.uuid4().hex[:8]}" + # Exercise the generated wire fields before curated request-ID helpers land. + stub = workspace_client._stub + create = openshell_pb2.CreateWorkspaceRequest( + name=name, request_id=str(uuid.uuid4()) + ) + delete = openshell_pb2.DeleteWorkspaceRequest( + name=name, request_id=str(uuid.uuid4()) + ) + try: + original = stub.CreateWorkspace(create, timeout=20) + replay, call = stub.CreateWorkspace.with_call(create, timeout=20) + assert original == replay + assert dict(call.initial_metadata())["openshell-replayed"] == "true" + + mismatch = openshell_pb2.CreateWorkspaceRequest( + name=name, request_id=create.request_id, labels={"changed": "true"} + ) + with pytest.raises(grpc.RpcError) as exc_info: + stub.CreateWorkspace(mismatch, timeout=20) + assert exc_info.value.code() == grpc.StatusCode.FAILED_PRECONDITION + + removed = stub.DeleteWorkspace(delete, timeout=20) + replacement = stub.CreateWorkspace( + openshell_pb2.CreateWorkspaceRequest(name=name), timeout=20 + ) + assert replacement.workspace.metadata.id != original.workspace.metadata.id + assert stub.DeleteWorkspace(delete, timeout=20) == removed + fetched = stub.GetWorkspace( + openshell_pb2.GetWorkspaceRequest(name=name), timeout=20 + ) + assert fetched.workspace.metadata.id == replacement.workspace.metadata.id + finally: + with contextlib.suppress(Exception): + workspace_client.delete(name)