chore: Add support for custom extension names and validate them

Signed-off-by: Tudor <plugaru.tudor@protonmail.com>

Author: PlugaruT

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 @@ def __init__(self, attributes: dict, data: Optional[dict] = None) -> '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 @@ def __validate_attribute(self, attributes: dict):
 
         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 @@ def __validate_attribute(self, attributes: dict):
 
         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 @@ def __validate_attribute(self, attributes: dict):
             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]
         """

tests/test_core/test_v1/test_event.py

@@ -2,6 +2,7 @@
 
 import pytest
 from datetime import datetime
+from typing import Any, Optional
 
 
 @pytest.mark.parametrize(
@@ -13,7 +14,7 @@
         ({"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,15 +28,15 @@ 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"})
 
     assert str(e.value) == error
 
 
 @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