test_models_stability.py
1 # 2 # Copyright 2025 Alibaba Group Holding Ltd. 3 # 4 # Licensed under the Apache License, Version 2.0 (the "License"); 5 # you may not use this file except in compliance with the License. 6 # You may obtain a copy of the License at 7 # 8 # http://www.apache.org/licenses/LICENSE-2.0 9 # 10 # Unless required by applicable law or agreed to in writing, software 11 # distributed under the License is distributed on an "AS IS" BASIS, 12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13 # See the License for the specific language governing permissions and 14 # limitations under the License. 15 # 16 from __future__ import annotations 17 18 from datetime import datetime, timezone 19 20 import pytest 21 22 from opensandbox.api.lifecycle.models.create_sandbox_response import ( 23 CreateSandboxResponse as ApiCreateSandboxResponse, 24 ) 25 from opensandbox.api.lifecycle.models.image_spec import ImageSpec as ApiImageSpec 26 from opensandbox.api.lifecycle.models.sandbox import Sandbox as ApiSandbox 27 from opensandbox.api.lifecycle.types import UNSET 28 from opensandbox.models.execd import ( 29 Execution, 30 ExecutionError, 31 ExecutionLogs, 32 ExecutionResult, 33 OutputMessage, 34 ) 35 from opensandbox.models.filesystem import MoveEntry, WriteEntry 36 from opensandbox.models.sandboxes import ( 37 OSSFS, 38 PVC, 39 Host, 40 SandboxFilter, 41 SandboxImageAuth, 42 SandboxImageSpec, 43 SandboxInfo, 44 SandboxStatus, 45 Volume, 46 ) 47 48 49 def test_sandbox_image_spec_supports_positional_image() -> None: 50 spec = SandboxImageSpec("python:3.11") 51 assert spec.image == "python:3.11" 52 53 54 def test_sandbox_image_spec_rejects_blank_image() -> None: 55 with pytest.raises(ValueError): 56 SandboxImageSpec(" ") 57 58 59 def test_api_image_spec_tolerates_omitted_auth() -> None: 60 spec = ApiImageSpec.from_dict({"uri": "python:3.11"}) 61 assert spec.uri == "python:3.11" 62 assert spec.auth is UNSET 63 64 65 def test_api_create_sandbox_response_tolerates_omitted_optional_fields() -> None: 66 response = ApiCreateSandboxResponse.from_dict( 67 { 68 "id": "sandbox-1", 69 "status": {"state": "Running"}, 70 "createdAt": "2025-01-01T00:00:00Z", 71 "entrypoint": ["/bin/sh"], 72 } 73 ) 74 assert response.metadata is UNSET 75 assert response.expires_at is UNSET 76 assert response.status.last_transition_at is UNSET 77 78 79 def test_api_sandbox_tolerates_omitted_optional_fields() -> None: 80 sandbox = ApiSandbox.from_dict( 81 { 82 "id": "sandbox-1", 83 "image": {"uri": "python:3.11"}, 84 "status": {"state": "Running"}, 85 "entrypoint": ["/bin/sh"], 86 "createdAt": "2025-01-01T00:00:00Z", 87 } 88 ) 89 assert sandbox.metadata is UNSET 90 assert sandbox.expires_at is UNSET 91 assert sandbox.status.last_transition_at is UNSET 92 93 94 def test_sandbox_image_auth_rejects_blank_username_and_password() -> None: 95 with pytest.raises(ValueError): 96 SandboxImageAuth(username=" ", password="x") 97 with pytest.raises(ValueError): 98 SandboxImageAuth(username="u", password=" ") 99 100 101 def test_sandbox_filter_validations() -> None: 102 SandboxFilter(page=0, page_size=1) 103 with pytest.raises(ValueError): 104 SandboxFilter(page=-1) 105 with pytest.raises(ValueError): 106 SandboxFilter(page_size=0) 107 108 109 def test_sandbox_status_and_info_alias_dump_is_stable() -> None: 110 status = SandboxStatus( 111 state="RUNNING", last_transition_at=datetime(2025, 1, 1, tzinfo=timezone.utc) 112 ) 113 info = SandboxInfo( 114 id=str(__import__("uuid").uuid4()), 115 status=status, 116 entrypoint=["/bin/sh"], 117 expires_at=datetime(2025, 1, 2, tzinfo=timezone.utc), 118 created_at=datetime(2025, 1, 1, tzinfo=timezone.utc), 119 image=SandboxImageSpec("python:3.11"), 120 metadata={"k": "v"}, 121 ) 122 123 dumped = info.model_dump(by_alias=True, mode="json") 124 assert "expires_at" in dumped 125 assert "created_at" in dumped 126 assert dumped["status"]["last_transition_at"].endswith(("Z", "+00:00")) 127 128 129 def test_sandbox_info_supports_manual_cleanup_expiration() -> None: 130 info = SandboxInfo( 131 id=str(__import__("uuid").uuid4()), 132 status=SandboxStatus(state="RUNNING"), 133 entrypoint=["/bin/sh"], 134 expires_at=None, 135 created_at=datetime(2025, 1, 1, tzinfo=timezone.utc), 136 image=SandboxImageSpec("python:3.11"), 137 ) 138 139 dumped = info.model_dump(by_alias=True, mode="json") 140 assert dumped["expires_at"] is None 141 142 143 def test_filesystem_models_aliases_and_validation() -> None: 144 m = MoveEntry(source="/a", destination="/b") 145 assert m.src == "/a" 146 assert m.dest == "/b" 147 148 with pytest.raises(ValueError): 149 WriteEntry(path=" ", data="x") 150 151 152 # ============================================================================ 153 # Volume Model Tests 154 # ============================================================================ 155 156 157 def test_host_backend_requires_absolute_path() -> None: 158 backend = Host(path="/data/shared") 159 assert backend.path == "/data/shared" 160 161 with pytest.raises(ValueError, match="absolute path"): 162 Host(path="relative/path") 163 164 def test_host_backend_accepts_unix_root_path() -> None: 165 """Unix root path '/' must be accepted.""" 166 assert Host(path="/").path == "/" 167 168 169 def test_host_backend_accepts_unix_nested_path() -> None: 170 """Unix nested absolute path must be accepted.""" 171 assert Host(path="/mnt/host/project").path == "/mnt/host/project" 172 173 174 def test_host_backend_accepts_windows_backslash_path() -> None: 175 """Windows drive path with backslashes must be accepted.""" 176 backend = Host(path="D:\\sandbox-mnt\\ReMe") 177 assert backend.path == "D:\\sandbox-mnt\\ReMe" 178 179 180 def test_host_backend_accepts_windows_forward_slash_path() -> None: 181 """Windows drive path with forward slashes must be accepted.""" 182 backend = Host(path="D:/sandbox-mnt/ReMe") 183 assert backend.path == "D:/sandbox-mnt/ReMe" 184 185 186 def test_host_backend_accepts_windows_drive_root() -> None: 187 """Windows drive root (e.g. 'Z:\\') must be accepted.""" 188 assert Host(path="Z:\\").path == "Z:\\" 189 190 191 def test_host_backend_accepts_windows_lowercase_drive() -> None: 192 """Lowercase drive letter must be accepted.""" 193 assert Host(path="a:/lower").path == "a:/lower" 194 195 196 def test_host_backend_rejects_relative_path() -> None: 197 """Relative path without leading separator must be rejected.""" 198 with pytest.raises(ValueError, match="absolute path"): 199 Host(path="relative/path") 200 201 202 def test_host_backend_rejects_dot_relative_path() -> None: 203 """Dot-relative paths must be rejected.""" 204 with pytest.raises(ValueError, match="absolute path"): 205 Host(path="./local") 206 207 208 def test_host_backend_rejects_parent_traversal_path() -> None: 209 """Parent-traversal paths must be rejected.""" 210 with pytest.raises(ValueError, match="absolute path"): 211 Host(path="../parent") 212 213 214 def test_host_backend_rejects_empty_path() -> None: 215 """Empty string must be rejected.""" 216 with pytest.raises(ValueError, match="absolute path"): 217 Host(path="") 218 219 def test_pvc_backend_rejects_blank_claim_name() -> None: 220 backend = PVC(claimName="my-pvc") 221 assert backend.claim_name == "my-pvc" 222 223 with pytest.raises(ValueError, match="blank"): 224 PVC(claimName=" ") 225 226 227 def test_ossfs_backend_default_version_is_2_0() -> None: 228 backend = OSSFS( 229 bucket="bucket-test-3", 230 endpoint="oss-cn-hangzhou.aliyuncs.com", 231 accessKeyId="ak", 232 accessKeySecret="sk", 233 ) 234 assert backend.version == "2.0" 235 236 237 def test_volume_with_host_backend() -> None: 238 vol = Volume( 239 name="data", 240 host=Host(path="/data/shared"), 241 mountPath="/mnt/data", 242 ) 243 assert vol.name == "data" 244 assert vol.host is not None 245 assert vol.host.path == "/data/shared" 246 assert vol.pvc is None 247 assert vol.mount_path == "/mnt/data" 248 assert vol.read_only is False # default is read-write 249 assert vol.sub_path is None 250 251 252 def test_volume_with_pvc_backend() -> None: 253 vol = Volume( 254 name="models", 255 pvc=PVC(claimName="shared-models"), 256 mountPath="/mnt/models", 257 readOnly=True, 258 subPath="v1", 259 ) 260 assert vol.name == "models" 261 assert vol.host is None 262 assert vol.pvc is not None 263 assert vol.pvc.claim_name == "shared-models" 264 assert vol.mount_path == "/mnt/models" 265 assert vol.read_only is True 266 assert vol.sub_path == "v1" 267 268 269 def test_volume_rejects_blank_name() -> None: 270 with pytest.raises(ValueError, match="blank"): 271 Volume( 272 name=" ", 273 host=Host(path="/data"), 274 mountPath="/mnt", 275 ) 276 277 278 def test_volume_requires_absolute_mount_path() -> None: 279 with pytest.raises(ValueError, match="absolute path"): 280 Volume( 281 name="test", 282 host=Host(path="/data"), 283 mountPath="relative/path", 284 ) 285 286 287 def test_volume_serialization_uses_aliases() -> None: 288 vol = Volume( 289 name="test", 290 pvc=PVC(claimName="my-pvc"), 291 mountPath="/mnt/test", 292 readOnly=True, 293 subPath="sub", 294 ) 295 dumped = vol.model_dump(by_alias=True, mode="json") 296 assert "mountPath" in dumped 297 assert "readOnly" in dumped 298 assert "subPath" in dumped 299 assert dumped["pvc"]["claimName"] == "my-pvc" 300 assert dumped["readOnly"] is True 301 302 303 def test_volume_rejects_no_backend() -> None: 304 """Volume must have exactly one backend specified.""" 305 with pytest.raises(ValueError, match="none was provided"): 306 Volume( 307 name="test", 308 mountPath="/mnt/test", 309 ) 310 311 312 def test_volume_rejects_multiple_backends() -> None: 313 """Volume must have exactly one backend, not multiple.""" 314 with pytest.raises(ValueError, match="multiple were provided"): 315 Volume( 316 name="test", 317 host=Host(path="/data"), 318 pvc=PVC(claimName="my-pvc"), 319 mountPath="/mnt/test", 320 ) 321 322 323 # ============================================================================ 324 # Execution __str__ and .text Tests 325 # ============================================================================ 326 327 328 def _make_output(text: str, *, is_error: bool = False) -> OutputMessage: 329 return OutputMessage(text=text, timestamp=0, is_error=is_error) 330 331 332 def _make_result(text: str) -> ExecutionResult: 333 return ExecutionResult(text=text, timestamp=0) 334 335 336 def test_execution_str_stdout_only() -> None: 337 ex = Execution( 338 logs=ExecutionLogs( 339 stdout=[_make_output("hello"), _make_output("world")], 340 ), 341 ) 342 assert str(ex) == "hello\nworld" 343 344 345 def test_execution_str_with_stderr() -> None: 346 ex = Execution( 347 logs=ExecutionLogs( 348 stdout=[_make_output("ok")], 349 stderr=[_make_output("warn", is_error=True)], 350 ), 351 ) 352 assert str(ex) == "ok\n[stderr]\nwarn" 353 354 355 def test_execution_str_with_error() -> None: 356 ex = Execution( 357 error=ExecutionError(name="RuntimeError", value="boom", timestamp=0), 358 ) 359 assert str(ex) == "[error] RuntimeError: boom" 360 361 362 def test_execution_str_empty() -> None: 363 ex = Execution() 364 assert str(ex) == "" 365 assert ex.complete is None 366 assert ex.exit_code is None 367 368 369 def test_execution_text_property() -> None: 370 ex = Execution( 371 logs=ExecutionLogs( 372 stdout=[_make_output("line1"), _make_output("line2")], 373 stderr=[_make_output("ignored", is_error=True)], 374 ), 375 ) 376 assert ex.text == "line1\nline2" 377 378 379 def test_execution_text_includes_results() -> None: 380 """code-interpreter stores return values in result, not stdout.""" 381 ex = Execution( 382 result=[_make_result("4")], 383 ) 384 assert ex.text == "4" 385 assert str(ex) == "4" 386 387 388 def test_execution_text_combines_stdout_and_results() -> None: 389 ex = Execution( 390 logs=ExecutionLogs( 391 stdout=[_make_output("3.11.14")], 392 ), 393 result=[_make_result("4")], 394 ) 395 assert ex.text == "3.11.14\n4" 396 397 398 def test_execution_text_strips_trailing_newlines() -> None: 399 """code-interpreter streaming sends chunks with trailing newlines.""" 400 ex = Execution( 401 logs=ExecutionLogs( 402 stdout=[_make_output("1\n"), _make_output("2\n")], 403 ), 404 ) 405 assert ex.text == "1\n2" 406 assert str(ex) == "1\n2"