test_validators.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 fastapi import HTTPException 17 18 from opensandbox_server.api.schema import Host, OSSFS, PVC, Volume, PlatformSpec 19 from opensandbox_server.services.constants import SandboxErrorCodes 20 from opensandbox_server.services.validators import ( 21 ensure_metadata_labels, 22 ensure_platform_valid, 23 ensure_timeout_within_limit, 24 ensure_valid_host_path, 25 ensure_valid_mount_path, 26 ensure_valid_pvc_name, 27 ensure_valid_sub_path, 28 ensure_valid_volume_name, 29 ensure_volumes_valid, 30 ) 31 32 def test_ensure_platform_valid_accepts_windows_amd64(): 33 platform = PlatformSpec(os="windows", arch="amd64") 34 assert ensure_platform_valid(platform) is None 35 assert platform.os == "windows" 36 assert platform.arch == "amd64" 37 38 39 def test_ensure_platform_valid_accepts_windows_arm64(): 40 platform = PlatformSpec(os="windows", arch="arm64") 41 assert ensure_platform_valid(platform) is None 42 assert platform.os == "windows" 43 assert platform.arch == "arm64" 44 45 def test_ensure_platform_valid_rejects_unsupported_os(): 46 platform = PlatformSpec(os="darwin", arch="amd64") 47 with pytest.raises(HTTPException) as exc_info: 48 assert ensure_platform_valid(platform) is None 49 assert exc_info.value.status_code == 400 50 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_PARAMETER 51 52 def test_ensure_metadata_labels_accepts_common_k8s_forms(): 53 valid_metadata = { 54 "app": "web", 55 "k8s.io/name": "app-1", 56 "example.com/label": "a.b_c-1", 57 "team": "A1_b-2.c", 58 "empty": "", 59 } 60 61 assert ensure_metadata_labels(valid_metadata) is None 62 63 def test_ensure_metadata_labels_allows_none_or_empty(): 64 assert ensure_metadata_labels(None) is None 65 assert ensure_metadata_labels({}) is None 66 67 def test_ensure_metadata_labels_rejects_name_too_long(): 68 """Label name part exceeding 63 characters should be rejected.""" 69 long_name = "a" * 64 70 with pytest.raises(HTTPException) as exc_info: 71 assert ensure_metadata_labels({long_name: "value"}) is None 72 assert exc_info.value.status_code == 400 73 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_METADATA_LABEL 74 75 def test_ensure_metadata_labels_rejects_prefix_too_long(): 76 """Label prefix (DNS subdomain) exceeding 253 characters should be rejected.""" 77 # Build a prefix that is longer than 253 chars: 5 labels of 62 chars = 314 chars 78 label_part = "a" * 62 79 long_prefix = ".".join([label_part] * 5) # 62*5 + 4 = 314 chars 80 key = f"{long_prefix}/name" 81 with pytest.raises(HTTPException) as exc_info: 82 assert ensure_metadata_labels({key: "value"}) is None 83 assert exc_info.value.status_code == 400 84 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_METADATA_LABEL 85 86 def test_ensure_metadata_labels_accepts_key_with_max_length_prefix_and_name(): 87 """Valid key where prefix <= 253 chars and name <= 63 chars but total > 253 should be accepted.""" 88 label_part = "a" * 62 89 prefix = ".".join([label_part] * 4) 90 assert len(prefix) == 251 91 key = f"{prefix}/valid-name" 92 assert ensure_metadata_labels({key: "value"}) is None 93 94 def test_ensure_metadata_labels_rejects_invalid_prefix_format(): 95 """Label prefix with invalid DNS subdomain characters should be rejected.""" 96 with pytest.raises(HTTPException) as exc_info: 97 assert ensure_metadata_labels({"INVALID_PREFIX.io/name": "value"}) is None 98 assert exc_info.value.status_code == 400 99 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_METADATA_LABEL 100 101 def test_ensure_metadata_labels_rejects_value_too_long(): 102 """Label value exceeding 63 characters should be rejected.""" 103 long_value = "a" * 64 104 with pytest.raises(HTTPException) as exc_info: 105 assert ensure_metadata_labels({"app": long_value}) is None 106 assert exc_info.value.status_code == 400 107 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_METADATA_LABEL 108 109 def test_ensure_metadata_labels_rejects_key_with_empty_prefix(): 110 """Key with an empty prefix (starts with '/') should be rejected.""" 111 with pytest.raises(HTTPException) as exc_info: 112 assert ensure_metadata_labels({"/name": "value"}) is None 113 assert exc_info.value.status_code == 400 114 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_METADATA_LABEL 115 116 def test_ensure_metadata_labels_rejects_reserved_prefix(): 117 """User metadata must not use the opensandbox.io/ reserved prefix.""" 118 with pytest.raises(HTTPException) as exc_info: 119 assert ensure_metadata_labels({"opensandbox.io/expires-at": "2030-01-01T00:00:00Z"}) is None 120 assert exc_info.value.status_code == 400 121 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_METADATA_LABEL 122 assert "reserved prefix" in exc_info.value.detail["message"] 123 124 def test_ensure_metadata_labels_rejects_manual_cleanup_key(): 125 """User must not inject the manual-cleanup lifecycle label.""" 126 with pytest.raises(HTTPException) as exc_info: 127 assert ensure_metadata_labels({"opensandbox.io/manual-cleanup": "true"}) is None 128 assert exc_info.value.status_code == 400 129 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_METADATA_LABEL 130 assert "reserved prefix" in exc_info.value.detail["message"] 131 132 def test_ensure_metadata_labels_rejects_arbitrary_reserved_key(): 133 """Any key under opensandbox.io/ should be rejected, not just known labels.""" 134 with pytest.raises(HTTPException) as exc_info: 135 assert ensure_metadata_labels({"opensandbox.io/custom": "value"}) is None 136 assert exc_info.value.status_code == 400 137 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_METADATA_LABEL 138 139 def test_ensure_timeout_within_limit_allows_equal_boundary(): 140 assert ensure_timeout_within_limit(3600, 3600) is None 141 142 def test_ensure_timeout_within_limit_allows_disabled_upper_bound(): 143 assert ensure_timeout_within_limit(7200, None) is None 144 145 def test_ensure_timeout_within_limit_rejects_timeout_above_limit(): 146 with pytest.raises(HTTPException) as exc_info: 147 assert ensure_timeout_within_limit(3601, 3600) is None 148 149 assert exc_info.value.status_code == 400 150 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_PARAMETER 151 152 def test_ensure_timeout_within_limit_rejects_unrepresentable_timeout(): 153 with pytest.raises(HTTPException) as exc_info: 154 ensure_timeout_within_limit(10**20, None) 155 156 assert exc_info.value.status_code == 400 157 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_PARAMETER 158 assert "too large" in exc_info.value.detail["message"] 159 160 class TestEnsureValidVolumeName: 161 162 def test_valid_simple_name(self): 163 """Simple lowercase names should be valid.""" 164 assert ensure_valid_volume_name("workdir") is None 165 assert ensure_valid_volume_name("data") is None 166 assert ensure_valid_volume_name("models") is None 167 168 def test_valid_name_with_numbers(self): 169 """Names with numbers should be valid.""" 170 assert ensure_valid_volume_name("data1") is None 171 assert ensure_valid_volume_name("vol2") is None 172 assert ensure_valid_volume_name("123") is None 173 174 def test_valid_name_with_hyphens(self): 175 """Names with hyphens should be valid.""" 176 assert ensure_valid_volume_name("my-volume") is None 177 assert ensure_valid_volume_name("data-cache-1") is None 178 assert ensure_valid_volume_name("a-b-c") is None 179 180 def test_empty_name_raises(self): 181 """Empty name should raise HTTPException.""" 182 with pytest.raises(HTTPException) as exc_info: 183 ensure_valid_volume_name("") 184 assert exc_info.value.status_code == 400 185 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_VOLUME_NAME 186 187 def test_name_too_long_raises(self): 188 """Name exceeding 63 characters should raise HTTPException.""" 189 long_name = "a" * 64 190 with pytest.raises(HTTPException) as exc_info: 191 ensure_valid_volume_name(long_name) 192 assert exc_info.value.status_code == 400 193 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_VOLUME_NAME 194 195 def test_uppercase_name_raises(self): 196 """Uppercase letters should raise HTTPException.""" 197 with pytest.raises(HTTPException) as exc_info: 198 ensure_valid_volume_name("MyVolume") 199 assert exc_info.value.status_code == 400 200 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_VOLUME_NAME 201 202 def test_underscore_name_raises(self): 203 """Underscores should raise HTTPException (not valid DNS label).""" 204 with pytest.raises(HTTPException) as exc_info: 205 ensure_valid_volume_name("my_volume") 206 assert exc_info.value.status_code == 400 207 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_VOLUME_NAME 208 209 def test_name_starting_with_hyphen_raises(self): 210 """Names starting with hyphen should raise HTTPException.""" 211 with pytest.raises(HTTPException) as exc_info: 212 ensure_valid_volume_name("-volume") 213 assert exc_info.value.status_code == 400 214 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_VOLUME_NAME 215 216 def test_name_ending_with_hyphen_raises(self): 217 """Names ending with hyphen should raise HTTPException.""" 218 with pytest.raises(HTTPException) as exc_info: 219 ensure_valid_volume_name("volume-") 220 assert exc_info.value.status_code == 400 221 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_VOLUME_NAME 222 223 class TestEnsureValidMountPath: 224 225 def test_valid_absolute_path(self): 226 """Absolute paths should be valid.""" 227 assert ensure_valid_mount_path("/mnt/data") is None 228 assert ensure_valid_mount_path("/") is None 229 assert ensure_valid_mount_path("/home/user/work") is None 230 231 def test_empty_path_raises(self): 232 """Empty path should raise HTTPException.""" 233 with pytest.raises(HTTPException) as exc_info: 234 ensure_valid_mount_path("") 235 assert exc_info.value.status_code == 400 236 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_MOUNT_PATH 237 238 def test_relative_path_raises(self): 239 """Relative paths should raise HTTPException.""" 240 with pytest.raises(HTTPException) as exc_info: 241 ensure_valid_mount_path("data/files") 242 assert exc_info.value.status_code == 400 243 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_MOUNT_PATH 244 245 def test_path_not_starting_with_slash_raises(self): 246 """Paths not starting with '/' should raise HTTPException.""" 247 with pytest.raises(HTTPException) as exc_info: 248 ensure_valid_mount_path("mnt/data") 249 assert exc_info.value.status_code == 400 250 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_MOUNT_PATH 251 252 class TestEnsureValidSubPath: 253 254 def test_none_subpath_valid(self): 255 """None subpath should be valid.""" 256 assert ensure_valid_sub_path(None) is None 257 258 def test_empty_subpath_valid(self): 259 """Empty string subpath should be valid.""" 260 assert ensure_valid_sub_path("") is None 261 262 def test_relative_subpath_valid(self): 263 """Relative paths should be valid.""" 264 assert ensure_valid_sub_path("task-001") is None 265 assert ensure_valid_sub_path("user/data") is None 266 assert ensure_valid_sub_path("a/b/c") is None 267 268 def test_absolute_subpath_raises(self): 269 """Absolute paths should raise HTTPException.""" 270 with pytest.raises(HTTPException) as exc_info: 271 ensure_valid_sub_path("/absolute/path") 272 assert exc_info.value.status_code == 400 273 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_SUB_PATH 274 275 def test_path_traversal_raises(self): 276 """Path traversal (..) should raise HTTPException.""" 277 with pytest.raises(HTTPException) as exc_info: 278 ensure_valid_sub_path("../parent") 279 assert exc_info.value.status_code == 400 280 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_SUB_PATH 281 282 def test_embedded_path_traversal_raises(self): 283 """Embedded path traversal should raise HTTPException.""" 284 with pytest.raises(HTTPException) as exc_info: 285 ensure_valid_sub_path("a/../b") 286 assert exc_info.value.status_code == 400 287 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_SUB_PATH 288 289 class TestEnsureValidHostPath: 290 291 def test_valid_absolute_path(self): 292 """Absolute paths should be valid.""" 293 assert ensure_valid_host_path("/data/opensandbox") is None 294 assert ensure_valid_host_path("/tmp") is None 295 296 def test_valid_windows_absolute_path(self): 297 """Windows absolute paths should be valid.""" 298 assert ensure_valid_host_path(r"D:\sandbox-mnt\ReMe") is None 299 assert ensure_valid_host_path("D:/sandbox-mnt/ReMe") is None 300 301 def test_valid_windows_drive_root(self): 302 """Windows drive roots should be valid absolute paths.""" 303 assert ensure_valid_host_path("D:\\") is None 304 assert ensure_valid_host_path("D:/") is None 305 306 def test_empty_path_raises(self): 307 """Empty path should raise HTTPException.""" 308 with pytest.raises(HTTPException) as exc_info: 309 ensure_valid_host_path("") 310 assert exc_info.value.status_code == 400 311 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_HOST_PATH 312 313 def test_relative_path_raises(self): 314 """Relative paths should raise HTTPException.""" 315 with pytest.raises(HTTPException) as exc_info: 316 ensure_valid_host_path("data/files") 317 assert exc_info.value.status_code == 400 318 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_HOST_PATH 319 320 def test_path_with_traversal_raises(self): 321 """Paths with traversal should raise HTTPException.""" 322 with pytest.raises(HTTPException) as exc_info: 323 ensure_valid_host_path("/data/../etc/passwd") 324 assert exc_info.value.status_code == 400 325 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_HOST_PATH 326 327 def test_path_with_double_slash_raises(self): 328 """Paths with double slashes should raise HTTPException.""" 329 with pytest.raises(HTTPException) as exc_info: 330 ensure_valid_host_path("/data//files") 331 assert exc_info.value.status_code == 400 332 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_HOST_PATH 333 334 def test_allowed_prefix_match(self): 335 """Paths under allowed prefixes should be valid.""" 336 allowed = ["/data/opensandbox", "/tmp/sandbox"] 337 assert ensure_valid_host_path("/data/opensandbox/user-a", allowed) is None 338 assert ensure_valid_host_path("/tmp/sandbox/task-1", allowed) is None 339 340 def test_allowed_prefix_exact_match(self): 341 """Exact prefix match should be valid.""" 342 allowed = ["/data/opensandbox"] 343 assert ensure_valid_host_path("/data/opensandbox", allowed) is None 344 345 def test_allowed_prefix_match_windows_paths(self): 346 """Windows paths under an allowed Windows prefix should be valid.""" 347 allowed = [r"D:\sandbox-mnt"] 348 assert ensure_valid_host_path(r"D:\sandbox-mnt\ReMe", allowed) is None 349 assert ensure_valid_host_path("D:/sandbox-mnt/ReMe", allowed) is None 350 351 def test_allowed_prefix_match_windows_paths_is_case_insensitive_for_drive(self): 352 """Drive-letter casing differences should not break allowlist checks.""" 353 allowed = ["D:/sandbox-mnt"] 354 assert ensure_valid_host_path("d:/sandbox-mnt/ReMe", allowed) is None 355 356 def test_path_not_in_allowed_prefix_raises(self): 357 """Paths not under allowed prefixes should raise HTTPException.""" 358 allowed = ["/data/opensandbox"] 359 with pytest.raises(HTTPException) as exc_info: 360 ensure_valid_host_path("/etc/passwd", allowed) 361 assert exc_info.value.status_code == 400 362 assert exc_info.value.detail["code"] == SandboxErrorCodes.HOST_PATH_NOT_ALLOWED 363 364 def test_partial_prefix_match_raises(self): 365 """Partial prefix matches should not be allowed.""" 366 allowed = ["/data/opensandbox"] 367 with pytest.raises(HTTPException) as exc_info: 368 ensure_valid_host_path("/data/opensandbox-evil", allowed) 369 assert exc_info.value.status_code == 400 370 assert exc_info.value.detail["code"] == SandboxErrorCodes.HOST_PATH_NOT_ALLOWED 371 372 class TestEnsureValidPvcName: 373 374 def test_valid_simple_name(self): 375 """Simple lowercase names should be valid.""" 376 assert ensure_valid_pvc_name("my-pvc") is None 377 assert ensure_valid_pvc_name("data-volume") is None 378 assert ensure_valid_pvc_name("pvc1") is None 379 380 def test_empty_name_raises(self): 381 """Empty name should raise HTTPException.""" 382 with pytest.raises(HTTPException) as exc_info: 383 ensure_valid_pvc_name("") 384 assert exc_info.value.status_code == 400 385 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_PVC_NAME 386 387 def test_name_too_long_raises(self): 388 """Name exceeding 253 characters should raise HTTPException.""" 389 long_name = "a" * 254 390 with pytest.raises(HTTPException) as exc_info: 391 ensure_valid_pvc_name(long_name) 392 assert exc_info.value.status_code == 400 393 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_PVC_NAME 394 395 def test_uppercase_name_raises(self): 396 """Uppercase letters should raise HTTPException.""" 397 with pytest.raises(HTTPException) as exc_info: 398 ensure_valid_pvc_name("MyPVC") 399 assert exc_info.value.status_code == 400 400 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_PVC_NAME 401 402 def test_underscore_name_raises(self): 403 """Underscores should raise HTTPException.""" 404 with pytest.raises(HTTPException) as exc_info: 405 ensure_valid_pvc_name("my_pvc") 406 assert exc_info.value.status_code == 400 407 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_PVC_NAME 408 409 class TestEnsureVolumesValid: 410 411 def test_none_volumes_valid(self): 412 """None volumes should be valid.""" 413 assert ensure_volumes_valid(None) is None 414 415 def test_empty_volumes_valid(self): 416 """Empty volumes list should be valid.""" 417 assert ensure_volumes_valid([]) is None 418 419 def test_valid_host_volume(self): 420 """Valid host volume should pass validation.""" 421 volume = Volume( 422 name="workdir", 423 host=Host(path="/data/opensandbox"), 424 mount_path="/mnt/work", 425 read_only=False, 426 ) 427 assert ensure_volumes_valid([volume]) is None 428 429 def test_valid_pvc_volume(self): 430 """Valid PVC volume should pass validation.""" 431 volume = Volume( 432 name="models", 433 pvc=PVC(claim_name="shared-models-pvc"), 434 mount_path="/mnt/models", 435 read_only=True, 436 ) 437 assert ensure_volumes_valid([volume]) is None 438 439 def test_valid_ossfs_volume(self): 440 """Valid OSSFS volume should pass validation.""" 441 volume = Volume( 442 name="oss-data", 443 ossfs=OSSFS( 444 bucket="bucket-test-3", 445 endpoint="oss-cn-hangzhou.aliyuncs.com", 446 access_key_id="AKIDEXAMPLE", 447 access_key_secret="SECRETEXAMPLE", 448 ), 449 mount_path="/mnt/data", 450 read_only=False, 451 sub_path="task-001", 452 ) 453 assert ensure_volumes_valid([volume]) is None 454 455 def test_valid_volume_with_subpath(self): 456 """Valid volume with subPath should pass validation.""" 457 volume = Volume( 458 name="workdir", 459 host=Host(path="/data/opensandbox"), 460 mount_path="/mnt/work", 461 read_only=False, 462 sub_path="task-001", 463 ) 464 assert ensure_volumes_valid([volume]) is None 465 466 def test_multiple_valid_volumes(self): 467 """Multiple valid volumes should pass validation.""" 468 volumes = [ 469 Volume( 470 name="workdir", 471 host=Host(path="/data/opensandbox"), 472 mount_path="/mnt/work", 473 read_only=False, 474 ), 475 Volume( 476 name="models", 477 pvc=PVC(claim_name="shared-models-pvc"), 478 mount_path="/mnt/models", 479 read_only=True, 480 ), 481 ] 482 assert ensure_volumes_valid(volumes) is None 483 484 def test_duplicate_volume_name_raises(self): 485 """Duplicate volume names should raise HTTPException.""" 486 volumes = [ 487 Volume( 488 name="workdir", 489 host=Host(path="/data/a"), 490 mount_path="/mnt/a", 491 read_only=False, 492 ), 493 Volume( 494 name="workdir", # Duplicate name 495 host=Host(path="/data/b"), 496 mount_path="/mnt/b", 497 read_only=False, 498 ), 499 ] 500 with pytest.raises(HTTPException) as exc_info: 501 ensure_volumes_valid(volumes) 502 assert exc_info.value.status_code == 400 503 assert exc_info.value.detail["code"] == SandboxErrorCodes.DUPLICATE_VOLUME_NAME 504 505 def test_invalid_volume_name_rejected_by_pydantic(self): 506 """Invalid volume name should be rejected by Pydantic pattern validation.""" 507 from pydantic import ValidationError 508 509 # Pydantic validates the pattern before our validators run 510 with pytest.raises(ValidationError) as exc_info: 511 Volume( 512 name="Invalid_Name", # Invalid: uppercase and underscore 513 host=Host(path="/data/opensandbox"), 514 mount_path="/mnt/work", 515 read_only=False, 516 ) 517 assert "name" in str(exc_info.value) 518 519 def test_invalid_mount_path_rejected_by_pydantic(self): 520 """Invalid mount path should be rejected by Pydantic pattern validation.""" 521 from pydantic import ValidationError 522 523 # Pydantic validates the pattern before our validators run 524 with pytest.raises(ValidationError) as exc_info: 525 Volume( 526 name="workdir", 527 host=Host(path="/data/opensandbox"), 528 mount_path="relative/path", # Invalid: not absolute 529 read_only=False, 530 ) 531 assert "mount_path" in str(exc_info.value) 532 533 def test_invalid_subpath_raises(self): 534 """Invalid subPath should raise HTTPException.""" 535 volume = Volume( 536 name="workdir", 537 host=Host(path="/data/opensandbox"), 538 mount_path="/mnt/work", 539 read_only=False, 540 sub_path="../escape", # Invalid: path traversal 541 ) 542 with pytest.raises(HTTPException) as exc_info: 543 ensure_volumes_valid([volume]) 544 assert exc_info.value.status_code == 400 545 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_SUB_PATH 546 547 def test_host_path_allowlist_enforced(self): 548 """Host path allowlist should be enforced.""" 549 volume = Volume( 550 name="workdir", 551 host=Host(path="/etc/passwd"), # Not in allowed list 552 mount_path="/mnt/work", 553 read_only=False, 554 ) 555 with pytest.raises(HTTPException) as exc_info: 556 ensure_volumes_valid([volume], allowed_host_prefixes=["/data/opensandbox"]) 557 assert exc_info.value.status_code == 400 558 assert exc_info.value.detail["code"] == SandboxErrorCodes.HOST_PATH_NOT_ALLOWED 559 560 def test_ossfs_invalid_version_rejected_by_schema(self): 561 """Unsupported OSSFS version should be rejected by schema validation.""" 562 from pydantic import ValidationError 563 564 with pytest.raises(ValidationError): 565 OSSFS( 566 bucket="bucket-test-3", 567 endpoint="oss-cn-hangzhou.aliyuncs.com", 568 version="3.0", # type: ignore[arg-type] 569 access_key_id="AKIDEXAMPLE", 570 access_key_secret="SECRETEXAMPLE", 571 ) 572 573 def test_ossfs_missing_inline_credentials_raises(self): 574 """Missing inline credentials should raise HTTPException.""" 575 volume = Volume( 576 name="oss-data", 577 ossfs=OSSFS( 578 bucket="bucket-test-3", 579 endpoint="oss-cn-hangzhou.aliyuncs.com", 580 access_key_id="AKIDEXAMPLE", 581 access_key_secret="SECRETEXAMPLE", 582 ), 583 mount_path="/mnt/data", 584 ) 585 volume.ossfs.access_key_id = None 586 with pytest.raises(HTTPException) as exc_info: 587 ensure_volumes_valid([volume]) 588 assert exc_info.value.status_code == 400 589 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_OSSFS_CREDENTIALS 590 591 def test_ossfs_v1_options_reject_prefixed_entries(self): 592 """OSSFS options should reject prefixed entries for 1.0.""" 593 volume = Volume( 594 name="oss-data", 595 ossfs=OSSFS( 596 bucket="bucket-test-3", 597 endpoint="oss-cn-hangzhou.aliyuncs.com", 598 version="1.0", 599 options=["--allow_other"], 600 access_key_id="AKIDEXAMPLE", 601 access_key_secret="SECRETEXAMPLE", 602 ), 603 mount_path="/mnt/data", 604 ) 605 with pytest.raises(HTTPException) as exc_info: 606 ensure_volumes_valid([volume]) 607 assert exc_info.value.status_code == 400 608 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_OSSFS_OPTION 609 610 def test_ossfs_v2_options_reject_prefixed_entries(self): 611 """OSSFS options should reject prefixed entries for 2.0.""" 612 volume = Volume( 613 name="oss-data", 614 ossfs=OSSFS( 615 bucket="bucket-test-3", 616 endpoint="oss-cn-hangzhou.aliyuncs.com", 617 version="2.0", 618 options=["-o allow_other"], 619 access_key_id="AKIDEXAMPLE", 620 access_key_secret="SECRETEXAMPLE", 621 ), 622 mount_path="/mnt/data", 623 ) 624 with pytest.raises(HTTPException) as exc_info: 625 ensure_volumes_valid([volume]) 626 assert exc_info.value.status_code == 400 627 assert exc_info.value.detail["code"] == SandboxErrorCodes.INVALID_OSSFS_OPTION 628 629 def test_invalid_pvc_name_rejected_by_pydantic(self): 630 """Invalid PVC name should be rejected by Pydantic pattern validation.""" 631 from pydantic import ValidationError 632 633 # Pydantic validates the pattern before our validators run 634 with pytest.raises(ValidationError) as exc_info: 635 PVC(claim_name="Invalid_PVC") # Invalid: uppercase and underscore 636 assert "claim_name" in str(exc_info.value)