test_schema.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 import pytest 16 from pydantic import ValidationError 17 18 from opensandbox_server.api.schema import ( 19 CreateSandboxRequest, 20 Host, 21 ImageSpec, 22 OSSFS, 23 PlatformSpec, 24 PVC, 25 ResourceLimits, 26 Volume, 27 ) 28 29 30 31 class TestHost: 32 33 def test_valid_path(self): 34 backend = Host(path="/data/opensandbox") 35 assert backend.path == "/data/opensandbox" 36 37 def test_valid_windows_path(self): 38 backend = Host(path=r"D:\sandbox-mnt\ReMe") 39 assert backend.path == r"D:\sandbox-mnt\ReMe" 40 41 def test_path_required(self): 42 with pytest.raises(ValidationError) as exc_info: 43 Host() # type: ignore 44 errors = exc_info.value.errors() 45 assert any(e["loc"] == ("path",) for e in errors) 46 47 def test_serialization(self): 48 backend = Host(path="/data/opensandbox") 49 data = backend.model_dump() 50 assert data == {"path": "/data/opensandbox"} 51 52 def test_deserialization(self): 53 data = {"path": "/data/opensandbox"} 54 backend = Host.model_validate(data) 55 assert backend.path == "/data/opensandbox" 56 57 58 class TestPVC: 59 60 def test_valid_claim_name(self): 61 backend = PVC(claim_name="my-pvc") 62 assert backend.claim_name == "my-pvc" 63 64 def test_claim_name_alias(self): 65 data = {"claimName": "my-pvc"} 66 backend = PVC.model_validate(data) 67 assert backend.claim_name == "my-pvc" 68 69 def test_serialization_uses_alias(self): 70 backend = PVC(claim_name="my-pvc") 71 data = backend.model_dump(by_alias=True, exclude_none=True) 72 assert data == { 73 "claimName": "my-pvc", 74 "createIfNotExists": True, 75 "deleteOnSandboxTermination": False, 76 } 77 78 def test_serialization_with_provisioning_hints(self): 79 """Provisioning hints should serialize with aliases.""" 80 backend = PVC( 81 claim_name="my-pvc", 82 storage_class="ssd", 83 storage="5Gi", 84 access_modes=["ReadWriteOnce"], 85 ) 86 data = backend.model_dump(by_alias=True, exclude_none=True) 87 assert data == { 88 "claimName": "my-pvc", 89 "createIfNotExists": True, 90 "deleteOnSandboxTermination": False, 91 "storageClass": "ssd", 92 "storage": "5Gi", 93 "accessModes": ["ReadWriteOnce"], 94 } 95 96 def test_claim_name_required(self): 97 with pytest.raises(ValidationError) as exc_info: 98 PVC() # type: ignore 99 errors = exc_info.value.errors() 100 assert any("claim_name" in str(e["loc"]) or "claimName" in str(e["loc"]) for e in errors) 101 102 103 class TestOSSFS: 104 105 def test_valid_ossfs(self): 106 backend = OSSFS( 107 bucket="bucket-test-3", 108 endpoint="oss-cn-hangzhou.aliyuncs.com", 109 version="2.0", 110 options=["allow_other"], 111 access_key_id="AKIDEXAMPLE", 112 access_key_secret="SECRETEXAMPLE", 113 ) 114 assert backend.bucket == "bucket-test-3" 115 assert backend.version == "2.0" 116 assert backend.access_key_id == "AKIDEXAMPLE" 117 118 def test_default_ossfs_version_is_2_0(self): 119 backend = OSSFS( 120 bucket="bucket-test-3", 121 endpoint="oss-cn-hangzhou.aliyuncs.com", 122 access_key_id="AKIDEXAMPLE", 123 access_key_secret="SECRETEXAMPLE", 124 ) 125 assert backend.version == "2.0" 126 127 def test_inline_credentials_required(self): 128 with pytest.raises(ValidationError): 129 OSSFS( # type: ignore 130 bucket="bucket-test-3", 131 endpoint="oss-cn-hangzhou.aliyuncs.com", 132 ) 133 134 135 class TestVolume: 136 137 def test_valid_host_volume(self): 138 volume = Volume( 139 name="workdir", 140 host=Host(path="/data/opensandbox"), 141 mount_path="/mnt/work", 142 read_only=False, 143 ) 144 assert volume.name == "workdir" 145 assert volume.host is not None 146 assert volume.host.path == "/data/opensandbox" 147 assert volume.mount_path == "/mnt/work" 148 assert volume.read_only is False 149 assert volume.pvc is None 150 assert volume.sub_path is None 151 152 def test_valid_pvc_volume(self): 153 volume = Volume( 154 name="models", 155 pvc=PVC(claim_name="shared-models-pvc"), 156 mount_path="/mnt/models", 157 read_only=True, 158 ) 159 assert volume.name == "models" 160 assert volume.pvc is not None 161 assert volume.pvc.claim_name == "shared-models-pvc" 162 assert volume.mount_path == "/mnt/models" 163 assert volume.read_only is True 164 assert volume.host is None 165 166 def test_valid_volume_with_subpath(self): 167 volume = Volume( 168 name="workdir", 169 host=Host(path="/data/opensandbox"), 170 mount_path="/mnt/work", 171 read_only=False, 172 sub_path="task-001", 173 ) 174 assert volume.sub_path == "task-001" 175 176 def test_valid_ossfs_volume(self): 177 volume = Volume( 178 name="data", 179 ossfs=OSSFS( 180 bucket="bucket-test-3", 181 endpoint="oss-cn-hangzhou.aliyuncs.com", 182 access_key_id="AKIDEXAMPLE", 183 access_key_secret="SECRETEXAMPLE", 184 ), 185 mount_path="/mnt/data", 186 sub_path="task-001", 187 ) 188 assert volume.ossfs is not None 189 assert volume.ossfs.access_key_id == "AKIDEXAMPLE" 190 assert volume.sub_path == "task-001" 191 192 def test_no_backend_raises(self): 193 with pytest.raises(ValidationError) as exc_info: 194 Volume( 195 name="workdir", 196 mount_path="/mnt/work", 197 read_only=False, 198 ) 199 error_message = str(exc_info.value) 200 assert "backend" in error_message.lower() 201 202 def test_multiple_backends_raises(self): 203 with pytest.raises(ValidationError) as exc_info: 204 Volume( 205 name="workdir", 206 host=Host(path="/data/opensandbox"), 207 pvc=PVC(claim_name="my-pvc"), 208 mount_path="/mnt/work", 209 read_only=False, 210 ) 211 error_message = str(exc_info.value) 212 assert "backend" in error_message.lower() 213 214 def test_serialization_host_volume(self): 215 volume = Volume( 216 name="workdir", 217 host=Host(path="/data/opensandbox"), 218 mount_path="/mnt/work", 219 read_only=False, 220 sub_path="task-001", 221 ) 222 data = volume.model_dump(by_alias=True, exclude_none=True) 223 assert data == { 224 "name": "workdir", 225 "host": {"path": "/data/opensandbox"}, 226 "mountPath": "/mnt/work", 227 "readOnly": False, 228 "subPath": "task-001", 229 } 230 231 def test_serialization_pvc_volume(self): 232 volume = Volume( 233 name="models", 234 pvc=PVC(claim_name="shared-models-pvc"), 235 mount_path="/mnt/models", 236 read_only=True, 237 ) 238 data = volume.model_dump(by_alias=True, exclude_none=True) 239 assert data == { 240 "name": "models", 241 "pvc": { 242 "claimName": "shared-models-pvc", 243 "createIfNotExists": True, 244 "deleteOnSandboxTermination": False, 245 }, 246 "mountPath": "/mnt/models", 247 "readOnly": True, 248 } 249 250 def test_deserialization_host_volume(self): 251 data = { 252 "name": "workdir", 253 "host": {"path": "/data/opensandbox"}, 254 "mountPath": "/mnt/work", 255 "readOnly": False, 256 "subPath": "task-001", 257 } 258 volume = Volume.model_validate(data) 259 assert volume.name == "workdir" 260 assert volume.host is not None 261 assert volume.host.path == "/data/opensandbox" 262 assert volume.mount_path == "/mnt/work" 263 assert volume.read_only is False 264 assert volume.sub_path == "task-001" 265 266 def test_deserialization_pvc_volume(self): 267 data = { 268 "name": "models", 269 "pvc": {"claimName": "shared-models-pvc"}, 270 "mountPath": "/mnt/models", 271 "readOnly": True, 272 } 273 volume = Volume.model_validate(data) 274 assert volume.name == "models" 275 assert volume.pvc is not None 276 assert volume.pvc.claim_name == "shared-models-pvc" 277 assert volume.mount_path == "/mnt/models" 278 assert volume.read_only is True 279 280 def test_serialization_ossfs_volume(self): 281 volume = Volume( 282 name="data", 283 ossfs=OSSFS( 284 bucket="bucket-test-3", 285 endpoint="oss-cn-hangzhou.aliyuncs.com", 286 access_key_id="AKIDEXAMPLE", 287 access_key_secret="SECRETEXAMPLE", 288 ), 289 mount_path="/mnt/data", 290 read_only=False, 291 sub_path="task-001", 292 ) 293 data = volume.model_dump(by_alias=True, exclude_none=True) 294 assert data["ossfs"]["bucket"] == "bucket-test-3" 295 assert data["ossfs"]["accessKeyId"] == "AKIDEXAMPLE" 296 assert data["subPath"] == "task-001" 297 298 299 class TestCreateSandboxRequestWithVolumes: 300 301 def test_request_without_timeout_uses_manual_cleanup(self): 302 request = CreateSandboxRequest( 303 image=ImageSpec(uri="python:3.11"), 304 resource_limits=ResourceLimits({"cpu": "500m", "memory": "512Mi"}), 305 entrypoint=["python", "-c", "print('hello')"], 306 ) 307 assert request.timeout is None 308 309 def test_request_without_volumes(self): 310 request = CreateSandboxRequest( 311 image=ImageSpec(uri="python:3.11"), 312 timeout=3600, 313 resource_limits=ResourceLimits({"cpu": "500m", "memory": "512Mi"}), 314 entrypoint=["python", "-c", "print('hello')"], 315 ) 316 assert request.volumes is None 317 assert request.secure_access is False 318 319 def test_request_with_secure_access(self): 320 request = CreateSandboxRequest.model_validate( 321 { 322 "image": {"uri": "python:3.11"}, 323 "timeout": 3600, 324 "resourceLimits": {"cpu": "500m", "memory": "512Mi"}, 325 "entrypoint": ["python", "-c", "print('hello')"], 326 "secureAccess": True, 327 } 328 ) 329 assert request.secure_access is True 330 331 data = request.model_dump(by_alias=True, exclude_none=True) 332 assert data["secureAccess"] is True 333 334 def test_request_with_empty_volumes(self): 335 request = CreateSandboxRequest( 336 image=ImageSpec(uri="python:3.11"), 337 timeout=3600, 338 resource_limits=ResourceLimits({"cpu": "500m", "memory": "512Mi"}), 339 entrypoint=["python", "-c", "print('hello')"], 340 volumes=[], 341 ) 342 assert request.volumes == [] 343 344 def test_request_with_host_volume(self): 345 request = CreateSandboxRequest( 346 image=ImageSpec(uri="python:3.11"), 347 timeout=3600, 348 resource_limits=ResourceLimits({"cpu": "500m", "memory": "512Mi"}), 349 entrypoint=["python", "-c", "print('hello')"], 350 volumes=[ 351 Volume( 352 name="workdir", 353 host=Host(path="/data/opensandbox"), 354 mount_path="/mnt/work", 355 read_only=False, 356 ) 357 ], 358 ) 359 assert request.volumes is not None 360 assert len(request.volumes) == 1 361 assert request.volumes[0].name == "workdir" 362 363 def test_request_with_pvc_volume(self): 364 request = CreateSandboxRequest( 365 image=ImageSpec(uri="python:3.11"), 366 timeout=3600, 367 resource_limits=ResourceLimits({"cpu": "500m", "memory": "512Mi"}), 368 entrypoint=["python", "-c", "print('hello')"], 369 volumes=[ 370 Volume( 371 name="models", 372 pvc=PVC(claim_name="shared-models-pvc"), 373 mount_path="/mnt/models", 374 read_only=True, 375 ) 376 ], 377 ) 378 assert request.volumes is not None 379 assert len(request.volumes) == 1 380 assert request.volumes[0].pvc is not None 381 assert request.volumes[0].pvc.claim_name == "shared-models-pvc" 382 383 def test_request_with_multiple_volumes(self): 384 request = CreateSandboxRequest( 385 image=ImageSpec(uri="python:3.11"), 386 timeout=3600, 387 resource_limits=ResourceLimits({"cpu": "500m", "memory": "512Mi"}), 388 entrypoint=["python", "-c", "print('hello')"], 389 volumes=[ 390 Volume( 391 name="workdir", 392 host=Host(path="/data/opensandbox"), 393 mount_path="/mnt/work", 394 read_only=False, 395 ), 396 Volume( 397 name="models", 398 pvc=PVC(claim_name="shared-models-pvc"), 399 mount_path="/mnt/models", 400 read_only=True, 401 ), 402 ], 403 ) 404 assert request.volumes is not None 405 assert len(request.volumes) == 2 406 407 def test_request_with_platform(self): 408 request = CreateSandboxRequest( 409 image=ImageSpec(uri="python:3.11"), 410 timeout=3600, 411 platform=PlatformSpec(os="linux", arch="arm64"), 412 resource_limits=ResourceLimits({"cpu": "500m", "memory": "512Mi"}), 413 entrypoint=["python", "-c", "print('hello')"], 414 ) 415 assert request.platform is not None 416 assert request.platform.os == "linux" 417 assert request.platform.arch == "arm64" 418 419 def test_serialization_with_volumes(self): 420 request = CreateSandboxRequest( 421 image=ImageSpec(uri="python:3.11"), 422 timeout=3600, 423 resource_limits=ResourceLimits({"cpu": "500m", "memory": "512Mi"}), 424 entrypoint=["python", "-c", "print('hello')"], 425 volumes=[ 426 Volume( 427 name="workdir", 428 host=Host(path="/data/opensandbox"), 429 mount_path="/mnt/work", 430 read_only=False, 431 sub_path="task-001", 432 ) 433 ], 434 ) 435 data = request.model_dump(by_alias=True, exclude_none=True) 436 assert "volumes" in data 437 assert len(data["volumes"]) == 1 438 assert data["volumes"][0]["name"] == "workdir" 439 assert data["volumes"][0]["mountPath"] == "/mnt/work" 440 assert data["volumes"][0]["readOnly"] is False 441 assert data["volumes"][0]["subPath"] == "task-001" 442 443 def test_deserialization_with_volumes(self): 444 data = { 445 "image": {"uri": "python:3.11"}, 446 "timeout": 3600, 447 "resourceLimits": {"cpu": "500m", "memory": "512Mi"}, 448 "entrypoint": ["python", "-c", "print('hello')"], 449 "volumes": [ 450 { 451 "name": "workdir", 452 "host": {"path": "/data/opensandbox"}, 453 "mountPath": "/mnt/work", 454 "readOnly": False, 455 "subPath": "task-001", 456 }, 457 { 458 "name": "models", 459 "pvc": {"claimName": "shared-models-pvc"}, 460 "mountPath": "/mnt/models", 461 "readOnly": True, 462 }, 463 ], 464 } 465 request = CreateSandboxRequest.model_validate(data) 466 assert request.volumes is not None 467 assert len(request.volumes) == 2 468 469 assert request.volumes[0].name == "workdir" 470 assert request.volumes[0].host is not None 471 assert request.volumes[0].host.path == "/data/opensandbox" 472 assert request.volumes[0].mount_path == "/mnt/work" 473 assert request.volumes[0].read_only is False 474 assert request.volumes[0].sub_path == "task-001" 475 476 assert request.volumes[1].name == "models" 477 assert request.volumes[1].pvc is not None 478 assert request.volumes[1].pvc.claim_name == "shared-models-pvc" 479 assert request.volumes[1].mount_path == "/mnt/models" 480 assert request.volumes[1].read_only is True 481 482 def test_deserialization_with_platform(self): 483 data = { 484 "image": {"uri": "python:3.11"}, 485 "platform": {"os": "linux", "arch": "amd64"}, 486 "timeout": 3600, 487 "resourceLimits": {"cpu": "500m", "memory": "512Mi"}, 488 "entrypoint": ["python", "-c", "print('hello')"], 489 } 490 request = CreateSandboxRequest.model_validate(data) 491 assert request.platform is not None 492 assert request.platform.os == "linux" 493 assert request.platform.arch == "amd64" 494 495 def test_request_rejects_zero_timeout(self): 496 with pytest.raises(ValidationError): 497 CreateSandboxRequest( 498 image=ImageSpec(uri="python:3.11"), 499 timeout=0, 500 resource_limits=ResourceLimits({"cpu": "500m"}), 501 entrypoint=["python", "-c", "print('hello')"], 502 ) 503 504 def test_request_allows_timeout_above_previous_hardcoded_limit(self): 505 request = CreateSandboxRequest( 506 image=ImageSpec(uri="python:3.11"), 507 timeout=172800, 508 resource_limits=ResourceLimits({"cpu": "500m", "memory": "512Mi"}), 509 entrypoint=["python", "-c", "print('hello')"], 510 ) 511 512 assert request.timeout == 172800