feat: base `CloudEvent` class as per v1 specs, including attribute validation
Signed-off-by: Tudor Plugaru <plugaru.tudor@protonmail.com>
This commit is contained in:
parent
9101ab470b
commit
a2ac76224b
|
@ -0,0 +1,71 @@
|
|||
from typing import Optional
|
||||
from datetime import datetime
|
||||
|
||||
REQUIRED_ATTRIBUTES = {"id", "source", "type", "specversion"}
|
||||
OPTIONAL_ATTRIBUTES = {"datacontenttype", "dataschema", "subject", "time"}
|
||||
|
||||
|
||||
class CloudEvent:
|
||||
def __init__(self, attributes: dict, data: Optional[dict] = None):
|
||||
self.__validate_attribute(attributes)
|
||||
self._attributes = attributes
|
||||
self._data = data
|
||||
|
||||
def __validate_attribute(self, attributes: dict):
|
||||
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")
|
||||
|
||||
def get_attribute(self, attribute: str):
|
||||
return self._attributes[attribute]
|
||||
|
||||
def get_data(self):
|
||||
return self._data
|
|
@ -0,0 +1,158 @@
|
|||
from cloudevents.core.v1.event import CloudEvent
|
||||
|
||||
import pytest
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
@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, missing_attribute):
|
||||
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, error):
|
||||
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, error):
|
||||
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, error):
|
||||
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, error):
|
||||
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, error):
|
||||
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, error):
|
||||
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, error):
|
||||
with pytest.raises((ValueError, TypeError)) as e:
|
||||
CloudEvent(
|
||||
{
|
||||
"id": "1",
|
||||
"source": "/",
|
||||
"type": "test",
|
||||
"specversion": "1.0",
|
||||
"dataschema": dataschema,
|
||||
}
|
||||
)
|
||||
|
||||
assert str(e.value) == error
|
Loading…
Reference in New Issue