feat: event attribute get operation support

cloudevents/http/event.py

@@ -27,7 +27,7 @@ class CloudEvent:
     """
 
     def __init__(
-        self, attributes: typing.Dict[str, str], data: typing.Any = None
+            self, attributes: typing.Dict[str, str], data: typing.Any = None
     ):
         """
         Event Constructor
@@ -77,6 +77,15 @@ def __eq__(self, other):
     def __getitem__(self, key):
         return self._attributes[key]
 
+    def get(self, key: str, default: typing.Optional[typing.Any] = None) -> typing.Optional[typing.Any]:
+        """
+        Get a an attribute value, return default if does not exist.
+        MUST NOT throw exception when the key does not exist
+        :param key: attribute name
+        :param default: value which will be returned if the attribute does not exist
+        """
+        return self._attributes.get(key, default)
+
     def __setitem__(self, key, value):
         self._attributes[key] = value
 

cloudevents/tests/test_http_cloudevent.py

@@ -5,9 +5,14 @@
 from cloudevents.http.util import _json_or_string
 
 
-@pytest.mark.parametrize("specversion", ["0.3", "1.0"])
-def test_http_cloudevent_equality(specversion):
-    attributes = {
+@pytest.fixture(params=["0.3", "1.0"])
+def specversion(request):
+    return request.param
+
+
+@pytest.fixture()
+def dummy_attributes(specversion):
+    return {
         "source": "<source>",
         "specversion": specversion,
         "id": "my-id",
@@ -16,48 +21,46 @@ def test_http_cloudevent_equality(specversion):
         "datacontenttype": "application/json",
         "subject": "my-subject",
     }
-    data = '{"name":"john"}'
-    event1 = CloudEvent(attributes, data)
-    event2 = CloudEvent(attributes, data)
+
+
+_my_dummy_data = '{"name":"john"}'
+_your_dummy_data = '{"name":"paul"}'
+assert _my_dummy_data != _your_dummy_data
+
+
+def test_http_cloudevent_equality(dummy_attributes):
+    data = _my_dummy_data
+    event1 = CloudEvent(dummy_attributes, data)
+    event2 = CloudEvent(dummy_attributes, data)
     assert event1 == event2
     # Test different attributes
-    for key in attributes:
+    for key in dummy_attributes:
         if key == "specversion":
             continue
         else:
-            attributes[key] = f"noise-{key}"
-        event3 = CloudEvent(attributes, data)
-        event2 = CloudEvent(attributes, data)
+            dummy_attributes[key] = f"noise-{key}"
+        event3 = CloudEvent(dummy_attributes, data)
+        event2 = CloudEvent(dummy_attributes, data)
         assert event2 == event3
         assert event1 != event2 and event3 != event1
 
     # Test different data
-    data = '{"name":"paul"}'
-    event3 = CloudEvent(attributes, data)
-    event2 = CloudEvent(attributes, data)
+    data = _your_dummy_data
+    event3 = CloudEvent(dummy_attributes, data)
+    event2 = CloudEvent(dummy_attributes, data)
     assert event2 == event3
     assert event1 != event2 and event3 != event1
 
 
-@pytest.mark.parametrize("specversion", ["0.3", "1.0"])
-def test_http_cloudevent_mutates_equality(specversion):
-    attributes = {
-        "source": "<source>",
-        "specversion": specversion,
-        "id": "my-id",
-        "time": "tomorrow",
-        "type": "tests.cloudevents.override",
-        "datacontenttype": "application/json",
-        "subject": "my-subject",
-    }
-    data = '{"name":"john"}'
-    event1 = CloudEvent(attributes, data)
-    event2 = CloudEvent(attributes, data)
-    event3 = CloudEvent(attributes, data)
+def test_http_cloudevent_mutates_equality(dummy_attributes):
+    data = _my_dummy_data
+    event1 = CloudEvent(dummy_attributes, data)
+    event2 = CloudEvent(dummy_attributes, data)
+    event3 = CloudEvent(dummy_attributes, data)
 
     assert event1 == event2
     # Test different attributes
-    for key in attributes:
+    for key in dummy_attributes:
         if key == "specversion":
             continue
         else:
@@ -67,8 +70,8 @@ def test_http_cloudevent_mutates_equality(specversion):
         assert event1 != event2 and event3 != event1
 
     # Test different data
-    event2.data = '{"name":"paul"}'
-    event3.data = '{"name":"paul"}'
+    event2.data = _your_dummy_data
+    event3.data = _your_dummy_data
     assert event2 == event3
     assert event1 != event2 and event3 != event1
 
@@ -119,3 +122,39 @@ def test_cloudevent_general_overrides():
 
 def test_none_json_or_string():
     assert _json_or_string(None) is None
+
+
+@pytest.fixture()
+def dummy_event(dummy_attributes):
+    return CloudEvent(attributes=dummy_attributes, data=_my_dummy_data)
+
+
+@pytest.fixture()
+def non_exiting_attribute_name(dummy_event):
+    result = "nonexisting"
+    assert result not in dummy_event
+    return result
+
+
+def test_get_operation_on_non_existing_attribute_must_not_raise_exception(dummy_event, non_exiting_attribute_name):
+    dummy_event.get(non_exiting_attribute_name)
+
+
+def test_get_operation_must_return_attribute_value_of_an_existing_attribute_matching_the_given_key(dummy_event):
+    assert dummy_event.get("source") == dummy_event["source"]
+
+
+def test_get_operation_on_non_existing_attribute_must_return_none_by_default(dummy_event, non_exiting_attribute_name):
+    assert dummy_event.get(non_exiting_attribute_name) is None
+
+
+def test_get_operation_on_non_existing_attribute_must_return_default_value_if_given(dummy_event,
+                                                                                    non_exiting_attribute_name):
+    dummy_value = "Hello World"
+    assert dummy_event.get(non_exiting_attribute_name, dummy_value) == dummy_value
+
+
+def test_get_operation_on_non_existing_attribute_should_not_copy_default_value(dummy_event,
+                                                                               non_exiting_attribute_name):
+    dummy_value = object()
+    assert dummy_event.get(non_exiting_attribute_name, dummy_value) is dummy_value