feat: base CloudEvent class as per v1 specs, including attribute validation

src/cloudevents/core/v1/__init__.py

@@ -0,0 +1,3 @@
+"""
+CloudEvent implementation for v1.0
+"""

src/cloudevents/core/v1/event.py

@@ -0,0 +1,140 @@
+#  Copyright 2018-Present The CloudEvents Authors
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, 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 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) -> None:
+        """
+        Create a new CloudEvent instance.
+
+        :param attributes: The attributes of the CloudEvent instance.
+        :type attributes: dict
+        :param data: The payload of the CloudEvent instance.
+        :type data: Optional[dict]
+
+        :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._attributes = attributes
+        self._data = data
+
+    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
+        ]
+        if missing_attributes:
+            raise ValueError(
+                f"Missing required attribute(s): {', '.join(missing_attributes)}"
+            )
+
+        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")
+
+        if not isinstance(attributes["source"], str):
+            raise TypeError("Attribute 'source' must be a string")
+
+        if not isinstance(attributes["type"], str):
+            raise TypeError("Attribute 'type' must be a string")
+
+        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'")
+
+        if "time" in attributes:
+            if not isinstance(attributes["time"], datetime):
+                raise TypeError("Attribute 'time' must be a datetime object")
+
+            if not attributes["time"].tzinfo:
+                raise ValueError("Attribute 'time' must be timezone aware")
+
+        if "subject" in attributes:
+            if not isinstance(attributes["subject"], str):
+                raise TypeError("Attribute 'subject' must be a string")
+
+            if not attributes["subject"]:
+                raise ValueError("Attribute 'subject' must not be empty")
+
+        if "datacontenttype" in attributes:
+            if not isinstance(attributes["datacontenttype"], str):
+                raise TypeError("Attribute 'datacontenttype' must be a string")
+
+            if not attributes["datacontenttype"]:
+                raise ValueError("Attribute 'datacontenttype' must not be empty")
+
+        if "dataschema" in attributes:
+            if not isinstance(attributes["dataschema"], str):
+                raise TypeError("Attribute 'dataschema' must be a string")
+
+            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]
+        """
+        return self._attributes[attribute]
+
+    def get_data(self) -> Optional[dict]:
+        """
+        Retrieve data of the event.
+
+        :return: The data of the event.
+        :rtype: Optional[dict]
+        """
+        return self._data

tests/test_core/test_v1/test_event.py

@@ -0,0 +1,196 @@
+from cloudevents.core.v1.event import CloudEvent
+
+import pytest
+from datetime import datetime
+from typing import Any, Optional
+
+
+@pytest.mark.parametrize(
+    "attributes, missing_attribute",
+    [
+        ({"source": "/", "type": "test", "specversion": "1.0"}, "id"),
+        ({"id": "1", "type": "test", "specversion": "1.0"}, "source"),
+        ({"id": "1", "source": "/", "specversion": "1.0"}, "type"),
+        ({"id": "1", "source": "/", "type": "test"}, "specversion"),
+    ],
+)
+def test_missing_required_attribute(attributes: dict, missing_attribute: str) -> None:
+    with pytest.raises(ValueError) as e:
+        CloudEvent(attributes)
+
+    assert str(e.value) == f"Missing required attribute(s): {missing_attribute}"
+
+
+@pytest.mark.parametrize(
+    "id,error",
+    [
+        (None, "Attribute 'id' must not be None"),
+        (12, "Attribute 'id' must be a string"),
+    ],
+)
+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: Any, error: str) -> None:
+    with pytest.raises((ValueError, TypeError)) as e:
+        CloudEvent({"id": "1", "source": source, "type": "test", "specversion": "1.0"})
+
+    assert str(e.value) == error
+
+
+@pytest.mark.parametrize(
+    "specversion,error",
+    [
+        (1.0, "Attribute 'specversion' must be a string"),
+        ("1.4", "Attribute 'specversion' must be '1.0'"),
+    ],
+)
+def test_specversion_validation(specversion: Any, error: str) -> None:
+    with pytest.raises((ValueError, TypeError)) as e:
+        CloudEvent(
+            {"id": "1", "source": "/", "type": "test", "specversion": specversion}
+        )
+
+    assert str(e.value) == error
+
+
+@pytest.mark.parametrize(
+    "time,error",
+    [
+        ("2023-10-25T17:09:19.736166Z", "Attribute 'time' must be a datetime object"),
+        (
+            datetime(2023, 10, 25, 17, 9, 19, 736166),
+            "Attribute 'time' must be timezone aware",
+        ),
+    ],
+)
+def test_time_validation(time: Any, error: str) -> None:
+    with pytest.raises((ValueError, TypeError)) as e:
+        CloudEvent(
+            {
+                "id": "1",
+                "source": "/",
+                "type": "test",
+                "specversion": "1.0",
+                "time": time,
+            }
+        )
+
+    assert str(e.value) == error
+
+
+@pytest.mark.parametrize(
+    "subject,error",
+    [
+        (1234, "Attribute 'subject' must be a string"),
+        (
+            "",
+            "Attribute 'subject' must not be empty",
+        ),
+    ],
+)
+def test_subject_validation(subject: Any, error: str) -> None:
+    with pytest.raises((ValueError, TypeError)) as e:
+        CloudEvent(
+            {
+                "id": "1",
+                "source": "/",
+                "type": "test",
+                "specversion": "1.0",
+                "subject": subject,
+            }
+        )
+
+    assert str(e.value) == error
+
+
+@pytest.mark.parametrize(
+    "datacontenttype,error",
+    [
+        (1234, "Attribute 'datacontenttype' must be a string"),
+        (
+            "",
+            "Attribute 'datacontenttype' must not be empty",
+        ),
+    ],
+)
+def test_datacontenttype_validation(datacontenttype: Any, error: str) -> None:
+    with pytest.raises((ValueError, TypeError)) as e:
+        CloudEvent(
+            {
+                "id": "1",
+                "source": "/",
+                "type": "test",
+                "specversion": "1.0",
+                "datacontenttype": datacontenttype,
+            }
+        )
+
+    assert str(e.value) == error
+
+
+@pytest.mark.parametrize(
+    "dataschema,error",
+    [
+        (1234, "Attribute 'dataschema' must be a string"),
+        (
+            "",
+            "Attribute 'dataschema' must not be empty",
+        ),
+    ],
+)
+def test_dataschema_validation(dataschema: Any, error: str) -> None:
+    with pytest.raises((ValueError, TypeError)) as e:
+        CloudEvent(
+            {
+                "id": "1",
+                "source": "/",
+                "type": "test",
+                "specversion": "1.0",
+                "dataschema": dataschema,
+            }
+        )
+
+    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