wehub-resource-sync
e0e362d700
SDK Tests / changes (push) Successful in 2m29s
Real E2E Tests / changes (push) Successful in 2m29s
Deploy Docs Pages / build (push) Has been cancelled
Deploy Docs Pages / deploy (push) Has been cancelled
Real E2E Tests / JavaScript E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Python E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Java E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / C# E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Go E2E (docker bridge) (push) Has been cancelled
Real E2E Tests / Real E2E CI (push) Has been cancelled
SDK Tests / SDK CI (push) Has been cancelled
SDK Tests / CLI Tests (push) Has been cancelled
SDK Tests / Python SDK Quality (code-interpreter) (push) Has been cancelled
SDK Tests / Python SDK Quality (sandbox) (push) Has been cancelled
SDK Tests / Python SDK Tests (code-interpreter) (push) Has been cancelled
SDK Tests / JavaScript SDK Quality And Tests (code-interpreter) (push) Has been cancelled
SDK Tests / JavaScript SDK Quality And Tests (sandbox) (push) Has been cancelled
SDK Tests / Python SDK Tests (sandbox) (push) Has been cancelled
SDK Tests / CLI Quality (push) Has been cancelled
SDK Tests / Kotlin SDK Quality And Tests (sandbox) (push) Has been cancelled
SDK Tests / Kotlin SDK Quality And Tests (code-interpreter) (push) Has been cancelled
SDK Tests / C# SDK Quality And Tests (code-interpreter) (push) Has been cancelled
SDK Tests / C# SDK Quality And Tests (sandbox) (push) Has been cancelled
SDK Tests / Go SDK Quality And Tests (push) Has been cancelled
3044 lines
116 KiB
Python
3044 lines
116 KiB
Python
# Copyright 2025 Alibaba Group Holding Ltd.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
import pytest
|
|
from types import SimpleNamespace
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import MagicMock
|
|
from fastapi import HTTPException
|
|
from kubernetes.client import ApiException
|
|
|
|
from opensandbox_server.api.schema import (
|
|
ImageSpec,
|
|
ImageAuth,
|
|
NetworkPolicy,
|
|
NetworkRule,
|
|
PlatformSpec,
|
|
)
|
|
from opensandbox_server.config import (
|
|
AppConfig,
|
|
EGRESS_MODE_DNS,
|
|
EGRESS_MODE_DNS_NFT,
|
|
EgressConfig,
|
|
ExecdInitResources,
|
|
KubernetesRuntimeConfig,
|
|
RuntimeConfig,
|
|
)
|
|
from opensandbox_server.services.constants import (
|
|
OPEN_SANDBOX_EGRESS_AUTH_HEADER,
|
|
OPENSANDBOX_EGRESS_MITMPROXY_TRANSPARENT,
|
|
OPENSANDBOX_RUNTIME_MOUNT_PATH,
|
|
OPENSANDBOX_RUNTIME_VOLUME_NAME,
|
|
SANDBOX_EGRESS_AUTH_TOKEN_METADATA_KEY,
|
|
)
|
|
from opensandbox_server.services.k8s.batchsandbox_provider import BatchSandboxProvider
|
|
from opensandbox_server.services.constants import OPENSANDBOX_EGRESS_TOKEN
|
|
from opensandbox_server.services.k8s.image_pull_secret_helper import IMAGE_AUTH_SECRET_PREFIX
|
|
from opensandbox_server.services.k8s.volume_helper import apply_volumes_to_pod_spec
|
|
|
|
|
|
def _app_config_with_template(template_file_path: str) -> AppConfig:
|
|
"""Build an AppConfig with a batchsandbox_template_file set."""
|
|
return AppConfig(
|
|
runtime=RuntimeConfig(type="kubernetes", execd_image="execd:test"),
|
|
kubernetes=KubernetesRuntimeConfig(
|
|
namespace="test-ns",
|
|
batchsandbox_template_file=template_file_path,
|
|
),
|
|
)
|
|
|
|
|
|
def _app_config_with_execd_resources(execd_init_resources: ExecdInitResources) -> AppConfig:
|
|
"""Build an AppConfig with execd_init_resources set."""
|
|
return AppConfig(
|
|
runtime=RuntimeConfig(type="kubernetes", execd_image="execd:test"),
|
|
kubernetes=KubernetesRuntimeConfig(
|
|
namespace="test-ns",
|
|
execd_init_resources=execd_init_resources,
|
|
),
|
|
)
|
|
|
|
|
|
def _app_config_with_image_pull_policy(image_pull_policy: str) -> AppConfig:
|
|
"""Build an AppConfig with image_pull_policy set."""
|
|
return AppConfig(
|
|
runtime=RuntimeConfig(type="kubernetes", execd_image="execd:test"),
|
|
kubernetes=KubernetesRuntimeConfig(
|
|
namespace="test-ns",
|
|
image_pull_policy=image_pull_policy,
|
|
),
|
|
)
|
|
|
|
|
|
def _app_config_with_egress_disable_ipv6(disable_ipv6: bool = True) -> AppConfig:
|
|
"""Build an AppConfig with ``egress.disable_ipv6`` set (privileged execd init when egress is used)."""
|
|
return AppConfig(
|
|
runtime=RuntimeConfig(type="kubernetes", execd_image="execd:test"),
|
|
kubernetes=KubernetesRuntimeConfig(namespace="test-ns"),
|
|
egress=EgressConfig(disable_ipv6=disable_ipv6),
|
|
)
|
|
|
|
|
|
class TestBatchSandboxProvider:
|
|
# ===== Initialization Tests =====
|
|
|
|
def test_init_without_template_creates_provider(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
|
|
assert provider.k8s_client == mock_k8s_client
|
|
assert provider.template_manager._template is None
|
|
assert provider.group == "sandbox.opensandbox.io"
|
|
assert provider.version == "v1alpha1"
|
|
assert provider.plural == "batchsandboxes"
|
|
|
|
def test_init_with_template_loads_template(self, mock_k8s_client, tmp_path):
|
|
template_file = tmp_path / "template.yaml"
|
|
template_file.write_text("spec:\n replicas: 1")
|
|
|
|
provider = BatchSandboxProvider(
|
|
mock_k8s_client, _app_config_with_template(str(template_file))
|
|
)
|
|
|
|
assert provider.template_manager._template is not None
|
|
|
|
def test_init_sets_crd_constants_correctly(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
|
|
assert provider.group == "sandbox.opensandbox.io"
|
|
assert provider.version == "v1alpha1"
|
|
assert provider.plural == "batchsandboxes"
|
|
|
|
# ===== Workload Creation Tests =====
|
|
|
|
def test_create_workload_builds_correct_manifest(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
expires_at = datetime(2025, 12, 31, 10, 0, 0, tzinfo=timezone.utc)
|
|
|
|
result = provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={"FOO": "bar"},
|
|
resource_limits={"cpu": "1", "memory": "1Gi"},
|
|
labels={"opensandbox.io/id": "test-id"},
|
|
expires_at=expires_at,
|
|
execd_image="execd:latest",
|
|
)
|
|
|
|
assert result == {"name": "test-id", "uid": "test-uid", "apiVersion": "sandbox.opensandbox.io/v1alpha1", "kind": "BatchSandbox"}
|
|
|
|
# Verify API call
|
|
call_args = mock_k8s_client.create_custom_object.call_args
|
|
body = call_args.kwargs["body"]
|
|
|
|
assert body["apiVersion"] == "sandbox.opensandbox.io/v1alpha1"
|
|
assert body["kind"] == "BatchSandbox"
|
|
assert body["metadata"]["name"] == "test-id"
|
|
assert body["metadata"]["namespace"] == "test-ns"
|
|
assert body["spec"]["replicas"] == 1
|
|
assert body["spec"]["expireTime"] == "2025-12-31T10:00:00+00:00"
|
|
assert "template" in body["spec"]
|
|
assert body["spec"]["template"]["spec"]["automountServiceAccountToken"] is False
|
|
assert "initContainers" in body["spec"]["template"]["spec"]
|
|
assert "containers" in body["spec"]["template"]["spec"]
|
|
assert "volumes" in body["spec"]["template"]["spec"]
|
|
|
|
def test_create_workload_injects_platform_node_selector(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={"cpu": "1", "memory": "1Gi"},
|
|
labels={"opensandbox.io/id": "test-id"},
|
|
expires_at=None,
|
|
execd_image="execd:latest",
|
|
platform=PlatformSpec(os="linux", arch="arm64"),
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
node_selector = body["spec"]["template"]["spec"]["nodeSelector"]
|
|
assert node_selector["kubernetes.io/os"] == "linux"
|
|
assert node_selector["kubernetes.io/arch"] == "arm64"
|
|
|
|
def test_create_workload_windows_profile_uses_windows_runtime_shape(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="dockurr/windows:latest"),
|
|
entrypoint=["cmd", "/c", "echo hello"],
|
|
env={"VERSION": "11"},
|
|
resource_limits={"cpu": "4", "memory": "8G", "disk": "64G"},
|
|
labels={"opensandbox.io/id": "test-id"},
|
|
expires_at=None,
|
|
execd_image="execd:latest",
|
|
platform=PlatformSpec(os="windows", arch="amd64"),
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
|
|
# windows profile should enforce requested arch, but not force os=windows.
|
|
node_selector = pod_spec.get("nodeSelector", {})
|
|
assert node_selector["kubernetes.io/arch"] == "amd64"
|
|
assert "kubernetes.io/os" not in node_selector
|
|
|
|
init_container = pod_spec["initContainers"][0]
|
|
assert init_container["command"] == ["/bin/sh", "-c"]
|
|
assert "install.bat" in init_container["args"][0]
|
|
assert "execd.exe" in init_container["args"][0]
|
|
|
|
main_container = pod_spec["containers"][0]
|
|
assert main_container["command"] == ["cmd", "/c", "echo hello"]
|
|
# Resources include QEMU memory overhead (8G + 2Gi overhead = 10Gi)
|
|
assert main_container["resources"]["limits"]["cpu"] == "4"
|
|
assert main_container["resources"]["limits"]["memory"] == "10Gi"
|
|
|
|
env_dict = {item["name"]: item["value"] for item in main_container.get("env", [])}
|
|
assert env_dict["VERSION"] == "11"
|
|
assert env_dict["CPU_CORES"] == "4"
|
|
assert env_dict["RAM_SIZE"] == "8G"
|
|
assert env_dict["DISK_SIZE"] == "64G"
|
|
assert env_dict["USER_PORTS"] == "44772,8080,3389,8006"
|
|
|
|
volume_names = {volume["name"] for volume in pod_spec.get("volumes", [])}
|
|
assert "opensandbox-win-oem" in volume_names
|
|
assert "opensandbox-win-kvm" in volume_names
|
|
assert "opensandbox-win-tun" in volume_names
|
|
|
|
def test_create_workload_windows_profile_default_entrypoint_uses_image_entrypoint(
|
|
self, mock_k8s_client
|
|
):
|
|
"""When entrypoint is the SDK default, command is removed so image ENTRYPOINT runs."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="dockurr/windows:latest"),
|
|
entrypoint=["tail", "-f", "/dev/null"],
|
|
env={"VERSION": "11"},
|
|
resource_limits={"cpu": "4", "memory": "8G", "disk": "64G"},
|
|
labels={"opensandbox.io/id": "test-id"},
|
|
expires_at=None,
|
|
execd_image="execd:latest",
|
|
platform=PlatformSpec(os="windows", arch="amd64"),
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
main_container = pod_spec["containers"][0]
|
|
# No command set - image default ENTRYPOINT will be used
|
|
assert "command" not in main_container
|
|
assert "args" not in main_container
|
|
|
|
def test_create_workload_windows_profile_merges_user_ports(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="dockurr/windows:latest"),
|
|
entrypoint=["cmd", "/c", "echo hello"],
|
|
env={"VERSION": "11", "USER_PORTS": "3000,44772"},
|
|
resource_limits={"cpu": "4", "memory": "8G", "disk": "64G"},
|
|
labels={"opensandbox.io/id": "test-id"},
|
|
expires_at=None,
|
|
execd_image="execd:latest",
|
|
platform=PlatformSpec(os="windows", arch="amd64"),
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
main_container = pod_spec["containers"][0]
|
|
env_dict = {item["name"]: item["value"] for item in main_container.get("env", [])}
|
|
assert env_dict["USER_PORTS"] == "3000,44772,8080,3389,8006"
|
|
|
|
def test_create_workload_windows_profile_rejects_arch_conflict_with_template_selector(
|
|
self, mock_k8s_client, tmp_path
|
|
):
|
|
template_file = tmp_path / "template.yaml"
|
|
template_file.write_text(
|
|
"""
|
|
spec:
|
|
template:
|
|
spec:
|
|
nodeSelector:
|
|
kubernetes.io/arch: arm64
|
|
"""
|
|
)
|
|
provider = BatchSandboxProvider(
|
|
mock_k8s_client, _app_config_with_template(str(template_file))
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="platform conflict with template nodeSelector"):
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="dockurr/windows:latest"),
|
|
entrypoint=["cmd", "/c", "echo hello"],
|
|
env={"VERSION": "11"},
|
|
resource_limits={"cpu": "4", "memory": "8G", "disk": "64G"},
|
|
labels={"opensandbox.io/id": "test-id"},
|
|
expires_at=None,
|
|
execd_image="execd:latest",
|
|
platform=PlatformSpec(os="windows", arch="amd64"),
|
|
)
|
|
|
|
def test_create_workload_rejects_platform_conflict_with_template_selector(
|
|
self, mock_k8s_client, tmp_path
|
|
):
|
|
template_file = tmp_path / "template.yaml"
|
|
template_file.write_text(
|
|
"""
|
|
spec:
|
|
template:
|
|
spec:
|
|
nodeSelector:
|
|
kubernetes.io/os: linux
|
|
kubernetes.io/arch: amd64
|
|
"""
|
|
)
|
|
provider = BatchSandboxProvider(
|
|
mock_k8s_client, _app_config_with_template(str(template_file))
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="platform conflict with template nodeSelector"):
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={"cpu": "1", "memory": "1Gi"},
|
|
labels={"opensandbox.io/id": "test-id"},
|
|
expires_at=None,
|
|
execd_image="execd:latest",
|
|
platform=PlatformSpec(os="linux", arch="arm64"),
|
|
)
|
|
|
|
def test_create_workload_rejects_platform_conflict_with_template_node_affinity(
|
|
self, mock_k8s_client, tmp_path
|
|
):
|
|
template_file = tmp_path / "template.yaml"
|
|
template_file.write_text(
|
|
"""
|
|
spec:
|
|
template:
|
|
spec:
|
|
affinity:
|
|
nodeAffinity:
|
|
requiredDuringSchedulingIgnoredDuringExecution:
|
|
nodeSelectorTerms:
|
|
- matchExpressions:
|
|
- key: kubernetes.io/arch
|
|
operator: In
|
|
values: ["amd64"]
|
|
"""
|
|
)
|
|
provider = BatchSandboxProvider(
|
|
mock_k8s_client, _app_config_with_template(str(template_file))
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="platform conflict with template nodeAffinity"):
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={"cpu": "1", "memory": "1Gi"},
|
|
labels={"opensandbox.io/id": "test-id"},
|
|
expires_at=None,
|
|
execd_image="execd:latest",
|
|
platform=PlatformSpec(os="linux", arch="arm64"),
|
|
)
|
|
|
|
def test_create_workload_builds_execd_init_container(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test", "uid": "uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:test",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
init_container = body["spec"]["template"]["spec"]["initContainers"][0]
|
|
|
|
assert init_container["name"] == "execd-installer"
|
|
assert init_container["image"] == "execd:test"
|
|
assert init_container["command"] == ["/bin/sh", "-c"]
|
|
assert "bootstrap.sh" in init_container["args"][0]
|
|
assert init_container["volumeMounts"][0]["name"] == "opensandbox-bin"
|
|
# No resources configured: resources field should be absent
|
|
assert "resources" not in init_container
|
|
|
|
def test_create_workload_init_container_with_configured_resources(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(
|
|
mock_k8s_client,
|
|
_app_config_with_execd_resources(
|
|
ExecdInitResources(
|
|
limits={"cpu": "100m", "memory": "128Mi"},
|
|
requests={"cpu": "50m", "memory": "64Mi"},
|
|
)
|
|
),
|
|
)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test", "uid": "uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:test",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
init_container = body["spec"]["template"]["spec"]["initContainers"][0]
|
|
assert init_container["resources"]["limits"] == {"cpu": "100m", "memory": "128Mi"}
|
|
assert init_container["resources"]["requests"] == {"cpu": "50m", "memory": "64Mi"}
|
|
|
|
def test_create_workload_sets_configured_image_pull_policy(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(
|
|
mock_k8s_client,
|
|
_app_config_with_image_pull_policy("Always"),
|
|
)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test", "uid": "uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:test",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
main_container = body["spec"]["template"]["spec"]["containers"][0]
|
|
assert main_container["imagePullPolicy"] == "Always"
|
|
|
|
def test_create_workload_wraps_entrypoint_with_bootstrap(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "sandbox-test", "uid": "uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/usr/bin/python", "app.py"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
main_container = body["spec"]["template"]["spec"]["containers"][0]
|
|
|
|
assert main_container["command"] == [
|
|
"/opt/opensandbox/bootstrap.sh",
|
|
"/usr/bin/python",
|
|
"app.py",
|
|
]
|
|
|
|
def test_create_workload_converts_env_to_list(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "sandbox-test", "uid": "uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={"FOO": "bar", "BAZ": "qux"},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
env_vars = body["spec"]["template"]["spec"]["containers"][0]["env"]
|
|
|
|
# Should have user env vars plus EXECD
|
|
assert len(env_vars) == 3
|
|
env_dict = {e["name"]: e["value"] for e in env_vars}
|
|
assert env_dict["FOO"] == "bar"
|
|
assert env_dict["BAZ"] == "qux"
|
|
# Verify EXECD is automatically injected
|
|
assert env_dict["EXECD"] == "/opt/opensandbox/execd"
|
|
|
|
def test_create_workload_merges_template_volumes_and_mounts(self, mock_k8s_client, tmp_path):
|
|
template_file = tmp_path / "template.yaml"
|
|
template_file.write_text(
|
|
"""
|
|
spec:
|
|
template:
|
|
spec:
|
|
volumes:
|
|
- name: sandbox-shared-data
|
|
emptyDir: {}
|
|
containers:
|
|
- name: sandbox
|
|
image: ubuntu:latest
|
|
volumeMounts:
|
|
- name: sandbox-shared-data
|
|
mountPath: /data
|
|
"""
|
|
)
|
|
provider = BatchSandboxProvider(
|
|
mock_k8s_client, _app_config_with_template(str(template_file))
|
|
)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "sandbox-test", "uid": "uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
spec = body["spec"]["template"]["spec"]
|
|
|
|
volume_names = [v["name"] for v in spec["volumes"]]
|
|
assert "sandbox-shared-data" in volume_names
|
|
assert "opensandbox-bin" in volume_names
|
|
|
|
# Runtime container should stay intact (template image should not override)
|
|
container = spec["containers"][0]
|
|
assert container["name"] == "sandbox"
|
|
assert container["image"] == "python:3.11"
|
|
|
|
mount_names = [m["name"] for m in container["volumeMounts"]]
|
|
assert "sandbox-shared-data" in mount_names
|
|
assert "opensandbox-bin" in mount_names
|
|
|
|
def test_create_workload_dedupes_template_volume_and_mount_names(
|
|
self, mock_k8s_client, tmp_path
|
|
):
|
|
template_file = tmp_path / "template.yaml"
|
|
template_file.write_text(
|
|
"""
|
|
spec:
|
|
template:
|
|
spec:
|
|
volumes:
|
|
- name: opensandbox-bin
|
|
emptyDir: {}
|
|
- name: sandbox-shared-data
|
|
emptyDir: {}
|
|
containers:
|
|
- name: sandbox
|
|
volumeMounts:
|
|
- name: opensandbox-bin
|
|
mountPath: /opt/opensandbox
|
|
- name: sandbox-shared-data
|
|
mountPath: /data
|
|
"""
|
|
)
|
|
provider = BatchSandboxProvider(
|
|
mock_k8s_client, _app_config_with_template(str(template_file))
|
|
)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "sandbox-test", "uid": "uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
spec = body["spec"]["template"]["spec"]
|
|
|
|
volume_names = [v["name"] for v in spec["volumes"]]
|
|
assert volume_names.count("opensandbox-bin") == 1
|
|
assert "sandbox-shared-data" in volume_names
|
|
|
|
mount_names = [m["name"] for m in spec["containers"][0]["volumeMounts"]]
|
|
assert mount_names.count("opensandbox-bin") == 1
|
|
assert "sandbox-shared-data" in mount_names
|
|
|
|
def test_create_workload_sets_resource_limits_and_requests(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "sandbox-test", "uid": "uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={"cpu": "1", "memory": "1Gi"},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
resources = body["spec"]["template"]["spec"]["containers"][0]["resources"]
|
|
|
|
assert resources["limits"] == {"cpu": "1", "memory": "1Gi"}
|
|
assert resources["requests"] == {"cpu": "1", "memory": "1Gi"}
|
|
|
|
def test_create_workload_uses_separate_resource_requests(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "sandbox-test", "uid": "uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={"cpu": "2", "memory": "2Gi"},
|
|
resource_requests={"cpu": "500m", "memory": "512Mi"},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
resources = body["spec"]["template"]["spec"]["containers"][0]["resources"]
|
|
|
|
assert resources["limits"] == {"cpu": "2", "memory": "2Gi"}
|
|
assert resources["requests"] == {"cpu": "500m", "memory": "512Mi"}
|
|
|
|
def test_create_workload_handles_empty_resource_limits(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "sandbox-test", "uid": "uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
container = body["spec"]["template"]["spec"]["containers"][0]
|
|
|
|
assert "resources" not in container
|
|
|
|
def test_create_workload_translates_gpu_to_nvidia_extended_resource(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "sandbox-test", "uid": "uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={"cpu": "1", "memory": "1Gi", "gpu": "2"},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
resources = body["spec"]["template"]["spec"]["containers"][0]["resources"]
|
|
|
|
assert resources["limits"]["nvidia.com/gpu"] == "2"
|
|
assert resources["requests"]["nvidia.com/gpu"] == "2"
|
|
# Raw key must not leak through as an unknown extended resource.
|
|
assert "gpu" not in resources["limits"]
|
|
assert "gpu" not in resources["requests"]
|
|
|
|
def test_create_workload_without_gpu_omits_nvidia_extended_resource(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "sandbox-test", "uid": "uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={"cpu": "1", "memory": "1Gi"},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
resources = body["spec"]["template"]["spec"]["containers"][0]["resources"]
|
|
|
|
assert "nvidia.com/gpu" not in resources["limits"]
|
|
assert "nvidia.com/gpu" not in resources["requests"]
|
|
|
|
def test_create_workload_rejects_gpu_all_sentinel(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
|
|
with pytest.raises(HTTPException) as excinfo:
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={"cpu": "1", "gpu": "all"},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
)
|
|
assert excinfo.value.status_code == 400
|
|
|
|
# ===== Workload Query Tests =====
|
|
|
|
def test_get_workload_finds_existing_sandbox(
|
|
self, mock_k8s_client, mock_batchsandbox_list_response
|
|
):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = mock_batchsandbox_list_response["items"][0]
|
|
|
|
result = provider.get_workload("test-id", "test-ns")
|
|
|
|
assert result is not None
|
|
assert result["metadata"]["name"] == "test-id"
|
|
|
|
def test_get_workload_returns_none_when_not_found(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = None
|
|
|
|
result = provider.get_workload("test-id", "test-ns")
|
|
|
|
assert result is None
|
|
|
|
def test_get_workload_falls_back_to_legacy_name(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.side_effect = [
|
|
None,
|
|
{"metadata": {"name": "sandbox-test-id"}},
|
|
]
|
|
|
|
result = provider.get_workload("test-id", "test-ns")
|
|
|
|
assert result["metadata"]["name"] == "sandbox-test-id"
|
|
assert mock_k8s_client.get_custom_object.call_args_list[0].kwargs["name"] == "test-id"
|
|
assert (
|
|
mock_k8s_client.get_custom_object.call_args_list[1].kwargs["name"] == "sandbox-test-id"
|
|
)
|
|
|
|
def test_get_workload_handles_404_gracefully(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
|
|
mock_k8s_client.get_custom_object.return_value = None
|
|
|
|
result = provider.get_workload("test-id", "test-ns")
|
|
|
|
assert result is None
|
|
|
|
def test_get_workload_reraises_non_404_exceptions(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
|
|
# Mock 500 exception
|
|
error = ApiException(status=500)
|
|
mock_k8s_client.get_custom_object.side_effect = error
|
|
|
|
with pytest.raises(ApiException) as exc_info:
|
|
provider.get_workload("test-id", "test-ns")
|
|
|
|
assert exc_info.value.status == 500
|
|
|
|
def test_get_workload_prefers_informer_cache(self, mock_k8s_client):
|
|
cached = {"metadata": {"name": "test-id"}}
|
|
mock_k8s_client.get_custom_object.return_value = cached
|
|
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
|
|
result = provider.get_workload("test-id", "test-ns")
|
|
|
|
assert result == cached
|
|
mock_k8s_client.get_custom_object.assert_called()
|
|
|
|
def test_get_workload_logs_unexpected_errors(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.side_effect = RuntimeError("Unexpected")
|
|
|
|
with pytest.raises(RuntimeError, match="Unexpected"):
|
|
provider.get_workload("test-id", "test-ns")
|
|
|
|
def test_create_workload_updates_informer_cache(self, mock_k8s_client):
|
|
created_body = {"metadata": {"name": "test-id", "uid": "test-uid"}}
|
|
mock_k8s_client.create_custom_object.return_value = created_body
|
|
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
|
|
expires_at = datetime(2025, 12, 31, tzinfo=timezone.utc)
|
|
|
|
result = provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={"FOO": "bar"},
|
|
resource_limits={"cpu": "1", "memory": "1Gi"},
|
|
labels={"opensandbox.io/id": "test-id"},
|
|
expires_at=expires_at,
|
|
execd_image="execd:latest",
|
|
)
|
|
|
|
assert result == {"name": "test-id", "uid": "test-uid", "apiVersion": "sandbox.opensandbox.io/v1alpha1", "kind": "BatchSandbox"}
|
|
|
|
# ===== Workload List Tests =====
|
|
|
|
def test_list_workloads_returns_items(self, mock_k8s_client, mock_batchsandbox_list_response):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.list_custom_objects.return_value = mock_batchsandbox_list_response["items"]
|
|
|
|
result = provider.list_workloads("test-ns", "opensandbox.io/id")
|
|
|
|
assert len(result) == 1
|
|
assert result[0]["metadata"]["name"] == "test-id"
|
|
|
|
def test_list_workloads_returns_empty_on_404(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.list_custom_objects.return_value = []
|
|
|
|
result = provider.list_workloads("test-ns", "opensandbox.io/id")
|
|
|
|
assert result == []
|
|
|
|
# ===== Workload Deletion Tests =====
|
|
|
|
def test_delete_workload_deletes_existing_sandbox(
|
|
self, mock_k8s_client, mock_batchsandbox_list_response
|
|
):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = mock_batchsandbox_list_response["items"][0]
|
|
|
|
provider.delete_workload("test-id", "test-ns")
|
|
|
|
mock_k8s_client.delete_custom_object.assert_called_once_with(
|
|
group="sandbox.opensandbox.io",
|
|
version="v1alpha1",
|
|
namespace="test-ns",
|
|
plural="batchsandboxes",
|
|
name="test-id",
|
|
grace_period_seconds=0,
|
|
)
|
|
|
|
def test_delete_workload_raises_when_not_found(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = None
|
|
|
|
with pytest.raises(Exception) as exc_info:
|
|
provider.delete_workload("test-id", "test-ns")
|
|
|
|
assert "not found" in str(exc_info.value)
|
|
|
|
def test_delete_workload_sets_grace_period_zero(
|
|
self, mock_k8s_client, mock_batchsandbox_list_response
|
|
):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = mock_batchsandbox_list_response["items"][0]
|
|
|
|
provider.delete_workload("test-id", "test-ns")
|
|
|
|
call_kwargs = mock_k8s_client.delete_custom_object.call_args.kwargs
|
|
assert call_kwargs["grace_period_seconds"] == 0
|
|
|
|
# ===== Expiration Time Management Tests =====
|
|
|
|
def test_update_expiration_patches_spec(self, mock_k8s_client, mock_batchsandbox_list_response):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = mock_batchsandbox_list_response["items"][0]
|
|
|
|
expires_at = datetime(2025, 12, 31, 0, 0, 0, tzinfo=timezone.utc)
|
|
provider.update_expiration("test-id", "test-ns", expires_at)
|
|
|
|
call_kwargs = mock_k8s_client.patch_custom_object.call_args.kwargs
|
|
assert call_kwargs["body"] == {"spec": {"expireTime": "2025-12-31T00:00:00+00:00"}}
|
|
|
|
def test_get_expiration_parses_iso_format(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {"spec": {"expireTime": "2025-12-31T10:00:00+00:00"}}
|
|
|
|
result = provider.get_expiration(workload)
|
|
|
|
assert result == datetime(2025, 12, 31, 10, 0, 0, tzinfo=timezone.utc)
|
|
|
|
def test_get_expiration_handles_z_suffix(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {"spec": {"expireTime": "2025-12-31T10:00:00Z"}}
|
|
|
|
result = provider.get_expiration(workload)
|
|
|
|
assert result == datetime(2025, 12, 31, 10, 0, 0, tzinfo=timezone.utc)
|
|
|
|
def test_get_expiration_returns_none_on_invalid_format(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {"spec": {"expireTime": "invalid-date"}}
|
|
|
|
# Should return None and not raise exception
|
|
result = provider.get_expiration(workload)
|
|
|
|
assert result is None
|
|
|
|
def test_get_expiration_returns_none_when_missing(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {"spec": {}}
|
|
|
|
result = provider.get_expiration(workload)
|
|
|
|
assert result is None
|
|
|
|
# ===== Status Retrieval Tests =====
|
|
|
|
def test_get_status_running_with_ip(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {
|
|
"status": {"replicas": 1, "ready": 1, "allocated": 1},
|
|
"metadata": {
|
|
"annotations": {"sandbox.opensandbox.io/endpoints": '["10.0.0.1"]'},
|
|
"creationTimestamp": "2025-12-24T10:00:00Z",
|
|
},
|
|
}
|
|
|
|
result = provider.get_status(workload)
|
|
|
|
assert result["state"] == "Running"
|
|
assert result["reason"] == "POD_READY_WITH_IP"
|
|
assert "IP" in result["message"]
|
|
|
|
def test_get_status_allocated_with_ip_not_ready(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {
|
|
"status": {"replicas": 1, "ready": 0, "allocated": 1},
|
|
"metadata": {
|
|
"annotations": {"sandbox.opensandbox.io/endpoints": '["10.0.0.1"]'},
|
|
"creationTimestamp": "2025-12-24T10:00:00Z",
|
|
},
|
|
}
|
|
|
|
result = provider.get_status(workload)
|
|
|
|
assert result["state"] == "Allocated"
|
|
assert result["reason"] == "IP_ASSIGNED"
|
|
|
|
def test_get_status_pending_scheduled(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {
|
|
"status": {"replicas": 1, "ready": 0, "allocated": 1},
|
|
"metadata": {"creationTimestamp": "2025-12-24T10:00:00Z"},
|
|
}
|
|
|
|
result = provider.get_status(workload)
|
|
|
|
assert result["state"] == "Pending"
|
|
assert result["reason"] == "POD_SCHEDULED"
|
|
|
|
def test_get_status_pending_when_endpoints_invalid_json(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {
|
|
"status": {"replicas": 1, "ready": 0, "allocated": 1},
|
|
"metadata": {
|
|
"annotations": {"sandbox.opensandbox.io/endpoints": "invalid-json"},
|
|
"creationTimestamp": "2025-12-24T10:00:00Z",
|
|
},
|
|
}
|
|
|
|
result = provider.get_status(workload)
|
|
|
|
assert result["state"] == "Pending"
|
|
assert result["reason"] == "POD_SCHEDULED"
|
|
|
|
def test_get_status_pending_when_endpoints_empty_array(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {
|
|
"status": {"replicas": 1, "ready": 0, "allocated": 1},
|
|
"metadata": {
|
|
"annotations": {"sandbox.opensandbox.io/endpoints": "[]"},
|
|
"creationTimestamp": "2025-12-24T10:00:00Z",
|
|
},
|
|
}
|
|
|
|
result = provider.get_status(workload)
|
|
|
|
assert result["state"] == "Pending"
|
|
assert result["reason"] == "POD_SCHEDULED"
|
|
|
|
def test_get_status_pending_unallocated(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {
|
|
"status": {"replicas": 1, "ready": 0, "allocated": 0},
|
|
"metadata": {"creationTimestamp": "2025-12-24T10:00:00Z"},
|
|
}
|
|
|
|
result = provider.get_status(workload)
|
|
|
|
assert result["state"] == "Pending"
|
|
assert result["reason"] == "BATCHSANDBOX_PENDING"
|
|
|
|
def test_get_status_returns_failed_when_pod_unschedulable(self):
|
|
mock_k8s_client = MagicMock()
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.list_pods.return_value = [
|
|
SimpleNamespace(
|
|
status=SimpleNamespace(
|
|
phase="Pending",
|
|
pod_ip=None,
|
|
conditions=[
|
|
SimpleNamespace(
|
|
type="PodScheduled",
|
|
status="False",
|
|
reason="Unschedulable",
|
|
message="0/1 nodes are available: 1 node(s) didn't match Pod's node affinity.",
|
|
)
|
|
],
|
|
)
|
|
)
|
|
]
|
|
workload = {
|
|
"spec": {
|
|
"template": {
|
|
"spec": {
|
|
"nodeSelector": {
|
|
"kubernetes.io/os": "linux",
|
|
"kubernetes.io/arch": "arm64",
|
|
}
|
|
}
|
|
}
|
|
},
|
|
"status": {
|
|
"replicas": 1,
|
|
"ready": 0,
|
|
"allocated": 0,
|
|
"selector": "app=sandbox",
|
|
},
|
|
"metadata": {
|
|
"namespace": "test-ns",
|
|
"creationTimestamp": "2025-12-24T10:00:00Z",
|
|
},
|
|
}
|
|
|
|
result = provider.get_status(workload)
|
|
|
|
assert result["state"] == "Failed"
|
|
assert result["reason"] == "POD_PLATFORM_UNSCHEDULABLE"
|
|
assert "didn't match Pod's node affinity" in result["message"]
|
|
|
|
def test_get_status_keeps_pending_for_generic_failed_scheduling(self):
|
|
mock_k8s_client = MagicMock()
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.list_pods.return_value = [
|
|
SimpleNamespace(
|
|
status=SimpleNamespace(
|
|
phase="Pending",
|
|
pod_ip=None,
|
|
conditions=[
|
|
SimpleNamespace(
|
|
type="PodScheduled",
|
|
status="False",
|
|
reason="Unschedulable",
|
|
message="0/1 nodes are available: 1 Insufficient cpu.",
|
|
)
|
|
],
|
|
)
|
|
)
|
|
]
|
|
workload = {
|
|
"status": {
|
|
"replicas": 1,
|
|
"ready": 0,
|
|
"allocated": 0,
|
|
"selector": "app=sandbox",
|
|
},
|
|
"metadata": {
|
|
"namespace": "test-ns",
|
|
"creationTimestamp": "2025-12-24T10:00:00Z",
|
|
},
|
|
}
|
|
|
|
result = provider.get_status(workload)
|
|
|
|
assert result["state"] == "Pending"
|
|
assert result["reason"] == "BATCHSANDBOX_PENDING"
|
|
|
|
def test_get_status_keeps_pending_when_non_platform_affinity_mismatch(self):
|
|
mock_k8s_client = MagicMock()
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.list_pods.return_value = [
|
|
SimpleNamespace(
|
|
status=SimpleNamespace(
|
|
phase="Pending",
|
|
pod_ip=None,
|
|
conditions=[
|
|
SimpleNamespace(
|
|
type="PodScheduled",
|
|
status="False",
|
|
reason="Unschedulable",
|
|
message="0/1 nodes are available: 1 node(s) didn't match Pod's node affinity.",
|
|
)
|
|
],
|
|
)
|
|
)
|
|
]
|
|
workload = {
|
|
"spec": {
|
|
"template": {
|
|
"spec": {
|
|
"nodeSelector": {
|
|
"kubernetes.io/os": "linux",
|
|
"kubernetes.io/arch": "arm64",
|
|
"zone": "cn-hz-a",
|
|
}
|
|
}
|
|
}
|
|
},
|
|
"status": {
|
|
"replicas": 1,
|
|
"ready": 0,
|
|
"allocated": 0,
|
|
"selector": "app=sandbox",
|
|
},
|
|
"metadata": {
|
|
"namespace": "test-ns",
|
|
"creationTimestamp": "2025-12-24T10:00:00Z",
|
|
},
|
|
}
|
|
|
|
result = provider.get_status(workload)
|
|
|
|
assert result["state"] == "Pending"
|
|
assert result["reason"] == "BATCHSANDBOX_PENDING"
|
|
|
|
def test_get_status_keeps_pending_for_mixed_capacity_and_affinity_message(self):
|
|
mock_k8s_client = MagicMock()
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.list_pods.return_value = [
|
|
SimpleNamespace(
|
|
status=SimpleNamespace(
|
|
phase="Pending",
|
|
pod_ip=None,
|
|
conditions=[
|
|
SimpleNamespace(
|
|
type="PodScheduled",
|
|
status="False",
|
|
reason="Unschedulable",
|
|
message=(
|
|
"0/2 nodes are available: 1 Insufficient cpu, "
|
|
"1 node(s) didn't match Pod's node affinity/selector."
|
|
),
|
|
)
|
|
],
|
|
)
|
|
)
|
|
]
|
|
workload = {
|
|
"spec": {
|
|
"template": {
|
|
"spec": {
|
|
"nodeSelector": {
|
|
"kubernetes.io/os": "linux",
|
|
"kubernetes.io/arch": "arm64",
|
|
}
|
|
}
|
|
}
|
|
},
|
|
"status": {
|
|
"replicas": 1,
|
|
"ready": 0,
|
|
"allocated": 0,
|
|
"selector": "app=sandbox",
|
|
},
|
|
"metadata": {
|
|
"namespace": "test-ns",
|
|
"creationTimestamp": "2025-12-24T10:00:00Z",
|
|
},
|
|
}
|
|
|
|
result = provider.get_status(workload)
|
|
|
|
assert result["state"] == "Pending"
|
|
assert result["reason"] == "BATCHSANDBOX_PENDING"
|
|
|
|
# ===== Endpoint Information Tests =====
|
|
|
|
def test_get_endpoint_info_parses_json_annotation(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {
|
|
"metadata": {"annotations": {"sandbox.opensandbox.io/endpoints": '["10.0.0.1"]'}}
|
|
}
|
|
|
|
result = provider.get_endpoint_info(workload, 8080, "sandbox-123")
|
|
|
|
assert result.endpoint == "10.0.0.1:8080"
|
|
assert result.headers is None
|
|
|
|
def test_get_endpoint_info_uses_first_ip(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {
|
|
"metadata": {
|
|
"annotations": {"sandbox.opensandbox.io/endpoints": '["10.0.0.1", "10.0.0.2"]'}
|
|
}
|
|
}
|
|
|
|
result = provider.get_endpoint_info(workload, 8080, "sandbox-123")
|
|
|
|
assert result.endpoint == "10.0.0.1:8080"
|
|
assert result.headers is None
|
|
|
|
def test_get_endpoint_info_returns_none_when_missing(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {"metadata": {"annotations": {}}}
|
|
|
|
result = provider.get_endpoint_info(workload, 8080, "sandbox-123")
|
|
|
|
assert result is None
|
|
|
|
def test_get_endpoint_info_returns_none_on_invalid_json(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {
|
|
"metadata": {"annotations": {"sandbox.opensandbox.io/endpoints": "invalid-json"}}
|
|
}
|
|
|
|
result = provider.get_endpoint_info(workload, 8080, "sandbox-123")
|
|
|
|
assert result is None
|
|
|
|
def test_get_endpoint_info_returns_none_on_empty_array(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {"metadata": {"annotations": {"sandbox.opensandbox.io/endpoints": "[]"}}}
|
|
|
|
result = provider.get_endpoint_info(workload, 8080, "sandbox-123")
|
|
|
|
assert result is None
|
|
|
|
# ===== Pool-based Creation Tests =====
|
|
|
|
def test_create_workload_poolref_ignores_image_spec(self, mock_k8s_client):
|
|
"""
|
|
Test that pool-based creation ignores image_spec parameter.
|
|
|
|
Pool already defines the image, so image_spec is not used even if provided.
|
|
This verifies backward compatibility - no error is raised.
|
|
"""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "sandbox-test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
result = provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["python", "app.py"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
extensions={"poolRef": "my-pool"},
|
|
)
|
|
|
|
# Should succeed and return workload info
|
|
assert result == {"name": "sandbox-test-id", "uid": "test-uid", "apiVersion": "sandbox.opensandbox.io/v1alpha1", "kind": "BatchSandbox"}
|
|
|
|
# Verify poolRef is used
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
assert body["spec"]["poolRef"] == "my-pool"
|
|
|
|
def test_create_workload_poolref_ignores_resource_limits(self, mock_k8s_client):
|
|
"""
|
|
Test that pool-based creation ignores resource_limits parameter.
|
|
|
|
Pool already defines the resources, so resource_limits is not used even if provided.
|
|
This verifies backward compatibility - no error is raised.
|
|
"""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "sandbox-test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
result = provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri=""),
|
|
entrypoint=["python", "app.py"],
|
|
env={},
|
|
resource_limits={"cpu": "1", "memory": "1Gi"},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
extensions={"poolRef": "my-pool"},
|
|
)
|
|
|
|
# Should succeed and return workload info
|
|
assert result == {"name": "sandbox-test-id", "uid": "test-uid", "apiVersion": "sandbox.opensandbox.io/v1alpha1", "kind": "BatchSandbox"}
|
|
|
|
# Verify poolRef is used
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
assert body["spec"]["poolRef"] == "my-pool"
|
|
|
|
def test_create_workload_poolref_allows_entrypoint_and_env(self, mock_k8s_client):
|
|
"""
|
|
Test that pool-based creation allows customizing entrypoint and env.
|
|
|
|
Verifies taskTemplate structure is correctly generated with user's entrypoint and env.
|
|
"""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "sandbox-test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
result = provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri=""),
|
|
entrypoint=["python", "app.py"],
|
|
env={"FOO": "bar"},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
extensions={"poolRef": "my-pool"},
|
|
)
|
|
|
|
assert result == {"name": "sandbox-test-id", "uid": "test-uid", "apiVersion": "sandbox.opensandbox.io/v1alpha1", "kind": "BatchSandbox"}
|
|
|
|
# Verify the call
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
assert body["spec"]["poolRef"] == "my-pool"
|
|
assert "taskTemplate" in body["spec"]
|
|
|
|
# Verify taskTemplate structure
|
|
task_template = body["spec"]["taskTemplate"]
|
|
assert "spec" in task_template
|
|
assert "process" in task_template["spec"]
|
|
command = task_template["spec"]["process"]["command"]
|
|
assert command[0] == "/bin/sh"
|
|
assert command[1] == "-c"
|
|
# Command should contain bootstrap.sh execution
|
|
# Example: /opt/opensandbox/bootstrap.sh python app.py &
|
|
assert "/opt/opensandbox/bootstrap.sh python app.py" in command[2]
|
|
assert command[2].endswith(" &")
|
|
assert task_template["spec"]["process"]["env"] == [{"name": "FOO", "value": "bar"}]
|
|
|
|
def test_build_task_template_with_env(self, mock_k8s_client):
|
|
"""
|
|
Test _build_task_template with environment variables.
|
|
|
|
Verifies:
|
|
- Command uses shell wrapper: /bin/sh -c "..."
|
|
- Entrypoint executed via bootstrap.sh in background (&)
|
|
- Env list formatted correctly for K8s
|
|
|
|
Generated command example:
|
|
/bin/sh -c "/opt/opensandbox/bootstrap.sh /usr/bin/python app.py &"
|
|
"""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
|
|
result = provider._build_task_template(
|
|
entrypoint=["/usr/bin/python", "app.py"], env={"KEY1": "value1", "KEY2": "value2"}
|
|
)
|
|
|
|
assert "spec" in result
|
|
assert "process" in result["spec"]
|
|
process_task = result["spec"]["process"]
|
|
|
|
# Verify command structure
|
|
command = process_task["command"]
|
|
assert command[0] == "/bin/sh"
|
|
assert command[1] == "-c"
|
|
# Should execute via bootstrap.sh in background (&)
|
|
assert "/opt/opensandbox/bootstrap.sh" in command[2]
|
|
assert "/usr/bin/python" in command[2]
|
|
assert "app.py" in command[2]
|
|
# Should end with & (run in background)
|
|
assert command[2].endswith("&")
|
|
|
|
# Verify env list
|
|
assert process_task["env"] == [
|
|
{"name": "KEY1", "value": "value1"},
|
|
{"name": "KEY2", "value": "value2"},
|
|
]
|
|
|
|
def test_build_task_template_without_env(self, mock_k8s_client):
|
|
"""
|
|
Test _build_task_template without environment variables.
|
|
|
|
Verifies command is wrapped in shell and executes via bootstrap.sh in background.
|
|
|
|
Generated command example:
|
|
/bin/sh -c "/opt/opensandbox/bootstrap.sh /usr/bin/python app.py &"
|
|
"""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
|
|
result = provider._build_task_template(entrypoint=["/usr/bin/python", "app.py"], env={})
|
|
|
|
assert "spec" in result
|
|
assert "process" in result["spec"]
|
|
process_task = result["spec"]["process"]
|
|
assert process_task["env"] == []
|
|
# Without env, command directly calls bootstrap.sh in background
|
|
command = process_task["command"]
|
|
assert command[0] == "/bin/sh"
|
|
assert command[1] == "-c"
|
|
# Check escaped entrypoint
|
|
assert "/opt/opensandbox/bootstrap.sh" in command[2]
|
|
assert "/usr/bin/python" in command[2]
|
|
assert "app.py" in command[2]
|
|
assert command[2].endswith(" &")
|
|
|
|
def test_build_task_template_uses_default_env_path(self, mock_k8s_client):
|
|
"""
|
|
Test that taskTemplate executes bootstrap.sh properly.
|
|
|
|
Verifies:
|
|
- Entrypoint is properly escaped
|
|
- Command runs in background
|
|
"""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
|
|
result = provider._build_task_template(
|
|
entrypoint=["python", "app.py"], env={"TEST_VAR": "test_value"}
|
|
)
|
|
|
|
command = result["spec"]["process"]["command"][2]
|
|
# Should execute bootstrap.sh in background
|
|
assert "/opt/opensandbox/bootstrap.sh" in command
|
|
assert "python" in command
|
|
assert "app.py" in command
|
|
assert command.endswith(" &")
|
|
|
|
def test_build_task_template_escapes_special_characters(self, mock_k8s_client):
|
|
"""
|
|
Test that taskTemplate properly escapes arguments with spaces, quotes, and special chars.
|
|
|
|
This prevents shell injection and ensures arguments are preserved correctly.
|
|
For example: ['python', '-c', 'print("a b")'] should work correctly.
|
|
"""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
|
|
result = provider._build_task_template(
|
|
entrypoint=["python", "-c", 'print("hello world")'],
|
|
env={"KEY": "value with spaces", "QUOTE": "it's fine"},
|
|
)
|
|
|
|
command = result["spec"]["process"]["command"][2]
|
|
|
|
# Verify entrypoint args are properly escaped
|
|
assert "python" in command
|
|
assert "-c" in command
|
|
# The python code with spaces and quotes should be properly escaped
|
|
assert "'print(" in command or '"print(' in command # Escaped
|
|
|
|
# Verify env is passed through env list, not in command
|
|
env_list = result["spec"]["process"]["env"]
|
|
assert {"name": "KEY", "value": "value with spaces"} in env_list
|
|
assert {"name": "QUOTE", "value": "it's fine"} in env_list
|
|
|
|
def test_create_workload_poolref_builds_correct_manifest(self, mock_k8s_client):
|
|
"""
|
|
Test complete pool-based BatchSandbox manifest structure.
|
|
|
|
Verifies:
|
|
- Basic metadata (apiVersion, kind, name, labels)
|
|
- Pool-specific fields (poolRef, taskTemplate, expireTime)
|
|
- No template field (pool mode doesn't use pod template)
|
|
"""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
expires_at = datetime(2025, 12, 31, 10, 0, 0, tzinfo=timezone.utc)
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri=""),
|
|
entrypoint=["python", "app.py"],
|
|
env={"FOO": "bar"},
|
|
resource_limits={},
|
|
labels={"test": "label"},
|
|
expires_at=expires_at,
|
|
execd_image="execd:latest",
|
|
extensions={"poolRef": "test-pool"},
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
|
|
# Verify basic structure
|
|
assert body["apiVersion"] == "sandbox.opensandbox.io/v1alpha1"
|
|
assert body["kind"] == "BatchSandbox"
|
|
assert body["metadata"]["name"] == "test-id"
|
|
assert body["metadata"]["labels"] == {"test": "label"}
|
|
|
|
# Verify pool-specific fields
|
|
assert body["spec"]["replicas"] == 1
|
|
assert body["spec"]["poolRef"] == "test-pool"
|
|
assert body["spec"]["expireTime"] == "2025-12-31T10:00:00+00:00"
|
|
assert "taskTemplate" in body["spec"]
|
|
|
|
# Verify no template field (pool-based doesn't use template)
|
|
assert "template" not in body["spec"]
|
|
|
|
def test_create_workload_poolref_default_entrypoint_no_env_omits_task_template(
|
|
self, mock_k8s_client
|
|
):
|
|
"""When entrypoint is SDK default and env is empty, taskTemplate is omitted."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="dockurr/windows:latest"),
|
|
entrypoint=["tail", "-f", "/dev/null"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=None,
|
|
execd_image="execd:latest",
|
|
extensions={"poolRef": "my-pool"},
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
assert body["spec"]["poolRef"] == "my-pool"
|
|
assert "taskTemplate" not in body["spec"]
|
|
|
|
def test_create_workload_poolref_default_entrypoint_with_env_includes_task_template(
|
|
self, mock_k8s_client
|
|
):
|
|
"""When entrypoint is SDK default but env is non-empty, taskTemplate is generated."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="dockurr/windows:latest"),
|
|
entrypoint=["tail", "-f", "/dev/null"],
|
|
env={"VERSION": "11"},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=None,
|
|
execd_image="execd:latest",
|
|
extensions={"poolRef": "my-pool"},
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
assert body["spec"]["poolRef"] == "my-pool"
|
|
assert "taskTemplate" in body["spec"]
|
|
task_template = body["spec"]["taskTemplate"]
|
|
assert task_template["spec"]["process"]["env"] == [{"name": "VERSION", "value": "11"}]
|
|
|
|
def test_create_workload_poolref_none_entrypoint_no_env_omits_task_template(self, mock_k8s_client):
|
|
"""When entrypoint is None and env is empty, taskTemplate is omitted.
|
|
|
|
SDK pool mode callers omit entrypoint entirely (None), expecting the pool's
|
|
default command to run. This must not raise a TypeError.
|
|
"""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=None,
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=None,
|
|
execd_image="execd:latest",
|
|
extensions={"poolRef": "my-pool"},
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
assert body["spec"]["poolRef"] == "my-pool"
|
|
assert "taskTemplate" not in body["spec"]
|
|
|
|
|
|
class TestBatchSandboxProviderEgress:
|
|
"""BatchSandboxProvider egress sidecar tests"""
|
|
|
|
def test_create_workload_without_network_policy_no_sidecar(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
expires_at = datetime(2025, 12, 31, 10, 0, 0, tzinfo=timezone.utc)
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=expires_at,
|
|
execd_image="execd:latest",
|
|
network_policy=None,
|
|
egress_image=None,
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
containers = pod_spec["containers"]
|
|
|
|
# Should only have main container
|
|
assert len(containers) == 1
|
|
assert containers[0]["name"] == "sandbox"
|
|
# Should not have securityContext with sysctls
|
|
assert "securityContext" not in pod_spec or "sysctls" not in pod_spec.get(
|
|
"securityContext", {}
|
|
)
|
|
|
|
def test_create_workload_with_network_policy_adds_sidecar(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(
|
|
mock_k8s_client,
|
|
_app_config_with_egress_disable_ipv6(),
|
|
)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
expires_at = datetime(2025, 12, 31, 10, 0, 0, tzinfo=timezone.utc)
|
|
network_policy = NetworkPolicy(
|
|
default_action="deny",
|
|
egress=[NetworkRule(action="allow", target="pypi.org")],
|
|
)
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=expires_at,
|
|
execd_image="execd:latest",
|
|
network_policy=network_policy,
|
|
egress_image="opensandbox/egress:v1.1.4",
|
|
credential_proxy_enabled=True,
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
containers = pod_spec["containers"]
|
|
|
|
# Should have both main container and sidecar
|
|
assert len(containers) == 2
|
|
|
|
# Find sidecar container
|
|
sidecar = next((c for c in containers if c["name"] == "egress"), None)
|
|
assert sidecar is not None
|
|
assert sidecar["image"] == "opensandbox/egress:v1.1.4"
|
|
|
|
# Verify sidecar has environment variable
|
|
env_vars = {e["name"]: e["value"] for e in sidecar.get("env", [])}
|
|
assert "OPENSANDBOX_EGRESS_RULES" in env_vars
|
|
assert env_vars["OPENSANDBOX_EGRESS_MODE"] == EGRESS_MODE_DNS
|
|
assert env_vars[OPENSANDBOX_EGRESS_MITMPROXY_TRANSPARENT] == "true"
|
|
|
|
caps = sidecar.get("securityContext", {}).get("capabilities", {})
|
|
assert "NET_ADMIN" in caps.get("add", [])
|
|
assert sidecar.get("securityContext", {}).get("privileged") is not True
|
|
assert "command" not in sidecar
|
|
assert sidecar["readinessProbe"]["httpGet"]["path"] == "/healthz"
|
|
assert sidecar["readinessProbe"]["httpGet"]["port"] == 18080
|
|
|
|
inits = pod_spec.get("initContainers", [])
|
|
assert len(inits) == 1
|
|
execd_init = inits[0]
|
|
assert execd_init["name"] == "execd-installer"
|
|
assert execd_init["image"] == "execd:latest"
|
|
assert execd_init.get("securityContext", {}).get("privileged") is True
|
|
assert "/proc/sys/net/ipv6/conf/all/disable_ipv6" in execd_init["args"][0]
|
|
|
|
main = next(c for c in containers if c["name"] == "sandbox")
|
|
main_env = {e["name"]: e["value"] for e in main["env"]}
|
|
assert main_env[OPENSANDBOX_EGRESS_MITMPROXY_TRANSPARENT] == "true"
|
|
assert "SSL_CERT_FILE" not in main_env
|
|
assert "REQUESTS_CA_BUNDLE" not in main_env
|
|
assert "CURL_CA_BUNDLE" not in main_env
|
|
assert "GIT_SSL_CAINFO" not in main_env
|
|
assert "NODE_EXTRA_CA_CERTS" not in main_env
|
|
assert "opensandbox-mitm-ca" not in {v["name"] for v in pod_spec["volumes"]}
|
|
assert {
|
|
"name": OPENSANDBOX_RUNTIME_VOLUME_NAME,
|
|
"mountPath": OPENSANDBOX_RUNTIME_MOUNT_PATH,
|
|
} in main["volumeMounts"]
|
|
assert {
|
|
"name": OPENSANDBOX_RUNTIME_VOLUME_NAME,
|
|
"mountPath": OPENSANDBOX_RUNTIME_MOUNT_PATH,
|
|
} in sidecar["volumeMounts"]
|
|
|
|
def test_create_workload_windows_profile_with_network_policy_keeps_ipv6_disable(
|
|
self, mock_k8s_client
|
|
):
|
|
provider = BatchSandboxProvider(
|
|
mock_k8s_client,
|
|
_app_config_with_egress_disable_ipv6(),
|
|
)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="dockurr/windows:latest"),
|
|
entrypoint=["cmd", "/c", "echo hello"],
|
|
env={"VERSION": "11"},
|
|
resource_limits={"cpu": "4", "memory": "8G", "disk": "64G"},
|
|
labels={},
|
|
expires_at=None,
|
|
execd_image="execd:latest",
|
|
platform=PlatformSpec(os="windows", arch="amd64"),
|
|
network_policy=NetworkPolicy(default_action="deny", egress=[]),
|
|
egress_image="opensandbox/egress:v1.1.4",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
|
|
sidecar = next((c for c in pod_spec["containers"] if c["name"] == "egress"), None)
|
|
assert sidecar is not None
|
|
|
|
main_container = next((c for c in pod_spec["containers"] if c["name"] == "sandbox"), None)
|
|
assert main_container is not None
|
|
main_caps = main_container.get("securityContext", {}).get("capabilities", {})
|
|
assert "NET_ADMIN" in main_caps.get("add", [])
|
|
assert "NET_RAW" in main_caps.get("add", [])
|
|
assert "NET_ADMIN" not in main_caps.get("drop", [])
|
|
|
|
execd_init = pod_spec["initContainers"][0]
|
|
assert execd_init.get("securityContext", {}).get("privileged") is True
|
|
assert "/proc/sys/net/ipv6/conf/all/disable_ipv6" in execd_init["args"][0]
|
|
|
|
def test_create_workload_with_network_policy_persists_annotation_and_sidecar_token(
|
|
self, mock_k8s_client
|
|
):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=None,
|
|
execd_image="execd:latest",
|
|
network_policy=NetworkPolicy(default_action="deny", egress=[]),
|
|
egress_image="opensandbox/egress:v1.1.4",
|
|
annotations={SANDBOX_EGRESS_AUTH_TOKEN_METADATA_KEY: "egress-token"},
|
|
egress_auth_token="egress-token",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
assert (
|
|
body["metadata"]["annotations"][SANDBOX_EGRESS_AUTH_TOKEN_METADATA_KEY]
|
|
== "egress-token"
|
|
)
|
|
|
|
containers = body["spec"]["template"]["spec"]["containers"]
|
|
sidecar = next((c for c in containers if c["name"] == "egress"), None)
|
|
assert sidecar is not None
|
|
env_vars = {e["name"]: e["value"] for e in sidecar.get("env", [])}
|
|
assert env_vars[OPENSANDBOX_EGRESS_TOKEN] == "egress-token"
|
|
assert env_vars["OPENSANDBOX_EGRESS_MODE"] == EGRESS_MODE_DNS
|
|
assert sidecar["readinessProbe"]["httpGet"]["httpHeaders"] == [
|
|
{"name": OPEN_SANDBOX_EGRESS_AUTH_HEADER, "value": "egress-token"}
|
|
]
|
|
|
|
def test_create_workload_with_egress_mode_dns_nft(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=None,
|
|
execd_image="execd:latest",
|
|
network_policy=NetworkPolicy(default_action="deny", egress=[]),
|
|
egress_image="opensandbox/egress:v1.1.4",
|
|
egress_mode=EGRESS_MODE_DNS_NFT,
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
containers = body["spec"]["template"]["spec"]["containers"]
|
|
sidecar = next((c for c in containers if c["name"] == "egress"), None)
|
|
assert sidecar is not None
|
|
env_vars = {e["name"]: e["value"] for e in sidecar.get("env", [])}
|
|
assert env_vars["OPENSANDBOX_EGRESS_MODE"] == EGRESS_MODE_DNS_NFT
|
|
|
|
def test_create_workload_with_network_policy_does_not_add_pod_ipv6_sysctls(
|
|
self, mock_k8s_client
|
|
):
|
|
"""IPv6 all.disable is applied in privileged execd init, not Pod sysctls."""
|
|
provider = BatchSandboxProvider(
|
|
mock_k8s_client,
|
|
_app_config_with_egress_disable_ipv6(),
|
|
)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
expires_at = datetime(2025, 12, 31, 10, 0, 0, tzinfo=timezone.utc)
|
|
network_policy = NetworkPolicy(
|
|
default_action="deny",
|
|
egress=[NetworkRule(action="allow", target="example.com")],
|
|
)
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=expires_at,
|
|
execd_image="execd:latest",
|
|
network_policy=network_policy,
|
|
egress_image="opensandbox/egress:v1.1.4",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
|
|
assert "securityContext" not in pod_spec or "sysctls" not in pod_spec.get(
|
|
"securityContext", {}
|
|
)
|
|
|
|
sidecar = next(c for c in pod_spec["containers"] if c["name"] == "egress")
|
|
assert "command" not in sidecar
|
|
execd_init = pod_spec["initContainers"][0]
|
|
assert execd_init["name"] == "execd-installer"
|
|
assert "/proc/sys/net/ipv6/conf/all/disable_ipv6" in execd_init["args"][0]
|
|
|
|
def test_create_workload_with_egress_skips_ipv6_disable_when_not_configured(
|
|
self, mock_k8s_client
|
|
):
|
|
"""With ``egress.disable_ipv6`` false, execd init is not privileged and does not write disable_ipv6."""
|
|
provider = BatchSandboxProvider(
|
|
mock_k8s_client,
|
|
_app_config_with_egress_disable_ipv6(False),
|
|
)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
network_policy = NetworkPolicy(
|
|
default_action="deny",
|
|
egress=[NetworkRule(action="allow", target="example.com")],
|
|
)
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=None,
|
|
execd_image="execd:latest",
|
|
network_policy=network_policy,
|
|
egress_image="opensandbox/egress:v1.1.4",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
execd_init = pod_spec["initContainers"][0]
|
|
assert execd_init["name"] == "execd-installer"
|
|
assert "securityContext" not in execd_init
|
|
assert "/proc/sys/net/ipv6/conf/all/disable_ipv6" not in execd_init["args"][0]
|
|
|
|
def test_create_workload_with_network_policy_drops_net_admin_from_main_container(
|
|
self, mock_k8s_client
|
|
):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
expires_at = datetime(2025, 12, 31, 10, 0, 0, tzinfo=timezone.utc)
|
|
network_policy = NetworkPolicy(
|
|
default_action="deny",
|
|
egress=[NetworkRule(action="allow", target="example.com")],
|
|
)
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=expires_at,
|
|
execd_image="execd:latest",
|
|
network_policy=network_policy,
|
|
egress_image="opensandbox/egress:v1.1.4",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
containers = pod_spec["containers"]
|
|
|
|
# Find main container
|
|
main_container = next((c for c in containers if c["name"] == "sandbox"), None)
|
|
assert main_container is not None
|
|
|
|
# Verify main container has securityContext
|
|
assert "securityContext" in main_container
|
|
assert "capabilities" in main_container["securityContext"]
|
|
assert "drop" in main_container["securityContext"]["capabilities"]
|
|
assert "NET_ADMIN" in main_container["securityContext"]["capabilities"]["drop"]
|
|
|
|
def test_create_workload_without_egress_image_no_sidecar(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
expires_at = datetime(2025, 12, 31, 10, 0, 0, tzinfo=timezone.utc)
|
|
network_policy = NetworkPolicy(
|
|
default_action="deny",
|
|
egress=[NetworkRule(action="allow", target="example.com")],
|
|
)
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=expires_at,
|
|
execd_image="execd:latest",
|
|
network_policy=network_policy,
|
|
egress_image=None,
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
containers = pod_spec["containers"]
|
|
|
|
# Should only have main container
|
|
assert len(containers) == 1
|
|
assert containers[0]["name"] == "sandbox"
|
|
|
|
def test_egress_sidecar_contains_network_policy_in_env(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
expires_at = datetime(2025, 12, 31, 10, 0, 0, tzinfo=timezone.utc)
|
|
network_policy = NetworkPolicy(
|
|
default_action="deny",
|
|
egress=[
|
|
NetworkRule(action="allow", target="pypi.org"),
|
|
NetworkRule(action="deny", target="*.malicious.com"),
|
|
],
|
|
)
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=expires_at,
|
|
execd_image="execd:latest",
|
|
network_policy=network_policy,
|
|
egress_image="opensandbox/egress:v1.1.4",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
containers = pod_spec["containers"]
|
|
|
|
sidecar = next((c for c in containers if c["name"] == "egress"), None)
|
|
assert sidecar is not None
|
|
|
|
env_vars = {e["name"]: e["value"] for e in sidecar.get("env", [])}
|
|
assert "OPENSANDBOX_EGRESS_RULES" in env_vars
|
|
|
|
# Verify the environment variable contains valid JSON with network policy
|
|
import json
|
|
|
|
policy_json = json.loads(env_vars["OPENSANDBOX_EGRESS_RULES"])
|
|
assert policy_json["defaultAction"] == "deny"
|
|
assert len(policy_json["egress"]) == 2
|
|
assert policy_json["egress"][0]["action"] == "allow"
|
|
assert policy_json["egress"][0]["target"] == "pypi.org"
|
|
|
|
def test_main_container_no_security_context_without_network_policy(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
expires_at = datetime(2025, 12, 31, 10, 0, 0, tzinfo=timezone.utc)
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=expires_at,
|
|
execd_image="execd:latest",
|
|
network_policy=None,
|
|
egress_image=None,
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
containers = pod_spec["containers"]
|
|
|
|
main_container = containers[0]
|
|
# Main container should not have securityContext when no network policy
|
|
assert "securityContext" not in main_container
|
|
|
|
def test_create_workload_with_network_policy_works_with_template(
|
|
self, mock_k8s_client, tmp_path
|
|
):
|
|
template_file = tmp_path / "template.yaml"
|
|
template_file.write_text(
|
|
"""
|
|
spec:
|
|
template:
|
|
spec:
|
|
volumes:
|
|
- name: sandbox-shared-data
|
|
emptyDir: {}
|
|
"""
|
|
)
|
|
provider = BatchSandboxProvider(
|
|
mock_k8s_client, _app_config_with_template(str(template_file))
|
|
)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
expires_at = datetime(2025, 12, 31, 10, 0, 0, tzinfo=timezone.utc)
|
|
network_policy = NetworkPolicy(
|
|
default_action="deny",
|
|
egress=[NetworkRule(action="allow", target="example.com")],
|
|
)
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=expires_at,
|
|
execd_image="execd:latest",
|
|
network_policy=network_policy,
|
|
egress_image="opensandbox/egress:v1.1.4",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
containers = pod_spec["containers"]
|
|
|
|
# Should have both main container and sidecar
|
|
assert len(containers) == 2
|
|
|
|
# Verify sidecar exists
|
|
sidecar = next((c for c in containers if c["name"] == "egress"), None)
|
|
assert sidecar is not None
|
|
|
|
# Pod-level IPv6 sysctls are not injected for egress (sidecar startup handles all.disable)
|
|
assert "securityContext" not in pod_spec or "sysctls" not in pod_spec.get(
|
|
"securityContext", {}
|
|
)
|
|
|
|
# Verify template volumes are still merged
|
|
volume_names = [v["name"] for v in pod_spec["volumes"]]
|
|
assert "sandbox-shared-data" in volume_names
|
|
assert "opensandbox-bin" in volume_names
|
|
|
|
# ===== Phase + Condition Validation Tests =====
|
|
|
|
def test_pause_sandbox_running_allows(self, mock_k8s_client):
|
|
"""Test pause allowed when Phase=Succeed."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {"phase": "Succeed", "conditions": []},
|
|
}
|
|
mock_k8s_client.patch_custom_object.return_value = {}
|
|
|
|
provider.pause_sandbox("test-id", "test-ns")
|
|
|
|
mock_k8s_client.patch_custom_object.assert_called_once()
|
|
call_kwargs = mock_k8s_client.patch_custom_object.call_args.kwargs
|
|
assert call_kwargs["body"] == {"spec": {"pause": True}}
|
|
|
|
def test_pause_sandbox_running_with_pause_failed_allows_retry(self, mock_k8s_client):
|
|
"""Test pause retry performs an internal nil->true double patch."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {
|
|
"phase": "Succeed",
|
|
"conditions": [
|
|
{"type": "PauseFailed", "status": "True", "reason": "SnapshotFailed"}
|
|
],
|
|
},
|
|
}
|
|
mock_k8s_client.patch_custom_object.return_value = {}
|
|
|
|
provider.pause_sandbox("test-id", "test-ns")
|
|
|
|
assert mock_k8s_client.patch_custom_object.call_count == 2
|
|
first_patch = mock_k8s_client.patch_custom_object.call_args_list[0].kwargs["body"]
|
|
second_patch = mock_k8s_client.patch_custom_object.call_args_list[1].kwargs["body"]
|
|
assert first_patch == {"spec": {"pause": None}}
|
|
assert second_patch == {"spec": {"pause": True}}
|
|
|
|
def test_patch_pause_with_retry_bridge_accepts_second_patch_timeout_when_readback_matches_target(
|
|
self, mock_k8s_client
|
|
):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
provider.patch_workload = MagicMock(
|
|
side_effect=[{}, ApiException(status=500, reason="timeout")]
|
|
)
|
|
provider.get_workload = MagicMock(
|
|
return_value={
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"spec": {"pause": True},
|
|
}
|
|
)
|
|
|
|
provider._patch_pause_with_retry_bridge("test-id", "test-ns", True)
|
|
|
|
assert provider.patch_workload.call_count == 2
|
|
first_call = provider.patch_workload.call_args_list[0].args
|
|
second_call = provider.patch_workload.call_args_list[1].args
|
|
assert first_call == ("test-id", "test-ns", {"spec": {"pause": None}})
|
|
assert second_call == ("test-id", "test-ns", {"spec": {"pause": True}})
|
|
provider.get_workload.assert_called_once_with("test-id", "test-ns")
|
|
|
|
def test_patch_pause_with_retry_bridge_retries_target_when_readback_still_nil(
|
|
self, mock_k8s_client
|
|
):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
provider.patch_workload = MagicMock(
|
|
side_effect=[{}, ApiException(status=500, reason="timeout"), {}]
|
|
)
|
|
provider.get_workload = MagicMock(
|
|
return_value={
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"spec": {"pause": None},
|
|
}
|
|
)
|
|
|
|
provider._patch_pause_with_retry_bridge("test-id", "test-ns", True)
|
|
|
|
assert provider.patch_workload.call_count == 3
|
|
first_call = provider.patch_workload.call_args_list[0].args
|
|
second_call = provider.patch_workload.call_args_list[1].args
|
|
third_call = provider.patch_workload.call_args_list[2].args
|
|
assert first_call == ("test-id", "test-ns", {"spec": {"pause": None}})
|
|
assert second_call == ("test-id", "test-ns", {"spec": {"pause": True}})
|
|
assert third_call == ("test-id", "test-ns", {"spec": {"pause": True}})
|
|
provider.get_workload.assert_called_once_with("test-id", "test-ns")
|
|
|
|
def test_pause_sandbox_pausing_rejects(self, mock_k8s_client):
|
|
"""Test pause rejected when Phase=Pausing."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {"phase": "Pausing", "conditions": []},
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="operation in progress"):
|
|
provider.pause_sandbox("test-id", "test-ns")
|
|
|
|
def test_pause_sandbox_resuming_rejects(self, mock_k8s_client):
|
|
"""Test pause rejected when Phase=Resuming."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {"phase": "Resuming", "conditions": []},
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="operation in progress"):
|
|
provider.pause_sandbox("test-id", "test-ns")
|
|
|
|
def test_pause_sandbox_paused_rejects(self, mock_k8s_client):
|
|
"""Test pause rejected when Phase=Paused."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {"phase": "Paused", "conditions": []},
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="already paused"):
|
|
provider.pause_sandbox("test-id", "test-ns")
|
|
|
|
def test_pause_sandbox_failed_rejects(self, mock_k8s_client):
|
|
"""Test pause rejected when Phase=Failed."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {"phase": "Failed", "conditions": []},
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="not available"):
|
|
provider.pause_sandbox("test-id", "test-ns")
|
|
|
|
def test_pause_sandbox_failed_with_pause_failed_rejects(self, mock_k8s_client):
|
|
"""Test pause rejected when Phase=Failed + PauseFailed=True (pod loss scenario)."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {
|
|
"phase": "Failed",
|
|
"conditions": [{"type": "PauseFailed", "status": "True", "reason": "PodNotFound"}],
|
|
},
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="pause caused pod loss"):
|
|
provider.pause_sandbox("test-id", "test-ns")
|
|
|
|
def test_pause_sandbox_pending_rejects(self, mock_k8s_client):
|
|
"""Test pause rejected when Phase=Pending."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {"phase": "Pending", "conditions": []},
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="being created"):
|
|
provider.pause_sandbox("test-id", "test-ns")
|
|
|
|
def test_resume_sandbox_paused_allows(self, mock_k8s_client):
|
|
"""Test resume allowed when Phase=Paused."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {"phase": "Paused", "conditions": []},
|
|
}
|
|
mock_k8s_client.patch_custom_object.return_value = {}
|
|
|
|
provider.resume_sandbox("test-id", "test-ns")
|
|
|
|
mock_k8s_client.patch_custom_object.assert_called_once()
|
|
call_kwargs = mock_k8s_client.patch_custom_object.call_args.kwargs
|
|
assert call_kwargs["body"] == {"spec": {"pause": False}}
|
|
|
|
def test_resume_sandbox_paused_with_resume_failed_allows_retry(self, mock_k8s_client):
|
|
"""Test resume retry performs an internal nil->false double patch."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {
|
|
"phase": "Paused",
|
|
"conditions": [
|
|
{"type": "ResumeFailed", "status": "True", "reason": "SnapshotNotReady"}
|
|
],
|
|
},
|
|
}
|
|
mock_k8s_client.patch_custom_object.return_value = {}
|
|
|
|
provider.resume_sandbox("test-id", "test-ns")
|
|
|
|
assert mock_k8s_client.patch_custom_object.call_count == 2
|
|
first_patch = mock_k8s_client.patch_custom_object.call_args_list[0].kwargs["body"]
|
|
second_patch = mock_k8s_client.patch_custom_object.call_args_list[1].kwargs["body"]
|
|
assert first_patch == {"spec": {"pause": None}}
|
|
assert second_patch == {"spec": {"pause": False}}
|
|
|
|
def test_resume_sandbox_resuming_rejects(self, mock_k8s_client):
|
|
"""Test resume rejected when Phase=Resuming."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {"phase": "Resuming", "conditions": []},
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="operation in progress"):
|
|
provider.resume_sandbox("test-id", "test-ns")
|
|
|
|
def test_resume_sandbox_pausing_rejects(self, mock_k8s_client):
|
|
"""Test resume rejected when Phase=Pausing."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {"phase": "Pausing", "conditions": []},
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="operation in progress"):
|
|
provider.resume_sandbox("test-id", "test-ns")
|
|
|
|
def test_resume_sandbox_running_rejects(self, mock_k8s_client):
|
|
"""Test resume rejected when Phase=Succeed."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {"phase": "Succeed", "conditions": []},
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="expected Paused"):
|
|
provider.resume_sandbox("test-id", "test-ns")
|
|
|
|
def test_get_status_succeed_phase_maps_to_running_state(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {
|
|
"status": {"phase": "Succeed"},
|
|
"metadata": {"creationTimestamp": "2025-12-24T10:00:00Z"},
|
|
}
|
|
|
|
result = provider.get_status(workload)
|
|
|
|
assert result["state"] == "Running"
|
|
assert result["reason"] == "RUNNING"
|
|
assert result["message"] == "Sandbox is running"
|
|
|
|
def test_get_status_failed_uses_condition_message(self):
|
|
provider = BatchSandboxProvider(MagicMock())
|
|
workload = {
|
|
"status": {
|
|
"phase": "Failed",
|
|
"conditions": [
|
|
{
|
|
"type": "ResumeFailed",
|
|
"status": "True",
|
|
"reason": "PodStartFailed",
|
|
"message": "Pod sandbox-abc-0: ImagePullBackOff - image not found",
|
|
}
|
|
],
|
|
},
|
|
"metadata": {"creationTimestamp": "2025-12-24T10:00:00Z"},
|
|
}
|
|
|
|
result = provider.get_status(workload)
|
|
|
|
assert result["state"] == "Failed"
|
|
assert result["reason"] == "FAILED"
|
|
assert result["message"] == "Pod sandbox-abc-0: ImagePullBackOff - image not found"
|
|
|
|
def test_resume_sandbox_failed_rejects(self, mock_k8s_client):
|
|
"""Test resume rejected when Phase=Failed."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {"phase": "Failed", "conditions": []},
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="not available"):
|
|
provider.resume_sandbox("test-id", "test-ns")
|
|
|
|
def test_resume_sandbox_failed_with_resume_failed_rejects(self, mock_k8s_client):
|
|
"""Test resume rejected when Phase=Failed + ResumeFailed=True (pod start failure)."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {
|
|
"phase": "Failed",
|
|
"conditions": [
|
|
{"type": "ResumeFailed", "status": "True", "reason": "PodStartFailed"}
|
|
],
|
|
},
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="resume caused pod start failure"):
|
|
provider.resume_sandbox("test-id", "test-ns")
|
|
|
|
def test_resume_sandbox_pending_rejects(self, mock_k8s_client):
|
|
"""Test resume rejected when Phase=Pending."""
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.get_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "namespace": "test-ns"},
|
|
"status": {"phase": "Pending", "conditions": []},
|
|
}
|
|
|
|
with pytest.raises(ValueError, match="being created"):
|
|
provider.resume_sandbox("test-id", "test-ns")
|
|
|
|
# ===== Image Auth Tests =====
|
|
|
|
def test_supports_image_auth_returns_true(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
assert provider.supports_image_auth() is True
|
|
|
|
def test_create_workload_with_image_auth_injects_image_pull_secrets(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "uid-123"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(
|
|
uri="registry.example.com/img:tag",
|
|
auth=ImageAuth(username="user", password="pass"),
|
|
),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pull_secrets = body["spec"]["template"]["spec"].get("imagePullSecrets")
|
|
assert pull_secrets == [{"name": f"{IMAGE_AUTH_SECRET_PREFIX}-test-id"}]
|
|
|
|
def test_create_workload_with_image_auth_creates_secret(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "uid-abc"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(
|
|
uri="registry.example.com/img:tag",
|
|
auth=ImageAuth(username="user", password="pass"),
|
|
),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
)
|
|
|
|
mock_k8s_client.create_secret.assert_called_once()
|
|
call_kwargs = mock_k8s_client.create_secret.call_args.kwargs
|
|
assert call_kwargs["namespace"] == "test-ns"
|
|
secret = call_kwargs["body"]
|
|
assert secret.type == "kubernetes.io/dockerconfigjson"
|
|
ref = secret.metadata.owner_references[0]
|
|
assert ref.uid == "uid-abc"
|
|
assert ref.kind == "BatchSandbox"
|
|
assert ref.name == "test-id"
|
|
|
|
def test_create_workload_without_image_auth_skips_secret(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "uid-123"}
|
|
}
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
)
|
|
|
|
mock_k8s_client.create_secret.assert_not_called()
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
assert "imagePullSecrets" not in body["spec"]["template"]["spec"]
|
|
|
|
def test_create_workload_with_image_auth_secret_failure_rolls_back_batchsandbox(
|
|
self, mock_k8s_client
|
|
):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "uid-123"}
|
|
}
|
|
mock_k8s_client.create_secret.side_effect = ApiException(status=403)
|
|
|
|
with pytest.raises(ApiException):
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(
|
|
uri="registry.example.com/img:tag",
|
|
auth=ImageAuth(username="user", password="pass"),
|
|
),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
)
|
|
|
|
mock_k8s_client.delete_custom_object.assert_called_once_with(
|
|
group=provider.group,
|
|
version=provider.version,
|
|
namespace="test-ns",
|
|
plural=provider.plural,
|
|
name="test-id",
|
|
grace_period_seconds=0,
|
|
)
|
|
|
|
# ===== Volume Support Tests =====
|
|
|
|
def test_create_workload_with_pvc_volume(self, mock_k8s_client):
|
|
"""
|
|
Test creating workload with PVC volume mount.
|
|
|
|
Verifies:
|
|
- PVC volume is correctly added to pod spec
|
|
- Volume mount is added to main container
|
|
- claimName is correctly set
|
|
"""
|
|
from opensandbox_server.api.schema import Volume, PVC
|
|
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
expires_at = datetime(2025, 12, 31, 10, 0, 0, tzinfo=timezone.utc)
|
|
|
|
volumes = [
|
|
Volume(
|
|
name="data-volume",
|
|
pvc=PVC(claim_name="my-pvc"),
|
|
mount_path="/mnt/data",
|
|
read_only=False,
|
|
)
|
|
]
|
|
|
|
result = provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=expires_at,
|
|
execd_image="execd:latest",
|
|
volumes=volumes,
|
|
)
|
|
|
|
assert result == {"name": "test-id", "uid": "test-uid", "apiVersion": "sandbox.opensandbox.io/v1alpha1", "kind": "BatchSandbox"}
|
|
|
|
def test_create_workload_poolref_rejects_platform(self, mock_k8s_client):
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
|
|
with pytest.raises(
|
|
ValueError, match="platform is not supported together with extensions.poolRef"
|
|
):
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={"opensandbox.io/id": "test-id"},
|
|
expires_at=None,
|
|
execd_image="execd:latest",
|
|
extensions={"poolRef": "warm-pool"},
|
|
platform=PlatformSpec(os="linux", arch="amd64"),
|
|
)
|
|
|
|
def test_create_workload_with_pvc_volume_readonly(self, mock_k8s_client):
|
|
"""
|
|
Test creating workload with read-only PVC volume mount.
|
|
"""
|
|
from opensandbox_server.api.schema import Volume, PVC
|
|
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
volumes = [
|
|
Volume(
|
|
name="models-volume",
|
|
pvc=PVC(claim_name="models-pvc"),
|
|
mount_path="/mnt/models",
|
|
read_only=True,
|
|
)
|
|
]
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
volumes=volumes,
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
|
|
main_container = pod_spec["containers"][0]
|
|
mounts = main_container.get("volumeMounts", [])
|
|
models_mount = next((m for m in mounts if m["name"] == "models-volume"), None)
|
|
models_volume = next((v for v in pod_spec["volumes"] if v["name"] == "models-volume"), None)
|
|
assert models_mount is not None
|
|
assert models_mount["readOnly"] is True
|
|
assert models_volume is not None
|
|
assert models_volume["persistentVolumeClaim"]["readOnly"] is True
|
|
|
|
def test_create_workload_with_pvc_volume_subpath(self, mock_k8s_client):
|
|
"""
|
|
Test creating workload with PVC volume mount with subPath.
|
|
"""
|
|
from opensandbox_server.api.schema import Volume, PVC
|
|
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
volumes = [
|
|
Volume(
|
|
name="data-volume",
|
|
pvc=PVC(claim_name="shared-pvc"),
|
|
mount_path="/mnt/data",
|
|
sub_path="task-001",
|
|
read_only=False,
|
|
)
|
|
]
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
volumes=volumes,
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
|
|
main_container = pod_spec["containers"][0]
|
|
mounts = main_container.get("volumeMounts", [])
|
|
data_mount = next((m for m in mounts if m["name"] == "data-volume"), None)
|
|
assert data_mount is not None
|
|
assert data_mount.get("subPath") == "task-001"
|
|
|
|
def test_create_workload_with_host_volume(self, mock_k8s_client):
|
|
"""
|
|
Test creating workload with hostPath volume mount.
|
|
"""
|
|
from opensandbox_server.api.schema import Volume, Host
|
|
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
volumes = [
|
|
Volume(
|
|
name="host-volume",
|
|
host=Host(path="/data/shared"),
|
|
mount_path="/mnt/host",
|
|
read_only=True,
|
|
)
|
|
]
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
volumes=volumes,
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
|
|
# Check volume definition
|
|
volumes_list = pod_spec.get("volumes", [])
|
|
host_volume = next((v for v in volumes_list if v["name"] == "host-volume"), None)
|
|
assert host_volume is not None
|
|
assert host_volume["hostPath"]["path"] == "/data/shared"
|
|
assert host_volume["hostPath"]["type"] == "DirectoryOrCreate"
|
|
|
|
# Check volume mount
|
|
main_container = pod_spec["containers"][0]
|
|
mounts = main_container.get("volumeMounts", [])
|
|
host_mount = next((m for m in mounts if m["name"] == "host-volume"), None)
|
|
assert host_mount is not None
|
|
assert host_mount["mountPath"] == "/mnt/host"
|
|
assert host_mount["readOnly"] is True
|
|
|
|
def test_create_workload_with_multiple_volumes(self, mock_k8s_client):
|
|
"""
|
|
Test creating workload with multiple volumes (PVC and hostPath).
|
|
"""
|
|
from opensandbox_server.api.schema import Volume, PVC, Host
|
|
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
mock_k8s_client.create_custom_object.return_value = {
|
|
"metadata": {"name": "test-id", "uid": "test-uid"}
|
|
}
|
|
|
|
volumes = [
|
|
Volume(
|
|
name="pvc-volume",
|
|
pvc=PVC(claim_name="data-pvc"),
|
|
mount_path="/mnt/data",
|
|
read_only=False,
|
|
),
|
|
Volume(
|
|
name="host-volume",
|
|
host=Host(path="/tmp/cache"),
|
|
mount_path="/mnt/cache",
|
|
read_only=True,
|
|
),
|
|
]
|
|
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
volumes=volumes,
|
|
)
|
|
|
|
body = mock_k8s_client.create_custom_object.call_args.kwargs["body"]
|
|
pod_spec = body["spec"]["template"]["spec"]
|
|
|
|
# Check both volumes exist
|
|
volumes_list = pod_spec.get("volumes", [])
|
|
assert len([v for v in volumes_list if v["name"] in ("pvc-volume", "host-volume")]) == 2
|
|
|
|
# Check both mounts exist
|
|
main_container = pod_spec["containers"][0]
|
|
mounts = main_container.get("volumeMounts", [])
|
|
mount_names = {m["name"] for m in mounts}
|
|
assert "pvc-volume" in mount_names
|
|
assert "host-volume" in mount_names
|
|
|
|
def test_create_workload_pool_mode_rejects_volumes(self, mock_k8s_client):
|
|
"""
|
|
Test that pool mode rejects volumes with clear error message.
|
|
"""
|
|
from opensandbox_server.api.schema import Volume, PVC
|
|
|
|
provider = BatchSandboxProvider(mock_k8s_client)
|
|
|
|
volumes = [
|
|
Volume(
|
|
name="data-volume",
|
|
pvc=PVC(claim_name="my-pvc"),
|
|
mount_path="/mnt/data",
|
|
)
|
|
]
|
|
|
|
with pytest.raises(ValueError, match="Pool mode does not support volumes"):
|
|
provider.create_workload(
|
|
sandbox_id="test-id",
|
|
namespace="test-ns",
|
|
image_spec=ImageSpec(uri="python:3.11"),
|
|
entrypoint=["/bin/bash"],
|
|
env={},
|
|
resource_limits={},
|
|
labels={},
|
|
expires_at=datetime(2025, 12, 31, tzinfo=timezone.utc),
|
|
execd_image="execd:latest",
|
|
extensions={"poolRef": "my-pool"},
|
|
volumes=volumes,
|
|
)
|
|
|
|
def test_apply_volumes_to_pod_spec_empty_volumes(self, mock_k8s_client):
|
|
"""
|
|
Test apply_volumes_to_pod_spec with empty volumes list.
|
|
"""
|
|
pod_spec = {
|
|
"containers": [{"name": "main", "volumeMounts": []}],
|
|
"volumes": [],
|
|
}
|
|
|
|
apply_volumes_to_pod_spec(pod_spec, [])
|
|
|
|
# Should not modify pod_spec
|
|
assert pod_spec["volumes"] == []
|
|
assert pod_spec["containers"][0]["volumeMounts"] == []
|
|
|
|
def test_apply_volumes_to_pod_spec_no_containers(self, mock_k8s_client):
|
|
"""
|
|
Test apply_volumes_to_pod_spec with no containers returns early without error.
|
|
"""
|
|
from opensandbox_server.api.schema import Volume, PVC
|
|
|
|
pod_spec = {"volumes": []}
|
|
volumes = [Volume(name="test", pvc=PVC(claim_name="pvc"), mount_path="/mnt")]
|
|
|
|
# Should not raise exception
|
|
apply_volumes_to_pod_spec(pod_spec, volumes)
|
|
|
|
# Pod spec should remain unchanged (no containers to mount to)
|
|
assert pod_spec["volumes"] == []
|
|
|
|
def test_apply_volumes_to_pod_spec_duplicate_internal_volume(self, mock_k8s_client):
|
|
"""
|
|
Test apply_volumes_to_pod_spec rejects volume names that collide with internal volumes.
|
|
"""
|
|
from opensandbox_server.api.schema import Volume, PVC
|
|
|
|
pod_spec = {
|
|
"containers": [{"name": "sandbox", "volumeMounts": []}],
|
|
"volumes": [{"name": "opensandbox-bin", "emptyDir": {}}],
|
|
}
|
|
volumes = [Volume(name="opensandbox-bin", pvc=PVC(claim_name="pvc"), mount_path="/mnt")]
|
|
|
|
# Should raise ValueError for duplicate volume name
|
|
with pytest.raises(ValueError) as exc_info:
|
|
apply_volumes_to_pod_spec(pod_spec, volumes)
|
|
|
|
assert "conflicts with an internal volume" in str(exc_info.value)
|
|
|
|
def test_apply_volumes_to_pod_spec_same_pvc_multiple_mounts(self, mock_k8s_client):
|
|
"""
|
|
When multiple Volume API objects share the same claim_name, only one
|
|
Kubernetes volume is created; multiple volumeMounts reference it (avoids
|
|
CSI driver issues from duplicate PVC volume definitions).
|
|
"""
|
|
from opensandbox_server.api.schema import Volume, PVC
|
|
|
|
pod_spec = {
|
|
"containers": [{"name": "main", "volumeMounts": []}],
|
|
"volumes": [],
|
|
}
|
|
volumes = [
|
|
Volume(
|
|
name="skills",
|
|
pvc=PVC(claim_name="oss-pvc-r"),
|
|
mount_path="/path/to/skills",
|
|
sub_path="skill-hub/publish",
|
|
read_only=True,
|
|
),
|
|
Volume(
|
|
name="draft",
|
|
pvc=PVC(claim_name="oss-pvc-r"),
|
|
mount_path="/path/to/draft",
|
|
sub_path="skill-hub/draft",
|
|
read_only=True,
|
|
),
|
|
]
|
|
|
|
apply_volumes_to_pod_spec(pod_spec, volumes)
|
|
|
|
# One volume definition for the shared PVC (first Volume name used)
|
|
assert len(pod_spec["volumes"]) == 1
|
|
assert pod_spec["volumes"][0]["name"] == "skills"
|
|
shared_volume = next((v for v in pod_spec["volumes"] if v["name"] == "skills"), None)
|
|
assert shared_volume is not None
|
|
assert shared_volume["persistentVolumeClaim"]["claimName"] == "oss-pvc-r"
|
|
assert shared_volume["persistentVolumeClaim"]["readOnly"] is True
|
|
|
|
# Two volumeMounts, both referencing the same volume name
|
|
mounts = pod_spec["containers"][0]["volumeMounts"]
|
|
assert len(mounts) == 2
|
|
by_path = {m["mountPath"]: m for m in mounts}
|
|
assert by_path["/path/to/skills"]["name"] == "skills"
|
|
assert by_path["/path/to/skills"].get("subPath") == "skill-hub/publish"
|
|
assert by_path["/path/to/draft"]["name"] == "skills"
|
|
assert by_path["/path/to/draft"].get("subPath") == "skill-hub/draft"
|
|
|
|
def test_apply_volumes_to_pod_spec_same_pvc_multiple_mounts_readwrite(self, mock_k8s_client):
|
|
"""Shared PVC mounts with read_only=False should keep source-level readOnly false."""
|
|
from opensandbox_server.api.schema import Volume, PVC
|
|
|
|
pod_spec = {
|
|
"containers": [{"name": "main", "volumeMounts": []}],
|
|
"volumes": [],
|
|
}
|
|
volumes = [
|
|
Volume(
|
|
name="skills",
|
|
pvc=PVC(claim_name="oss-pvc-rw"),
|
|
mount_path="/path/to/skills",
|
|
sub_path="skill-hub/publish",
|
|
read_only=False,
|
|
),
|
|
Volume(
|
|
name="draft",
|
|
pvc=PVC(claim_name="oss-pvc-rw"),
|
|
mount_path="/path/to/draft",
|
|
sub_path="skill-hub/draft",
|
|
read_only=False,
|
|
),
|
|
]
|
|
|
|
apply_volumes_to_pod_spec(pod_spec, volumes)
|
|
|
|
assert len(pod_spec["volumes"]) == 1
|
|
shared_volume = next((v for v in pod_spec["volumes"] if v["name"] == "skills"), None)
|
|
assert shared_volume is not None
|
|
assert shared_volume["persistentVolumeClaim"]["claimName"] == "oss-pvc-rw"
|
|
assert shared_volume["persistentVolumeClaim"]["readOnly"] is False
|
|
|
|
mounts = pod_spec["containers"][0]["volumeMounts"]
|
|
assert len(mounts) == 2
|
|
assert all(mount["name"] == "skills" for mount in mounts)
|
|
assert all(mount["readOnly"] is False for mount in mounts)
|
|
|
|
def test_apply_volumes_to_pod_spec_same_pvc_mixed_read_only_raises(self, mock_k8s_client):
|
|
"""Shared PVC mounts with mixed read_only values should fail fast."""
|
|
from opensandbox_server.api.schema import Volume, PVC
|
|
|
|
pod_spec = {
|
|
"containers": [{"name": "main", "volumeMounts": []}],
|
|
"volumes": [],
|
|
}
|
|
volumes = [
|
|
Volume(
|
|
name="skills",
|
|
pvc=PVC(claim_name="oss-pvc-mixed"),
|
|
mount_path="/path/to/skills",
|
|
read_only=False,
|
|
),
|
|
Volume(
|
|
name="draft",
|
|
pvc=PVC(claim_name="oss-pvc-mixed"),
|
|
mount_path="/path/to/draft",
|
|
read_only=True,
|
|
),
|
|
]
|
|
|
|
with pytest.raises(ValueError, match="mixed read_only values"):
|
|
apply_volumes_to_pod_spec(pod_spec, volumes)
|