chore: Add getters for attributes and test happy path

Signed-off-by: Tudor Plugaru <plugaru.tudor@protonmail.com>
This commit is contained in:
Tudor Plugaru 2024-11-09 22:02:23 +02:00
parent 42b4fe1d3a
commit f83c363cf5
No known key found for this signature in database
2 changed files with 97 additions and 7 deletions

View File

@ -127,15 +127,69 @@ class CloudEvent:
f"Extension attribute '{extension_attributes}' should only contain lowercase letters and numbers"
)
def get_attribute(self, attribute: str) -> Optional[Any]:
def get_id(self) -> str:
"""
Retrieve a value of an attribute of the event denoted by the given `attribute`.
Retrieve the ID of the event.
:param attribute: The name of the event attribute to retrieve the value for.
:return: The event attribute value.
:return: The ID of the event.
"""
return self._attributes[attribute]
return self._attributes["id"]
def get_source(self) -> str:
"""
Retrieve the source of the event.
:return: The source of the event.
"""
return self._attributes["source"]
def get_type(self) -> str:
"""
Retrieve the type of the event.
:return: The type of the event.
"""
return self._attributes["type"]
def get_specversion(self) -> str:
"""
Retrieve the specversion of the event.
:return: The specversion of the event.
"""
return self._attributes["specversion"]
def get_datacontenttype(self) -> Optional[str]:
"""
Retrieve the datacontenttype of the event.
:return: The datacontenttype of the event.
"""
return self._attributes.get("datacontenttype")
def get_dataschema(self) -> Optional[str]:
"""
Retrieve the dataschema of the event.
:return: The dataschema of the event.
"""
return self._attributes.get("dataschema")
def get_subject(self) -> Optional[str]:
"""
Retrieve the subject of the event.
:return: The subject of the event.
"""
return self._attributes.get("subject")
def get_time(self) -> Optional[datetime]:
"""
Retrieve the time of the event.
:return: The time of the event.
"""
return self._attributes.get("time")
def get_data(self) -> Optional[dict]:
"""

View File

@ -15,7 +15,7 @@
from cloudevents.core.v1.event import CloudEvent
import pytest
from datetime import datetime
from datetime import datetime, timezone
from typing import Any, Optional
@ -207,3 +207,39 @@ def test_custom_extension(extension_name: str, error: str) -> None:
)
assert str(e.value) == error
def test_cloud_event_constructor() -> None:
id = "1"
source = "/source"
type = "com.test.type"
specversion = "1.0"
datacontenttype = "application/json"
dataschema = "http://example.com/schema"
subject = "test_subject"
time = datetime.now(tz=timezone.utc)
data = {"key": "value"}
event = CloudEvent(
attributes={
"id": id,
"source": source,
"type": type,
"specversion": specversion,
"datacontenttype": datacontenttype,
"dataschema": dataschema,
"subject": subject,
"time": time,
},
data=data,
)
assert event.get_id() == id
assert event.get_source() == source
assert event.get_type() == type
assert event.get_specversion() == specversion
assert event.get_datacontenttype() == datacontenttype
assert event.get_dataschema() == dataschema
assert event.get_subject() == subject
assert event.get_time() == time
assert event.get_data() == data