cloudevents
/
sdk-python
mirror of https://github.com/cloudevents/sdk-python.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

chore: Returns all the errors at ones instead of raising early. Improve tests 

Signed-off-by: Tudor Plugaru <plugaru.tudor@protonmail.com>
This commit is contained in:
Tudor Plugaru 2024-11-11 20:35:54 +02:00
parent b2023250a7
commit c5e6df9ec8
No known key found for this signature in database
3 changed files with 171 additions and 137 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

135
src/cloudevents/core/v1/event.py
View File

@ -12,17 +12,19 @@
# License for the specific language governing permissions and limitations
# under the License.
from typing import Any, Optional, Final
from datetime import datetime
import re
from datetime import datetime
from typing import Any, Final, Optional
REQUIRED_ATTRIBUTES: Final[set[str]] = {"id", "source", "type", "specversion"}
OPTIONAL_ATTRIBUTES: Final[set[str]] = {
from cloudevents.core.v1.exceptions import CloudEventValidationError
REQUIRED_ATTRIBUTES: Final[list[str]] = ["id", "source", "type", "specversion"]
OPTIONAL_ATTRIBUTES: Final[list[str]] = [
"datacontenttype",
"dataschema",
"subject",
"time",
}
]
class CloudEvent:
@ -55,102 +57,129 @@ class CloudEvent:
See https://github.com/cloudevents/spec/blob/main/cloudevents/spec.md#required-attributes
"""
CloudEvent._validate_required_attributes(attributes)
CloudEvent._validate_attribute_types(attributes)
CloudEvent._validate_optional_attributes(attributes)
CloudEvent._validate_extension_attributes(attributes)
errors = {}
errors.update(CloudEvent._validate_required_attributes(attributes))
errors.update(CloudEvent._validate_attribute_types(attributes))
errors.update(CloudEvent._validate_optional_attributes(attributes))
errors.update(CloudEvent._validate_extension_attributes(attributes))
if errors:
raise CloudEventValidationError(errors)
@staticmethod
def _validate_required_attributes(attributes: dict[str, Any]) -> None:
def _validate_required_attributes(
attributes: dict[str, Any],
) -> dict[str, list[str]]:
"""
Validates that all required attributes are present.
:param attributes: The attributes of the CloudEvent instance.
:raises ValueError: If any of the required attributes are missing.
:return: A dictionary of validation error messages.
"""
errors = {}
missing_attributes = [
attr for attr in REQUIRED_ATTRIBUTES if attr not in attributes
]
if missing_attributes:
raise ValueError(
errors["required"] = [
f"Missing required attribute(s): {', '.join(missing_attributes)}"
)
]
return errors
@staticmethod
def _validate_attribute_types(attributes: dict[str, Any]) -> None:
def _validate_attribute_types(attributes: dict[str, Any]) -> dict[str, list[str]]:
"""
Validates the types of the required attributes.
:param attributes: The attributes of the CloudEvent instance.
:raises ValueError: If any of the required attributes have invalid values.
:raises TypeError: If any of the required attributes have invalid types.
:return: A dictionary of validation error messages.
"""
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'")
errors = {}
type_errors = []
if attributes.get("id") is None:
type_errors.append("Attribute 'id' must not be None")
if not isinstance(attributes.get("id"), str):
type_errors.append("Attribute 'id' must be a string")
if not isinstance(attributes.get("source"), str):
type_errors.append("Attribute 'source' must be a string")
if not isinstance(attributes.get("type"), str):
type_errors.append("Attribute 'type' must be a string")
if not isinstance(attributes.get("specversion"), str):
type_errors.append("Attribute 'specversion' must be a string")
if attributes.get("specversion") != "1.0":
type_errors.append("Attribute 'specversion' must be '1.0'")
if type_errors:
errors["type"] = type_errors
return errors
@staticmethod
def _validate_optional_attributes(attributes: dict[str, Any]) -> None:
def _validate_optional_attributes(
attributes: dict[str, Any],
) -> dict[str, list[str]]:
"""
Validates the types and values of the optional attributes.
:param attributes: The attributes of the CloudEvent instance.
:raises ValueError: If any of the optional attributes have invalid values.
:raises TypeError: If any of the optional attributes have invalid types.
:return: A dictionary of validation error messages.
"""
errors = {}
optional_errors = []
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")
optional_errors.append("Attribute 'time' must be a datetime object")
if hasattr(attributes["time"], "tzinfo") and not attributes["time"].tzinfo:
optional_errors.append("Attribute 'time' must be timezone aware")
if "subject" in attributes:
if not isinstance(attributes["subject"], str):
raise TypeError("Attribute 'subject' must be a string")
optional_errors.append("Attribute 'subject' must be a string")
if not attributes["subject"]:
raise ValueError("Attribute 'subject' must not be empty")
optional_errors.append("Attribute 'subject' must not be empty")
if "datacontenttype" in attributes:
if not isinstance(attributes["datacontenttype"], str):
raise TypeError("Attribute 'datacontenttype' must be a string")
optional_errors.append("Attribute 'datacontenttype' must be a string")
if not attributes["datacontenttype"]:
raise ValueError("Attribute 'datacontenttype' must not be empty")
optional_errors.append("Attribute 'datacontenttype' must not be empty")
if "dataschema" in attributes:
if not isinstance(attributes["dataschema"], str):
raise TypeError("Attribute 'dataschema' must be a string")
optional_errors.append("Attribute 'dataschema' must be a string")
if not attributes["dataschema"]:
raise ValueError("Attribute 'dataschema' must not be empty")
optional_errors.append("Attribute 'dataschema' must not be empty")
if optional_errors:
errors["optional"] = optional_errors
return errors
@staticmethod
def _validate_extension_attributes(attributes: dict[str, Any]) -> None:
def _validate_extension_attributes(
attributes: dict[str, Any],
) -> dict[str, list[str]]:
"""
Validates the extension attributes.
:param attributes: The attributes of the CloudEvent instance.
:raises ValueError: If any of the extension attributes have invalid values.
:return: A dictionary of validation error messages.
"""
for extension_attributes in (
set(attributes.keys()) - REQUIRED_ATTRIBUTES - OPTIONAL_ATTRIBUTES
):
if extension_attributes == "data":
raise ValueError(
errors = {}
extension_errors = []
extension_attributes = [
key
for key in attributes.keys()
if key not in REQUIRED_ATTRIBUTES and key not in OPTIONAL_ATTRIBUTES
]
for extension_attribute in extension_attributes:
if extension_attribute == "data":
extension_errors.append(
"Extension attribute 'data' is reserved and must not be used"
)
if not (1 <= len(extension_attributes) <= 20):
raise ValueError(
f"Extension attribute '{extension_attributes}' should be between 1 and 20 characters long"
if not (1 <= len(extension_attribute) <= 20):
extension_errors.append(
f"Extension attribute '{extension_attribute}' should be between 1 and 20 characters long"
)
if not re.match(r"^[a-z0-9]+$", extension_attributes):
raise ValueError(
f"Extension attribute '{extension_attributes}' should only contain lowercase letters and numbers"
if not re.match(r"^[a-z0-9]+$", extension_attribute):
extension_errors.append(
f"Extension attribute '{extension_attribute}' should only contain lowercase letters and numbers"
)
if extension_errors:
errors["extensions"] = extension_errors
return errors
def get_id(self) -> str:
"""

27
src/cloudevents/core/v1/exceptions.py Normal file
View File

@ -0,0 +1,27 @@
# Copyright 2018-Present The CloudEvents Authors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
class CloudEventValidationError(Exception):
"""
Custom exception for validation errors.
"""
def __init__(self, errors: dict[str, list[str]]) -> None:
super().__init__("Validation errors occurred")
self.errors = errors
def __str__(self) -> str:
error_messages = [
f"{key}: {', '.join(value)}" for key, value in self.errors.items()
]
return f"{super().__str__()}: {', '.join(error_messages)}"

142
tests/test_core/test_v1/test_event.py
View File

@ -12,79 +12,48 @@
# License for the specific language governing permissions and limitations
# under the License.
from cloudevents.core.v1.event import CloudEvent
from datetime import datetime, timezone
from typing import Any
import pytest
from datetime import datetime, timezone
from typing import Any, Optional
from cloudevents.core.v1.event import CloudEvent
from cloudevents.core.v1.exceptions import CloudEventValidationError
@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_attributes() -> None:
with pytest.raises(CloudEventValidationError) as e:
CloudEvent({})
assert e.value.errors == {
"required": ["Missing required attribute(s): id, source, type, specversion"],
"type": [
"Attribute 'id' must not be None",
"Attribute 'id' must be a string",
"Attribute 'source' must be a string",
"Attribute 'type' must be a string",
"Attribute 'specversion' must be a string",
"Attribute 'specversion' must be '1.0'",
],
)
def test_missing_required_attribute(attributes: dict, missing_attribute: str) -> None:
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: Optional[Any], error: str) -> None:
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: Any, error: str) -> None:
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: Any, error: str) -> None:
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"),
(
"2023-10-25T17:09:19.736166Z",
{"optional": ["Attribute 'time' must be a datetime object"]},
),
(
datetime(2023, 10, 25, 17, 9, 19, 736166),
"Attribute 'time' must be timezone aware",
{"optional": ["Attribute 'time' must be timezone aware"]},
),
(1, {"optional": ["Attribute 'time' must be a datetime object"]}),
],
)
def test_time_validation(time: Any, error: str) -> None:
with pytest.raises((ValueError, TypeError)) as e:
def test_time_validation(time: Any, error: dict) -> None:
with pytest.raises(CloudEventValidationError) as e:
CloudEvent(
{
"id": "1",
@ -95,21 +64,21 @@ def test_time_validation(time: Any, error: str) -> None:
}
)
assert str(e.value) == error
assert e.value.errors == error
@pytest.mark.parametrize(
"subject,error",
[
(1234, "Attribute 'subject' must be a string"),
(1234, {"optional": ["Attribute 'subject' must be a string"]}),
(
"",
"Attribute 'subject' must not be empty",
{"optional": ["Attribute 'subject' must not be empty"]},
),
],
)
def test_subject_validation(subject: Any, error: str) -> None:
with pytest.raises((ValueError, TypeError)) as e:
def test_subject_validation(subject: Any, error: dict) -> None:
with pytest.raises(CloudEventValidationError) as e:
CloudEvent(
{
"id": "1",
@ -120,21 +89,21 @@ def test_subject_validation(subject: Any, error: str) -> None:
}
)
assert str(e.value) == error
assert e.value.errors == error
@pytest.mark.parametrize(
"datacontenttype,error",
[
(1234, "Attribute 'datacontenttype' must be a string"),
(1234, {"optional": ["Attribute 'datacontenttype' must be a string"]}),
(
"",
"Attribute 'datacontenttype' must not be empty",
{"optional": ["Attribute 'datacontenttype' must not be empty"]},
),
],
)
def test_datacontenttype_validation(datacontenttype: Any, error: str) -> None:
with pytest.raises((ValueError, TypeError)) as e:
def test_datacontenttype_validation(datacontenttype: Any, error: dict) -> None:
with pytest.raises(CloudEventValidationError) as e:
CloudEvent(
{
"id": "1",
@ -145,21 +114,21 @@ def test_datacontenttype_validation(datacontenttype: Any, error: str) -> None:
}
)
assert str(e.value) == error
assert e.value.errors == error
@pytest.mark.parametrize(
"dataschema,error",
[
(1234, "Attribute 'dataschema' must be a string"),
(1234, {"optional": ["Attribute 'dataschema' must be a string"]}),
(
"",
"Attribute 'dataschema' must not be empty",
{"optional": ["Attribute 'dataschema' must not be empty"]},
),
],
)
def test_dataschema_validation(dataschema: Any, error: str) -> None:
with pytest.raises((ValueError, TypeError)) as e:
with pytest.raises(CloudEventValidationError) as e:
CloudEvent(
{
"id": "1",
@ -170,7 +139,7 @@ def test_dataschema_validation(dataschema: Any, error: str) -> None:
}
)
assert str(e.value) == error
assert e.value.errors == error
@pytest.mark.parametrize(
@ -178,24 +147,33 @@ def test_dataschema_validation(dataschema: Any, error: str) -> None:
[
(
"",
{
"extensions": [
"Extension attribute '' should be between 1 and 20 characters long",
"Extension attribute '' should only contain lowercase letters and numbers",
]
},
),
(
"thisisaverylongextension",
"Extension attribute 'thisisaverylongextension' should be between 1 and 20 characters long",
),
(
"ThisIsNotValid",
"Extension attribute 'ThisIsNotValid' should only contain lowercase letters and numbers",
{
"extensions": [
"Extension attribute 'thisisaverylongextension' should be between 1 and 20 characters long"
]
},
),
(
"data",
"Extension attribute 'data' is reserved and must not be used",
{
"extensions": [
"Extension attribute 'data' is reserved and must not be used"
]
},
),
],
)
def test_custom_extension(extension_name: str, error: str) -> None:
with pytest.raises(ValueError) as e:
def test_custom_extension(extension_name: str, error: dict) -> None:
with pytest.raises(CloudEventValidationError) as e:
CloudEvent(
{
"id": "1",
@ -206,7 +184,7 @@ def test_custom_extension(extension_name: str, error: str) -> None:
}
)
assert str(e.value) == error
assert e.value.errors == error
def test_cloud_event_constructor() -> None: