diff --git a/kubernetes/e2e_test/test_utils.py b/kubernetes/e2e_test/test_utils.py index ff9ebb35da..f774330b75 100644 --- a/kubernetes/e2e_test/test_utils.py +++ b/kubernetes/e2e_test/test_utils.py @@ -10,16 +10,21 @@ # License for the specific language governing permissions and limitations # under the License. +import datetime import unittest from decimal import Decimal from os import path +from typing import Optional import yaml +from pydantic import AliasChoices, BaseModel, Field, StrictStr from kubernetes import client, utils +from kubernetes.client.models.v1_object_meta import V1ObjectMeta from kubernetes.client.rest import ApiException from kubernetes.e2e_test import base from kubernetes.utils import quantity +from kubernetes.utils.duration import V1Duration, format_duration, parse_duration class TestUtils(unittest.TestCase): @@ -811,3 +816,148 @@ def test_format_quantity(self): quantity.format_quantity(Decimal("0.5006"), "m", quantize=Decimal(0)), "501m", ) + + +class TestDuration(unittest.TestCase): + + def test_parse_duration(self): + # Valid durations + self.assertEqual(parse_duration("0h"), datetime.timedelta(hours=0)) + self.assertEqual(parse_duration("0s"), datetime.timedelta(hours=0)) + self.assertEqual(parse_duration("0h0m0s"), datetime.timedelta(hours=0)) + self.assertEqual(parse_duration("1h"), datetime.timedelta(hours=1)) + self.assertEqual(parse_duration("30m"), datetime.timedelta(minutes=30)) + self.assertEqual(parse_duration("10s"), datetime.timedelta(seconds=10)) + self.assertEqual(parse_duration("500ms"), datetime.timedelta(milliseconds=500)) + self.assertEqual(parse_duration("2h30m"), datetime.timedelta(hours=2, minutes=30)) + self.assertEqual(parse_duration("150m"), datetime.timedelta(hours=2, minutes=30)) + self.assertEqual(parse_duration("7230s"), datetime.timedelta(hours=2, seconds=30)) + self.assertEqual(parse_duration("1h30m10s"), datetime.timedelta(hours=1, minutes=30, seconds=10)) + self.assertEqual(parse_duration("10s30m1h"), datetime.timedelta(hours=1, minutes=30, seconds=10)) + self.assertEqual(parse_duration("100ms200ms300ms"), datetime.timedelta(milliseconds=600)) + + # Invalid durations + with self.assertRaises(ValueError): + parse_duration("1d") # Invalid unit 'd' + with self.assertRaises(ValueError): + parse_duration("1") # Missing unit + with self.assertRaises(ValueError): + parse_duration("1m1") # Missing unit + with self.assertRaises(ValueError): + parse_duration("1h30m10s20ms50h") # Too many units + with self.assertRaises(ValueError): + parse_duration("999999h") # Too many digits + with self.assertRaises(ValueError): + parse_duration("1.5h") # Floating point is not supported + with self.assertRaises(ValueError): + parse_duration("-15m") # Negative durations are not supported + + def test_format_duration(self): + # Valid durations + self.assertEqual(format_duration(datetime.timedelta(0)), "0s") + self.assertEqual(format_duration(datetime.timedelta(hours=1)), "1h") + self.assertEqual(format_duration(datetime.timedelta(minutes=30)), "30m") + self.assertEqual(format_duration(datetime.timedelta(seconds=10)), "10s") + self.assertEqual(format_duration(datetime.timedelta(milliseconds=500)), "500ms") + self.assertEqual(format_duration(datetime.timedelta(hours=2, minutes=30)), "2h30m") + self.assertEqual(format_duration(datetime.timedelta(hours=1, minutes=30, seconds=10)), "1h30m10s") + self.assertEqual(format_duration(datetime.timedelta(milliseconds=600)), "600ms") + self.assertEqual(format_duration(datetime.timedelta(hours=2, milliseconds=600)), "2h600ms") + self.assertEqual(format_duration(datetime.timedelta(hours=2, minutes=30, milliseconds=600)), "2h30m600ms") + self.assertEqual(format_duration(datetime.timedelta(hours=2, minutes=30, seconds=10, milliseconds=600)), "2h30m10s600ms") + self.assertEqual(format_duration(datetime.timedelta(minutes=0.5)), "30s") + self.assertEqual(format_duration(datetime.timedelta(seconds=0.5)), "500ms") + self.assertEqual(format_duration(datetime.timedelta(days=10)), "240h") # 10 days = 240 hours + + # Invalid durations + with self.assertRaises(ValueError): + format_duration(datetime.timedelta(microseconds=100)) # Sub-millisecond precision + with self.assertRaises(ValueError): + format_duration(datetime.timedelta(milliseconds=0.5)) # Sub-millisecond precision + with self.assertRaises(ValueError): + format_duration(datetime.timedelta(days=10000)) # Out of range (more than 99999 hours) + with self.assertRaises(ValueError): + format_duration(datetime.timedelta(minutes=-15)) # Negative durations are not supported + + +class DurationSpec(BaseModel): + time_to_live: V1Duration = Field(validation_alias=AliasChoices("timeToLive", "time_to_live"), serialization_alias="timeToLive") + + +class ObjectWithDuration(BaseModel): + api_version: Optional[StrictStr] = Field(default=None, validation_alias=AliasChoices("apiVersion", "api_version"), serialization_alias="apiVersion") + kind: Optional[StrictStr] = None + metadata: Optional[V1ObjectMeta] = None + spec: Optional[DurationSpec] = None + + +class TestDurationType(unittest.TestCase): + def test_model_dump(self): + test_object = ObjectWithDuration( + metadata=V1ObjectMeta( + name="duration-test", + namespace="test", + ), + spec=DurationSpec( + time_to_live=datetime.timedelta(hours=3, minutes=5, seconds=2) + ) + ) + self.assertEqual( + test_object.model_dump(exclude_none=True, exclude_defaults=True, by_alias=True), + { + "metadata": { + "name": "duration-test", + "namespace": "test", + }, + "spec": { + "timeToLive": "3h5m2s", + }, + }, + ) + + def test_model_dump_json(self): + test_object = ObjectWithDuration( + metadata=V1ObjectMeta( + name="duration-test", + namespace="test", + ), + spec=DurationSpec( + time_to_live=datetime.timedelta(hours=3, minutes=5, seconds=2) + ) + ) + self.assertEqual(r'{"metadata":{"name":"duration-test","namespace":"test"},"spec":{"timeToLive":"3h5m2s"}}', + test_object.model_dump_json(exclude_none=True, exclude_defaults=True, by_alias=True) + ) + + def test_model_validate(self): + duration_object = ObjectWithDuration.model_validate({ + "metadata": { + "name": "duration-test", + "namespace": "test", + }, + "spec": { + "timeToLive": "3h5m2s", + }, + }, + ) + self.assertEqual(duration_object.metadata.name, "duration-test") + self.assertEqual(duration_object.metadata.namespace, "test") + self.assertEqual(duration_object.spec.time_to_live, datetime.timedelta(hours=3, minutes=5, seconds=2)) + + def test_model_validate_json(self): + duration_object = ObjectWithDuration.model_validate_json( + r'{"metadata":{"name":"duration-test","namespace":"test"},"spec":{"timeToLive":"3h5m2s"}}' + ) + self.assertEqual(duration_object.metadata.name, "duration-test") + self.assertEqual(duration_object.metadata.namespace, "test") + self.assertEqual(duration_object.spec.time_to_live, datetime.timedelta(hours=3, minutes=5, seconds=2)) + + def test_timedelta_input_is_accepted(self): + spec = DurationSpec(time_to_live=datetime.timedelta(minutes=5)) + self.assertEqual(spec.time_to_live, datetime.timedelta(minutes=5)) + + def test_invalid_object_raises_value_error(self): + with self.assertRaises(expected_exception=ValueError): + ObjectWithDuration.model_validate_json( + r'{"metadata":{"name":"duration-test","namespace":"test"},"spec":{"timeToLive":"3h5m2s-Tuesday"}}' + ) \ No newline at end of file diff --git a/kubernetes/utils/__init__.py b/kubernetes/utils/__init__.py index b691d9de92..9b7dc40366 100644 --- a/kubernetes/utils/__init__.py +++ b/kubernetes/utils/__init__.py @@ -16,7 +16,7 @@ from .create_from_yaml import (FailToCreateError, create_from_dict, create_from_yaml, create_from_directory) from .quantity import parse_quantity -from .duration import parse_duration +from .duration import parse_duration, format_duration from .metrics import (get_nodes_metrics, get_pods_metrics, get_pods_metrics_in_all_namespaces) from .retry import (Backoff, DEFAULT_BACKOFF, DEFAULT_RETRY, diff --git a/kubernetes/utils/duration.py b/kubernetes/utils/duration.py index 9b1db74a82..b446ab6353 100644 --- a/kubernetes/utils/duration.py +++ b/kubernetes/utils/duration.py @@ -11,12 +11,12 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -from typing import List - import datetime import re +from typing import Annotated, List, TypeAlias, Union import durationpy +from pydantic import BeforeValidator, PlainSerializer # Initialize our RE statically, rather than compiling for every call. This has # the downside that it'll get compiled at import time but that shouldn't @@ -28,7 +28,7 @@ maxDuration_ms = (((99999 * 3600) + (59 * 60) + 59) * 1_000) + 999 -def parse_duration(duration) -> datetime.timedelta: +def parse_duration(duration: str) -> datetime.timedelta: """ Parse GEP-2257 Duration format to a datetime.timedelta object. @@ -132,11 +132,10 @@ def format_duration(delta: datetime.timedelta) -> str: # Check range early. if delta < datetime.timedelta(0): - raise ValueError("Cannot express negative durations in GEP-2257: {}".format(delta)) + raise ValueError(f"Cannot express negative durations in GEP-2257: {delta}") if delta > datetime.timedelta(milliseconds=maxDuration_ms): - raise ValueError( - "Cannot express durations longer than 99999h59m59s999ms in GEP-2257: {}".format(delta)) + raise ValueError(f"Cannot express durations longer than 99999h59m59s999ms in GEP-2257: {delta}") # durationpy.to_str() is happy to use floating-point seconds, which # GEP-2257 is _not_ happy with. So start by peeling off any microseconds @@ -144,10 +143,7 @@ def format_duration(delta: datetime.timedelta) -> str: delta_us = delta.microseconds if (delta_us % 1000) != 0: - raise ValueError( - "Cannot express sub-millisecond precision in GEP-2257: {}" - .format(delta) - ) + raise ValueError(f"Cannot express sub-millisecond precision in GEP-2257: {delta}") # After that, do the usual div & mod tree to take seconds and get hours, # minutes, and seconds from it. @@ -172,3 +168,31 @@ def format_duration(delta: datetime.timedelta) -> str: output.append(f"{delta_us // 1000}ms") return "".join(output) + + +def validate_duration(value: Union[str, datetime.timedelta]) -> datetime.timedelta: + # Only GEP-2257 strings need parsing; timedelta values pass through. + if isinstance(value, str): + return parse_duration(value) + return value + + +V1Duration: TypeAlias = Annotated[ + datetime.timedelta, + BeforeValidator(func=validate_duration, json_schema_input_type="str"), + PlainSerializer(func=format_duration, return_type=str, when_used="unless-none"), +] +""" +Duration validator and serializer that can handle GEP-2257 Duration. + +Example: + class DurationSpec(BaseModel): + time_to_live: V1Duration = Field( + validation_alias=AliasChoices("timeToLive", "time_to_live"), + serialization_alias="timeToLive", + ) + + DurationSpec(time_to_live=datetime.timedelta(hours=1, minutes=5)) + DurationSpec.model_validate({"timeToLive": "2h3m7s"}) + DurationSpec.model_validate_json(r'{"timeToLive": "37s"}') +"""