feat: event attribute get operation support

cloudevents/http/event.py

@@ -77,6 +77,18 @@ 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
+        :returns: attribute value if exists, default otherwise
+        """
+        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,55 @@ def test_http_cloudevent_equality(specversion):
         "datacontenttype": "application/json",
         "subject": "my-subject",
     }
-    data = '{"name":"john"}'
-    event1 = CloudEvent(attributes, data)
-    event2 = CloudEvent(attributes, data)
+
+
+@pytest.fixture()
+def my_dummy_data():
+    return '{"name":"john"}'
+
+
+@pytest.fixture()
+def your_dummy_data():
+    return '{"name":"paul"}'
+
+
+def test_http_cloudevent_equality(
+    dummy_attributes, my_dummy_data, your_dummy_data
+):
+    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, my_dummy_data, your_dummy_data
+):
+    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 +79,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 +131,51 @@ 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, my_dummy_data):
+    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
+    )