|
2 | 2 | import io |
3 | 3 | import json |
4 | 4 |
|
5 | | -import requests |
| 5 | +import pytest |
| 6 | +from client import image_bytes |
6 | 7 | from PIL import Image |
| 8 | +from server import app |
7 | 9 |
|
8 | 10 | from cloudevents.sdk.http import ( |
9 | 11 | CloudEvent, |
|
12 | 14 | to_structured_http, |
13 | 15 | ) |
14 | 16 |
|
15 | | -resp = requests.get( |
16 | | - "https://raw.githubusercontent.com/cncf/artwork/master/projects/cloudevents/horizontal/color/cloudevents-horizontal-color.png" |
17 | | -) |
18 | | -image_bytes = resp.content |
19 | | - |
20 | 17 | image_fileobj = io.BytesIO(image_bytes) |
21 | 18 | image_expected_shape = (1880, 363) |
22 | 19 |
|
23 | 20 |
|
| 21 | +@pytest.fixture |
| 22 | +def client(): |
| 23 | + app.testing = True |
| 24 | + return app.test_client() |
| 25 | + |
| 26 | + |
24 | 27 | def test_create_binary_image(): |
25 | 28 | # Create image and turn image into bytes |
26 | 29 | attributes = { |
@@ -80,6 +83,44 @@ def test_create_structured_image(): |
80 | 83 | assert restore_image.size == image_expected_shape |
81 | 84 |
|
82 | 85 |
|
| 86 | +def test_server_structured(client): |
| 87 | + attributes = { |
| 88 | + "type": "com.example.base64", |
| 89 | + "source": "https://example.com/event-producer", |
| 90 | + } |
| 91 | + |
| 92 | + event = CloudEvent(attributes, image_bytes) |
| 93 | + |
| 94 | + # Create cloudevent HTTP headers and content |
| 95 | + # Note that to_structured_http will create a data_base64 data field in |
| 96 | + # specversion 1.0 (default specversion) if given |
| 97 | + # an event whose data field is of type bytes. |
| 98 | + headers, body = to_structured_http(event) |
| 99 | + |
| 100 | + # Send cloudevent |
| 101 | + r = client.post("/", headers=headers, data=body) |
| 102 | + assert r.status_code == 200 |
| 103 | + assert r.data.decode() == f"Found image of size {image_expected_shape}" |
| 104 | + |
| 105 | + |
| 106 | +def test_server_binary(client): |
| 107 | + # Create cloudevent |
| 108 | + attributes = { |
| 109 | + "type": "com.example.string", |
| 110 | + "source": "https://example.com/event-producer", |
| 111 | + } |
| 112 | + |
| 113 | + event = CloudEvent(attributes, image_bytes) |
| 114 | + |
| 115 | + # Create cloudevent HTTP headers and content |
| 116 | + headers, body = to_binary_http(event) |
| 117 | + |
| 118 | + # Send cloudevent |
| 119 | + r = client.post("/", headers=headers, data=body) |
| 120 | + assert r.status_code == 200 |
| 121 | + assert r.data.decode() == f"Found image of size {image_expected_shape}" |
| 122 | + |
| 123 | + |
83 | 124 | def test_image_content(): |
84 | 125 | # Get image and check size |
85 | 126 | im = Image.open(image_fileobj) |
|
0 commit comments