test_agent_sandbox_service.py
1 # Copyright 2025 Alibaba Group Holding Ltd. 2 # 3 # Licensed under the Apache License, Version 2.0 (the "License"); 4 # you may not use this file except in compliance with the License. 5 # You may obtain a copy of the License at 6 # 7 # http://www.apache.org/licenses/LICENSE-2.0 8 # 9 # Unless required by applicable law or agreed to in writing, software 10 # distributed under the License is distributed on an "AS IS" BASIS, 11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12 # See the License for the specific language governing permissions and 13 # limitations under the License. 14 15 from datetime import datetime, timezone 16 from unittest.mock import MagicMock, patch 17 18 import pytest 19 from fastapi import HTTPException 20 from pydantic import ValidationError 21 22 from opensandbox_server.api.schema import SandboxStatus 23 from opensandbox_server.config import ( 24 AppConfig, 25 RuntimeConfig, 26 ServerConfig, 27 KubernetesRuntimeConfig, 28 AgentSandboxRuntimeConfig, 29 ) 30 from opensandbox_server.services.k8s.kubernetes_service import KubernetesSandboxService 31 from opensandbox_server.services.constants import SandboxErrorCodes 32 33 @pytest.fixture 34 def agent_sandbox_runtime_config(): 35 """Provide agent-sandbox runtime configuration""" 36 return KubernetesRuntimeConfig( 37 kubeconfig_path="/tmp/test-kubeconfig", 38 namespace="test-namespace", 39 service_account="test-sa", 40 workload_provider="agent-sandbox", 41 ) 42 43 @pytest.fixture 44 def agent_sandbox_app_config(agent_sandbox_runtime_config): 45 """Provide complete app configuration (kubernetes + agent-sandbox provider)""" 46 return AppConfig( 47 server=ServerConfig( 48 host="0.0.0.0", 49 port=8080, 50 api_key="test-api-key", 51 ), 52 runtime=RuntimeConfig( 53 type="kubernetes", 54 execd_image="ghcr.io/opensandbox/execd:test", 55 ), 56 kubernetes=agent_sandbox_runtime_config, 57 agent_sandbox=AgentSandboxRuntimeConfig( 58 template_file=None, 59 shutdown_policy="Delete", 60 ingress_enabled=True, 61 ), 62 ) 63 64 @pytest.fixture 65 def app_config_docker(): 66 """Provide Docker type app configuration""" 67 return AppConfig( 68 server=ServerConfig( 69 host="0.0.0.0", 70 port=8080, 71 api_key="test-api-key", 72 ), 73 runtime=RuntimeConfig( 74 type="docker", 75 execd_image="ghcr.io/opensandbox/execd:test", 76 ), 77 kubernetes=None, 78 ) 79 80 class TestAgentSandboxServiceInit: 81 82 def test_init_with_valid_config_succeeds(self, agent_sandbox_runtime_config): 83 config = AppConfig( 84 server=ServerConfig( 85 host="0.0.0.0", 86 port=8080, 87 api_key="test-api-key", 88 ), 89 runtime=RuntimeConfig( 90 type="kubernetes", 91 execd_image="ghcr.io/opensandbox/execd:test", 92 ), 93 kubernetes=agent_sandbox_runtime_config, 94 agent_sandbox=AgentSandboxRuntimeConfig( 95 template_file="/tmp/template.yaml", 96 shutdown_policy="Retain", 97 ingress_enabled=True, 98 ), 99 ) 100 101 with patch("opensandbox_server.services.k8s.kubernetes_service.K8sClient") as mock_k8s_client, patch( 102 "opensandbox_server.services.k8s.kubernetes_service.create_workload_provider" 103 ) as mock_provider_factory: 104 mock_provider_factory.return_value = MagicMock() 105 106 service = KubernetesSandboxService(config) 107 108 assert service.namespace == agent_sandbox_runtime_config.namespace 109 assert service.execd_image == config.runtime.execd_image 110 mock_k8s_client.assert_called_once_with(agent_sandbox_runtime_config) 111 mock_provider_factory.assert_called_once() 112 call_kwargs = mock_provider_factory.call_args.kwargs 113 assert call_kwargs["provider_type"] == "agent-sandbox" 114 assert call_kwargs["app_config"].agent_sandbox.template_file == "/tmp/template.yaml" 115 assert call_kwargs["app_config"].agent_sandbox.shutdown_policy == "Retain" 116 assert call_kwargs["app_config"].kubernetes == agent_sandbox_runtime_config 117 118 def test_init_without_kubernetes_config_raises_error(self): 119 with pytest.raises(ValidationError, match="agent_sandbox block requires kubernetes.workload_provider"): 120 AppConfig( 121 server=ServerConfig( 122 host="0.0.0.0", 123 port=8080, 124 api_key="test-api-key", 125 ), 126 runtime=RuntimeConfig( 127 type="kubernetes", 128 execd_image="ghcr.io/opensandbox/execd:test", 129 ), 130 kubernetes=None, 131 agent_sandbox=AgentSandboxRuntimeConfig(), 132 ) 133 134 def test_init_with_wrong_runtime_type_raises_error(self, app_config_docker): 135 with pytest.raises(ValueError, match="requires runtime.type = 'kubernetes'"): 136 KubernetesSandboxService(app_config_docker) 137 138 def test_init_with_k8s_client_failure_raises_http_exception(self, agent_sandbox_app_config): 139 with patch("opensandbox_server.services.k8s.kubernetes_service.K8sClient") as mock_k8s_client: 140 mock_k8s_client.side_effect = Exception("Failed to load kubeconfig") 141 142 with pytest.raises(HTTPException) as exc_info: 143 KubernetesSandboxService(agent_sandbox_app_config) 144 145 assert exc_info.value.status_code == 503 146 assert "code" in exc_info.value.detail 147 assert exc_info.value.detail["code"] == SandboxErrorCodes.K8S_INITIALIZATION_ERROR 148 149 class TestAgentSandboxServiceBuildSandbox: 150 151 def test_build_sandbox_from_workload_dict(self): 152 service = object.__new__(KubernetesSandboxService) 153 service.namespace = "test-namespace" 154 service.workload_provider = MagicMock( 155 get_workload=MagicMock(), 156 get_expiration=MagicMock(return_value=datetime(2025, 12, 31, tzinfo=timezone.utc)), 157 get_status=MagicMock( 158 return_value={ 159 "state": "Running", 160 "reason": "Ready", 161 "message": "Ready", 162 "last_transition_at": datetime(2025, 12, 31, tzinfo=timezone.utc), 163 } 164 ), 165 ) 166 167 workload = { 168 "metadata": { 169 "labels": { 170 "opensandbox.io/id": "sandbox-id", 171 "team": "platform", 172 }, 173 "creationTimestamp": "2025-12-31T09:00:00Z", 174 }, 175 "spec": { 176 "podTemplate": { 177 "spec": { 178 "containers": [ 179 { 180 "image": "python:3.11", 181 "command": ["/bin/bash"], 182 } 183 ] 184 } 185 } 186 }, 187 } 188 service.workload_provider.get_workload.return_value = workload 189 190 sandbox = service.get_sandbox("sandbox-id") 191 192 assert sandbox.id == "sandbox-id" 193 assert sandbox.image.uri == "python:3.11" 194 assert sandbox.entrypoint == ["/bin/bash"] 195 assert sandbox.metadata == {"team": "platform"} 196 assert isinstance(sandbox.status, SandboxStatus) 197 assert sandbox.status.state == "Running"