mirror of
https://github.com/NVIDIA/OpenShell.git
synced 2026-10-18 05:39:50 +08:00
68 lines
1.8 KiB
Python
68 lines
1.8 KiB
Python
# 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 Any, cast
|
|
|
|
from navigator._proto import navigator_pb2
|
|
from navigator.sandbox import (
|
|
_PYTHON_CLOUDPICKLE_BOOTSTRAP,
|
|
_SANDBOX_PYTHON_BIN,
|
|
SandboxClient,
|
|
)
|
|
|
|
|
|
class _FakeStub:
|
|
def __init__(self) -> None:
|
|
self.request: navigator_pb2.ExecSandboxRequest | None = None
|
|
|
|
def ExecSandbox(
|
|
self,
|
|
request: navigator_pb2.ExecSandboxRequest,
|
|
timeout: float | None = None,
|
|
):
|
|
self.request = request
|
|
_ = timeout
|
|
yield navigator_pb2.ExecSandboxEvent(
|
|
exit=navigator_pb2.ExecSandboxExit(exit_code=0)
|
|
)
|
|
|
|
|
|
def _client_with_fake_stub(stub: _FakeStub) -> SandboxClient:
|
|
client = cast("SandboxClient", object.__new__(SandboxClient))
|
|
client._timeout = 30.0
|
|
client._stub = cast("Any", stub)
|
|
return client
|
|
|
|
|
|
def test_exec_sends_stdin_payload() -> None:
|
|
stub = _FakeStub()
|
|
client = _client_with_fake_stub(stub)
|
|
|
|
result = client.exec("sandbox-1", ["python", "-c", "print('ok')"], stdin=b"payload")
|
|
|
|
assert result.exit_code == 0
|
|
assert stub.request is not None
|
|
assert stub.request.stdin == b"payload"
|
|
|
|
|
|
def test_exec_python_serializes_callable_payload() -> None:
|
|
stub = _FakeStub()
|
|
client = _client_with_fake_stub(stub)
|
|
|
|
def add(a: int, b: int) -> int:
|
|
return a + b
|
|
|
|
result = client.exec_python("sandbox-1", add, args=(2, 3))
|
|
|
|
assert result.exit_code == 0
|
|
assert stub.request is not None
|
|
assert stub.request.command == [
|
|
_SANDBOX_PYTHON_BIN,
|
|
"-c",
|
|
_PYTHON_CLOUDPICKLE_BOOTSTRAP,
|
|
]
|
|
assert stub.request.environment["NEMOCLAW_PYFUNC_B64"]
|
|
assert stub.request.stdin == b""
|