Implemented to_json and from_json (#72)

* added test_to_json test

Signed-off-by: Curtis Mason <cumason@google.com>

* implemented to_json with tests

Signed-off-by: Curtis Mason <cumason@google.com>

* from_json and to_json tests

Signed-off-by: Curtis Mason <cumason@google.com>

* Tests for to_json being able to talk to from_json

Signed-off-by: Curtis Mason <cumason@google.com>

* lint fix

Signed-off-by: Curtis Mason <cumason@google.com>

* added documentation for to_json and from_json

Signed-off-by: Curtis Mason <cumason@google.com>

* lint fix

Signed-off-by: Curtis Mason <cumason@google.com>

* lint fix

Signed-off-by: Curtis Mason <cumason@google.com>
This commit is contained in:
Curtis Mason 2020-07-22 18:18:31 -04:00 committed by GitHub
parent b2a87a8af6
commit 0aa01ba5c6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 159 additions and 11 deletions

View File

@ -43,7 +43,8 @@ def from_http(
headers: typing.Dict[str, str],
data_unmarshaller: types.UnmarshallerType = None,
):
"""Unwrap a CloudEvent (binary or structured) from an HTTP request.
"""
Unwrap a CloudEvent (binary or structured) from an HTTP request.
:param data: the HTTP request body
:type data: typing.IO
:param headers: the HTTP headers
@ -82,5 +83,32 @@ def from_http(
return CloudEvent(attrs, event.data)
def from_json():
raise NotImplementedError
def to_json(
event: EventClass, data_marshaller: types.MarshallerType = None
) -> typing.Union[str, bytes]:
"""
Cast an EventClass into a json object
:param event: EventClass which will be converted into a json object
:type event: EventClass
:param data_marshaller: Callable function which will cast event.data
into a json object
:type data_marshaller: typing.Callable
:returns: json object representing the given event
"""
return to_structured_http(event, data_marshaller=data_marshaller)[1]
def from_json(
data: typing.Union[str, bytes],
data_unmarshaller: types.UnmarshallerType = None,
) -> EventClass:
"""
Cast json encoded data into an EventClass
:param data: json encoded cloudevent data
:type event: typing.Union[str, bytes]
:param data_unmarshaller: Callable function which will cast json encoded
data into a python object retrievable from returned EventClass.data
:type data_marshaller: typing.Callable
:returns: EventClass representing given cloudevent json object
"""
return from_http(data=data, headers={}, data_unmarshaller=data_unmarshaller)

View File

@ -176,11 +176,3 @@ def to_binary_http(
format=converters.TypeBinary,
data_marshaller=data_marshaller,
)
def to_json():
raise NotImplementedError
def from_json():
raise NotImplementedError

View File

@ -0,0 +1,128 @@
# All Rights Reserved.
#
# 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.
import base64
import json
import pytest
from cloudevents.sdk.http import CloudEvent, from_json, to_json
test_data = json.dumps({"data-key": "val"})
test_attributes = {
"type": "com.example.string",
"source": "https://example.com/event-producer",
}
@pytest.mark.parametrize("specversion", ["0.3", "1.0"])
def test_to_json(specversion):
event = CloudEvent(test_attributes, test_data)
event_json = to_json(event)
event_dict = json.loads(event_json)
for key, val in test_attributes.items():
assert event_dict[key] == val
assert event_dict["data"] == test_data
@pytest.mark.parametrize("specversion", ["0.3", "1.0"])
def test_to_json_base64(specversion):
data = b"test123"
event = CloudEvent(test_attributes, data)
event_json = to_json(event)
event_dict = json.loads(event_json)
for key, val in test_attributes.items():
assert event_dict[key] == val
# test data was properly marshalled into data_base64
data_base64 = event_dict["data_base64"].encode()
test_data_base64 = base64.b64encode(data)
assert data_base64 == test_data_base64
@pytest.mark.parametrize("specversion", ["0.3", "1.0"])
def test_from_json(specversion):
payload = {
"type": "com.example.string",
"source": "https://example.com/event-producer",
"id": "1234",
"specversion": specversion,
"data": {"data-key": "val"},
}
event = from_json(json.dumps(payload))
for key, val in payload.items():
if key == "data":
assert event.data == payload["data"]
else:
assert event[key] == val
@pytest.mark.parametrize("specversion", ["0.3", "1.0"])
def test_from_json_base64(specversion):
# Create base64 encoded data
raw_data = {"data-key": "val"}
data = json.dumps(raw_data).encode()
data_base64_str = base64.b64encode(data).decode()
# Create json payload
payload = {
"type": "com.example.string",
"source": "https://example.com/event-producer",
"id": "1234",
"specversion": specversion,
"data_base64": data_base64_str,
}
payload_json = json.dumps(payload)
# Create event
event = from_json(payload_json)
# Test fields were marshalled properly
for key, val in payload.items():
if key == "data_base64":
# Check data_base64 was unmarshalled properly
assert event.data == raw_data
else:
assert event[key] == val
@pytest.mark.parametrize("specversion", ["0.3", "1.0"])
def test_json_can_talk_to_itself(specversion):
event = CloudEvent(test_attributes, test_data)
event_json = to_json(event)
event = from_json(event_json)
for key, val in test_attributes.items():
assert event[key] == val
assert event.data == test_data
@pytest.mark.parametrize("specversion", ["0.3", "1.0"])
def test_json_can_talk_to_itself_base64(specversion):
data = b"test123"
event = CloudEvent(test_attributes, data)
event_json = to_json(event)
event = from_json(event_json)
for key, val in test_attributes.items():
assert event[key] == val
assert event.data == data