/ server / tests / test_extensions.py
test_extensions.py
  1  # Copyright 2026 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.extensions import (
 19      ACCESS_RENEW_EXTEND_SECONDS_KEY,
 20      ACCESS_RENEW_EXTEND_SECONDS_MAX,
 21      ACCESS_RENEW_EXTEND_SECONDS_METADATA_KEY,
 22      ACCESS_RENEW_EXTEND_SECONDS_MIN,
 23      apply_access_renew_extend_seconds_to_mapping,
 24      apply_extensions_to_annotations,
 25      validate_extensions
 26  )
 27  
 28  
 29  class TestValidateCreateSandboxExtensionsAccessRenewExtendSeconds:
 30      """access.renew.extend.seconds in [300, 86400] when present."""
 31  
 32      def test_omitted_extensions_ok(self):
 33          assert validate_extensions(None) is None
 34  
 35      def test_extensions_without_key_ok(self):
 36          assert validate_extensions({"other": "x"}) is None
 37  
 38      def test_boundary_min_ok(self):
 39          assert validate_extensions(
 40              {ACCESS_RENEW_EXTEND_SECONDS_KEY: str(ACCESS_RENEW_EXTEND_SECONDS_MIN)}
 41          ) is None
 42  
 43      def test_boundary_max_ok(self):
 44          assert validate_extensions(
 45              {ACCESS_RENEW_EXTEND_SECONDS_KEY: str(ACCESS_RENEW_EXTEND_SECONDS_MAX)}
 46          ) is None
 47  
 48      def test_typical_value_ok(self):
 49          assert validate_extensions({ACCESS_RENEW_EXTEND_SECONDS_KEY: "1800"}) is None
 50  
 51      def test_whitespace_trimmed_ok(self):
 52          assert validate_extensions({ACCESS_RENEW_EXTEND_SECONDS_KEY: "  1800  "}) is None
 53  
 54      def test_below_min_rejected(self):
 55          with pytest.raises(HTTPException) as exc:
 56              validate_extensions(
 57                  {ACCESS_RENEW_EXTEND_SECONDS_KEY: str(ACCESS_RENEW_EXTEND_SECONDS_MIN - 1)}
 58              )
 59          assert exc.value.status_code == 400
 60  
 61      def test_above_max_rejected(self):
 62          with pytest.raises(HTTPException) as exc:
 63              validate_extensions(
 64                  {ACCESS_RENEW_EXTEND_SECONDS_KEY: str(ACCESS_RENEW_EXTEND_SECONDS_MAX + 1)}
 65              )
 66          assert exc.value.status_code == 400
 67  
 68      def test_non_integer_string_rejected(self):
 69          with pytest.raises(HTTPException) as exc:
 70              validate_extensions({ACCESS_RENEW_EXTEND_SECONDS_KEY: "abc"})
 71          assert exc.value.status_code == 400
 72  
 73      def test_empty_string_rejected(self):
 74          with pytest.raises(HTTPException) as exc:
 75              validate_extensions({ACCESS_RENEW_EXTEND_SECONDS_KEY: ""})
 76          assert exc.value.status_code == 400
 77  
 78  
 79  class TestAccessRenewExtendSecondsStorage:
 80      def test_apply_to_mapping_with_mixed_extension_keys(self):
 81          m: dict[str, str] = {}
 82          apply_access_renew_extend_seconds_to_mapping(
 83              m,
 84              {"other": "x", ACCESS_RENEW_EXTEND_SECONDS_KEY: "3600"},
 85          )
 86          assert m == {ACCESS_RENEW_EXTEND_SECONDS_METADATA_KEY: "3600"}
 87  
 88      def test_apply_to_mapping_sets_default_key(self):
 89          m: dict[str, str] = {}
 90          apply_access_renew_extend_seconds_to_mapping(
 91              m, {ACCESS_RENEW_EXTEND_SECONDS_KEY: "1200"}
 92          )
 93          assert m == {ACCESS_RENEW_EXTEND_SECONDS_METADATA_KEY: "1200"}
 94  
 95      def test_apply_to_mapping_noop_when_key_absent(self):
 96          m: dict[str, str] = {"x": "1"}
 97          assert apply_access_renew_extend_seconds_to_mapping(m, {"poolRef": "p"}) is None
 98          assert m == {"x": "1"}
 99  
100  
101  class TestExtensionsToAnnotations:
102      """Extensions with opensandbox.extensions. prefix are propagated to annotations."""
103  
104      def test_single_extension_propagated(self):
105          annotations: dict[str, str] = {}
106          extensions = {"opensandbox.extensions.pool-ref": "my-pool"}
107          apply_extensions_to_annotations(annotations, extensions)
108          assert annotations == {"opensandbox.io/extensions.pool-ref": "my-pool"}
109  
110      def test_multiple_extensions_propagated(self):
111          annotations: dict[str, str] = {}
112          extensions = {
113              "opensandbox.extensions.pool-ref": "my-pool",
114              "opensandbox.extensions.custom-key": "custom-value",
115          }
116          apply_extensions_to_annotations(annotations, extensions)
117          assert annotations == {
118              "opensandbox.io/extensions.pool-ref": "my-pool",
119              "opensandbox.io/extensions.custom-key": "custom-value",
120          }
121  
122      def test_non_prefix_extension_not_propagated(self):
123          annotations: dict[str, str] = {}
124          extensions = {
125              "poolRef": "my-pool",
126              "other.key": "value",
127          }
128          apply_extensions_to_annotations(annotations, extensions)
129          assert annotations == {}
130  
131      def test_mixed_extensions_propagated_only_prefix_keys(self):
132          annotations: dict[str, str] = {}
133          extensions = {
134              "opensandbox.extensions.pool-ref": "my-pool",
135              "poolRef": "ignored",
136              "access.renew.extend.seconds": "1800",
137          }
138          apply_extensions_to_annotations(annotations, extensions)
139          assert annotations == {"opensandbox.io/extensions.pool-ref": "my-pool"}
140  
141      def test_empty_extensions_noop(self):
142          annotations: dict[str, str] = {"existing": "value"}
143          apply_extensions_to_annotations(annotations, None)
144          assert annotations == {"existing": "value"}
145  
146      def test_empty_extensions_dict_noop(self):
147          annotations: dict[str, str] = {"existing": "value"}
148          apply_extensions_to_annotations(annotations, {})
149          assert annotations == {"existing": "value"}
150  
151      def test_preserves_existing_annotations(self):
152          annotations: dict[str, str] = {"existing": "value"}
153          extensions = {"opensandbox.extensions.new-key": "new-value"}
154          apply_extensions_to_annotations(annotations, extensions)
155          assert annotations == {
156              "existing": "value",
157              "opensandbox.io/extensions.new-key": "new-value",
158          }