diff --git a/requirements/test.txt b/requirements/test.txt index 1d5f3087..9972370a 100644 --- a/requirements/test.txt +++ b/requirements/test.txt @@ -10,4 +10,4 @@ sanic aiohttp Pillow requests - +Flask diff --git a/samples/http-image-cloudevents/server.py b/samples/http-image-cloudevents/server.py index 048d0efa..7704a30a 100644 --- a/samples/http-image-cloudevents/server.py +++ b/samples/http-image-cloudevents/server.py @@ -13,7 +13,7 @@ # under the License. import io -from flask import Flask, request +from flask import Flask, Response, request from PIL import Image from cloudevents.sdk.http import from_http @@ -36,7 +36,7 @@ def home(): # Print print(f"Found event {event['id']} with image of size {image.size}") - return "", 204 + return f"Found image of size {image.size}", 200 if __name__ == "__main__": diff --git a/samples/http-image-cloudevents/test_image_sample.py b/samples/http-image-cloudevents/test_image_sample.py index 83436c1e..c65b3e6b 100644 --- a/samples/http-image-cloudevents/test_image_sample.py +++ b/samples/http-image-cloudevents/test_image_sample.py @@ -2,8 +2,10 @@ import io import json -import requests +import pytest +from client import image_bytes from PIL import Image +from server import app from cloudevents.sdk.http import ( CloudEvent, @@ -12,15 +14,16 @@ to_structured_http, ) -resp = requests.get( - "https://raw.githubusercontent.com/cncf/artwork/master/projects/cloudevents/horizontal/color/cloudevents-horizontal-color.png" -) -image_bytes = resp.content - image_fileobj = io.BytesIO(image_bytes) image_expected_shape = (1880, 363) +@pytest.fixture +def client(): + app.testing = True + return app.test_client() + + def test_create_binary_image(): # Create image and turn image into bytes attributes = { @@ -80,6 +83,44 @@ def test_create_structured_image(): assert restore_image.size == image_expected_shape +def test_send_structured_request(client): + attributes = { + "type": "com.example.base64", + "source": "https://example.com/event-producer", + } + + event = CloudEvent(attributes, image_bytes) + + # Create cloudevent HTTP headers and content + # Note that to_structured_http will create a data_base64 data field in + # specversion 1.0 (default specversion) if given + # an event whose data field is of type bytes. + headers, body = to_structured_http(event) + + # Send cloudevent + r = client.post("/", headers=headers, data=body) + assert r.status_code == 200 + assert r.data.decode() == f"Found image of size {image_expected_shape}" + + +def test_send_binary_request(client): + # Create cloudevent + attributes = { + "type": "com.example.string", + "source": "https://example.com/event-producer", + } + + event = CloudEvent(attributes, image_bytes) + + # Create cloudevent HTTP headers and content + headers, body = to_binary_http(event) + + # Send cloudevent + r = client.post("/", headers=headers, data=body) + assert r.status_code == 200 + assert r.data.decode() == f"Found image of size {image_expected_shape}" + + def test_image_content(): # Get image and check size im = Image.open(image_fileobj)