Story Songs That Made History

@@ -77,6 +77,22 @@ 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]:
+ """
+ Retrieves an event attribute value for the given key.
+ Returns the default value if not attribute for the given key exists.
+
+ MUST NOT throw an exception when the key does not exist.
+
+ :param key: The event attribute name.
+ :param default: The default value to be returned when
+ no attribute with the given key exists.
+ :returns: The event attribute value if exists, default value otherwise.
+ """
+ return self._attributes.get(key, default)
+
  def __setitem__(self, key, value):
  self._attributes[key] = value
 

Last Post Backdrop

@@ -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,49 @@ 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_must_return_attribute_value_if_exists(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
+ )