From 35dee7d2c1a620529f4f4c2969c83f2b73ce5ef5 Mon Sep 17 00:00:00 2001 From: Tudor Date: Sat, 9 Nov 2024 19:43:24 +0200 Subject: [PATCH] chore: Add support for custom extension names and validate them Signed-off-by: Tudor --- src/cloudevents/core/v1/event.py | 39 ++++++++++++++++--- tests/test_core/test_v1/test_event.py | 54 +++++++++++++++++++++++---- 2 files changed, 80 insertions(+), 13 deletions(-) diff --git a/src/cloudevents/core/v1/event.py b/src/cloudevents/core/v1/event.py index 9ae4895..eea6010 100644 --- a/src/cloudevents/core/v1/event.py +++ b/src/cloudevents/core/v1/event.py @@ -14,13 +14,14 @@ from typing import Any, Optional from datetime import datetime +import re REQUIRED_ATTRIBUTES = {"id", "source", "type", "specversion"} OPTIONAL_ATTRIBUTES = {"datacontenttype", "dataschema", "subject", "time"} class CloudEvent: - def __init__(self, attributes: dict, data: Optional[dict] = None) -> 'CloudEvent': + def __init__(self, attributes: dict, data: Optional[dict] = None) -> None: """ Create a new CloudEvent instance. @@ -32,11 +33,14 @@ class CloudEvent: :raises ValueError: If any of the required attributes are missing or have invalid values. :raises TypeError: If any of the attributes have invalid types. """ - self.__validate_attribute(attributes) + self._validate_attribute(attributes) self._attributes = attributes self._data = data - def __validate_attribute(self, attributes: dict): + def _validate_attribute(self, attributes: dict) -> None: + """ + Private method that validates the attributes of the CloudEvent as per the CloudEvents specification. + """ missing_attributes = [ attr for attr in REQUIRED_ATTRIBUTES if attr not in attributes ] @@ -47,6 +51,7 @@ class CloudEvent: if attributes["id"] is None: raise ValueError("Attribute 'id' must not be None") + if not isinstance(attributes["id"], str): raise TypeError("Attribute 'id' must be a string") @@ -58,6 +63,7 @@ class CloudEvent: if not isinstance(attributes["specversion"], str): raise TypeError("Attribute 'specversion' must be a string") + if attributes["specversion"] != "1.0": raise ValueError("Attribute 'specversion' must be '1.0'") @@ -89,13 +95,36 @@ class CloudEvent: if not attributes["dataschema"]: raise ValueError("Attribute 'dataschema' must not be empty") + for custom_extension in ( + set(attributes.keys()) - REQUIRED_ATTRIBUTES - OPTIONAL_ATTRIBUTES + ): + if custom_extension == "data": + raise ValueError( + "Extension attribute 'data' is reserved and must not be used" + ) + + if not custom_extension[0].isalpha(): + raise ValueError( + f"Extension attribute '{custom_extension}' should start with a letter" + ) + + if not (5 <= len(custom_extension) <= 20): + raise ValueError( + f"Extension attribute '{custom_extension}' should be between 5 and 20 characters long" + ) + + if not re.match(r"^[a-z0-9]+$", custom_extension): + raise ValueError( + f"Extension attribute '{custom_extension}' should only contain lowercase letters and numbers" + ) + def get_attribute(self, attribute: str) -> Optional[Any]: """ Retrieve a value of an attribute of the event denoted by the given `attribute`. - + :param attribute: The name of the event attribute to retrieve the value for. :type attribute: str - + :return: The event attribute value. :rtype: Optional[Any] """ diff --git a/tests/test_core/test_v1/test_event.py b/tests/test_core/test_v1/test_event.py index 11dd0b9..fcf541e 100644 --- a/tests/test_core/test_v1/test_event.py +++ b/tests/test_core/test_v1/test_event.py @@ -2,6 +2,7 @@ from cloudevents.core.v1.event import CloudEvent import pytest from datetime import datetime +from typing import Any, Optional @pytest.mark.parametrize( @@ -13,7 +14,7 @@ from datetime import datetime ({"id": "1", "source": "/", "type": "test"}, "specversion"), ], ) -def test_missing_required_attribute(attributes, missing_attribute) -> None: +def test_missing_required_attribute(attributes: dict, missing_attribute: str) -> None: with pytest.raises(ValueError) as e: CloudEvent(attributes) @@ -27,7 +28,7 @@ def test_missing_required_attribute(attributes, missing_attribute) -> None: (12, "Attribute 'id' must be a string"), ], ) -def test_id_validation(id, error) -> None: +def test_id_validation(id: Optional[Any], error: str) -> None: with pytest.raises((ValueError, TypeError)) as e: CloudEvent({"id": id, "source": "/", "type": "test", "specversion": "1.0"}) @@ -35,7 +36,7 @@ def test_id_validation(id, error) -> None: @pytest.mark.parametrize("source,error", [(123, "Attribute 'source' must be a string")]) -def test_source_validation(source, error) -> None: +def test_source_validation(source: Any, error: str) -> None: with pytest.raises((ValueError, TypeError)) as e: CloudEvent({"id": "1", "source": source, "type": "test", "specversion": "1.0"}) @@ -49,7 +50,7 @@ def test_source_validation(source, error) -> None: ("1.4", "Attribute 'specversion' must be '1.0'"), ], ) -def test_specversion_validation(specversion, error) -> None: +def test_specversion_validation(specversion: Any, error: str) -> None: with pytest.raises((ValueError, TypeError)) as e: CloudEvent( {"id": "1", "source": "/", "type": "test", "specversion": specversion} @@ -68,7 +69,7 @@ def test_specversion_validation(specversion, error) -> None: ), ], ) -def test_time_validation(time, error) -> None: +def test_time_validation(time: Any, error: str) -> None: with pytest.raises((ValueError, TypeError)) as e: CloudEvent( { @@ -93,7 +94,7 @@ def test_time_validation(time, error) -> None: ), ], ) -def test_subject_validation(subject, error) -> None: +def test_subject_validation(subject: Any, error: str) -> None: with pytest.raises((ValueError, TypeError)) as e: CloudEvent( { @@ -118,7 +119,7 @@ def test_subject_validation(subject, error) -> None: ), ], ) -def test_datacontenttype_validation(datacontenttype, error) -> None: +def test_datacontenttype_validation(datacontenttype: Any, error: str) -> None: with pytest.raises((ValueError, TypeError)) as e: CloudEvent( { @@ -143,7 +144,7 @@ def test_datacontenttype_validation(datacontenttype, error) -> None: ), ], ) -def test_dataschema_validation(dataschema, error) -> None: +def test_dataschema_validation(dataschema: Any, error: str) -> None: with pytest.raises((ValueError, TypeError)) as e: CloudEvent( { @@ -156,3 +157,40 @@ def test_dataschema_validation(dataschema, error) -> None: ) assert str(e.value) == error + + +@pytest.mark.parametrize( + "extension_name,error", + [ + ("123", "Extension attribute '123' should start with a letter"), + ( + "shrt", + "Extension attribute 'shrt' should be between 5 and 20 characters long", + ), + ( + "thisisaverylongextension", + "Extension attribute 'thisisaverylongextension' should be between 5 and 20 characters long", + ), + ( + "ThisIsNotValid", + "Extension attribute 'ThisIsNotValid' should only contain lowercase letters and numbers", + ), + ( + "data", + "Extension attribute 'data' is reserved and must not be used", + ), + ], +) +def test_custom_extension(extension_name: str, error: str) -> None: + with pytest.raises(ValueError) as e: + CloudEvent( + { + "id": "1", + "source": "/", + "type": "test", + "specversion": "1.0", + extension_name: "value", + } + ) + + assert str(e.value) == error