Skip to content

chore: add default values for id, time and specversion #264

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
PlugaruT wants to merge 1 commit into
base: v2
Choose a base branch
from
+136 −30
Open

chore: add default values for id, time and specversion #264

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion src/cloudevents/core/v03/event.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,9 @@
# under the License.

import re
import uuid
from collections import defaultdict
from datetime import datetime
from datetime import datetime, timezone
from typing import Any, Final

from cloudevents.core.base import BaseCloudEvent
Expand Down Expand Up @@ -59,6 +60,13 @@ def __init__(

:raises CloudEventValidationError: If any of the required attributes are missing or have invalid values.
"""
if "specversion" not in attributes:
attributes["specversion"] = SPECVERSION_V0_3
if "id" not in attributes:
attributes["id"] = str(uuid.uuid4())
if "time" not in attributes:
attributes["time"] = datetime.now(timezone.utc)

self._validate_attribute(attributes=attributes)
self._attributes: dict[str, Any] = attributes
self._data: dict[str, Any] | str | bytes | None = data
Expand Down
10 changes: 9 additions & 1 deletion src/cloudevents/core/v1/event.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,9 @@
# under the License.

import re
import uuid
from collections import defaultdict
from datetime import datetime
from datetime import datetime, timezone
from typing import Any, Final

from cloudevents.core.base import BaseCloudEvent
Expand Down Expand Up @@ -43,6 +44,13 @@ def __init__(
attributes: dict[str, Any],
data: dict[str, Any] | str | bytes | None = None,
) -> None:
if "specversion" not in attributes:
attributes["specversion"] = SPECVERSION_V1_0
if "id" not in attributes:
attributes["id"] = str(uuid.uuid4())
if "time" not in attributes:
attributes["time"] = datetime.now(timezone.utc)

self._validate_attribute(attributes=attributes)
self._attributes: dict[str, Any] = attributes
self._data: dict[str, Any] | str | bytes | None = data
Expand Down
73 changes: 59 additions & 14 deletions tests/test_core/test_v03/test_event.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,6 @@ def test_missing_required_attributes() -> None:
CloudEvent({})

expected_errors = {
"id": [
str(MissingRequiredAttributeError("id")),
str(InvalidAttributeValueError("id", "Attribute 'id' must not be None")),
str(InvalidAttributeTypeError("id", str)),
],
"source": [
str(MissingRequiredAttributeError("source")),
str(InvalidAttributeTypeError("source", str)),
Expand All @@ -45,15 +40,6 @@ def test_missing_required_attributes() -> None:
str(MissingRequiredAttributeError("type")),
str(InvalidAttributeTypeError("type", str)),
],
"specversion": [
str(MissingRequiredAttributeError("specversion")),
str(InvalidAttributeTypeError("specversion", str)),
str(
InvalidAttributeValueError(
"specversion", "Attribute 'specversion' must be '0.3'"
)
),
],
}

actual_errors = {
Expand Down Expand Up @@ -358,6 +344,65 @@ def test_custom_extension(extension_name: str, expected_error: dict) -> None:
assert actual_errors == expected_error


def test_default_specversion() -> None:
event = CloudEvent(
attributes={"source": "/source", "type": "test", "id": "1"},
)
assert event.get_specversion() == "0.3"


def test_default_id() -> None:
event = CloudEvent(
attributes={"source": "/source", "type": "test", "specversion": "0.3"},
)
assert isinstance(event.get_id(), str)
assert len(event.get_id()) == 36 # UUID4 format


def test_default_id_is_unique() -> None:
event1 = CloudEvent(attributes={"source": "/s", "type": "t"})
event2 = CloudEvent(attributes={"source": "/s", "type": "t"})
assert event1.get_id() != event2.get_id()


def test_default_time() -> None:
before = datetime.now(tz=timezone.utc)
event = CloudEvent(
attributes={"source": "/source", "type": "test", "specversion": "0.3"},
)
after = datetime.now(tz=timezone.utc)
assert event.get_time() is not None
assert before <= event.get_time() <= after
assert event.get_time().tzinfo is not None


def test_explicit_values_override_defaults() -> None:
custom_time = datetime(2024, 6, 15, 12, 0, 0, tzinfo=timezone.utc)
event = CloudEvent(
attributes={
"source": "/source",
"type": "test",
"specversion": "0.3",
"id": "my-custom-id",
"time": custom_time,
},
)
assert event.get_id() == "my-custom-id"
assert event.get_time() == custom_time
assert event.get_specversion() == "0.3"


def test_minimal_event_with_defaults() -> None:
event = CloudEvent(
attributes={"source": "/source", "type": "test"},
)
assert event.get_source() == "/source"
assert event.get_type() == "test"
assert event.get_specversion() == "0.3"
assert event.get_id() is not None
assert event.get_time() is not None


def test_cloud_event_v03_constructor() -> None:
"""Test creating a v0.3 CloudEvent with all attributes"""
id = "1"
Expand Down
73 changes: 59 additions & 14 deletions tests/test_core/test_v1/test_event.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,6 @@ def test_missing_required_attributes() -> None:
CloudEvent({})

expected_errors = {
"id": [
str(MissingRequiredAttributeError("id")),
str(InvalidAttributeValueError("id", "Attribute 'id' must not be None")),
str(InvalidAttributeTypeError("id", str)),
],
"source": [
str(MissingRequiredAttributeError("source")),
str(InvalidAttributeTypeError("source", str)),
Expand All @@ -45,15 +40,6 @@ def test_missing_required_attributes() -> None:
str(MissingRequiredAttributeError("type")),
str(InvalidAttributeTypeError("type", str)),
],
"specversion": [
str(MissingRequiredAttributeError("specversion")),
str(InvalidAttributeTypeError("specversion", str)),
str(
InvalidAttributeValueError(
"specversion", "Attribute 'specversion' must be '1.0'"
)
),
],
}

actual_errors = {
Expand Down Expand Up @@ -294,6 +280,65 @@ def test_custom_extension(extension_name: str, expected_error: dict) -> None:
assert actual_errors == expected_error


def test_default_specversion() -> None:
event = CloudEvent(
attributes={"source": "/source", "type": "test", "id": "1"},
)
assert event.get_specversion() == "1.0"


def test_default_id() -> None:
event = CloudEvent(
attributes={"source": "/source", "type": "test", "specversion": "1.0"},
)
assert isinstance(event.get_id(), str)
assert len(event.get_id()) == 36 # UUID4 format


def test_default_id_is_unique() -> None:
event1 = CloudEvent(attributes={"source": "/s", "type": "t"})
event2 = CloudEvent(attributes={"source": "/s", "type": "t"})
assert event1.get_id() != event2.get_id()


def test_default_time() -> None:
before = datetime.now(tz=timezone.utc)
event = CloudEvent(
attributes={"source": "/source", "type": "test", "specversion": "1.0"},
)
after = datetime.now(tz=timezone.utc)
assert event.get_time() is not None
assert before <= event.get_time() <= after
assert event.get_time().tzinfo is not None


def test_explicit_values_override_defaults() -> None:
custom_time = datetime(2024, 6, 15, 12, 0, 0, tzinfo=timezone.utc)
event = CloudEvent(
attributes={
"source": "/source",
"type": "test",
"specversion": "1.0",
"id": "my-custom-id",
"time": custom_time,
},
)
assert event.get_id() == "my-custom-id"
assert event.get_time() == custom_time
assert event.get_specversion() == "1.0"


def test_minimal_event_with_defaults() -> None:
event = CloudEvent(
attributes={"source": "/source", "type": "test"},
)
assert event.get_source() == "/source"
assert event.get_type() == "test"
assert event.get_specversion() == "1.0"
assert event.get_id() is not None
assert event.get_time() is not None


def test_cloud_event_constructor() -> None:
id = "1"
source = "/source"
Expand Down