1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
|
from testtools.matchers import Contains
import falcon
import io
import falcon.testing as testing
import six
class HelloResource:
sample_status = '200 OK'
sample_unicode = (six.u('Hello World! \x80') +
six.text_type(testing.rand_string(0, 0)))
sample_utf8 = sample_unicode.encode('utf-8')
def __init__(self, mode):
self.called = False
self.mode = mode
def on_get(self, req, resp):
self.called = True
self.req, self.resp = req, resp
resp.status = falcon.HTTP_200
if 'stream' in self.mode:
resp.stream = io.BytesIO(self.sample_utf8)
if 'stream_len' in self.mode:
resp.stream_len = len(self.sample_utf8)
if 'body' in self.mode:
if 'bytes' in self.mode:
resp.body = self.sample_utf8
else:
resp.body = self.sample_unicode
if 'data' in self.mode:
resp.data = self.sample_utf8
def on_head(self, req, resp):
self.on_get(req, resp)
class NoStatusResource:
def on_get(self, req, resp):
pass
class TestHelloWorld(testing.TestBase):
def before(self):
self.resource = HelloResource('body')
self.api.add_route(self.test_route, self.resource)
self.bytes_resource = HelloResource('body, bytes')
self.api.add_route('/bytes', self.bytes_resource)
self.data_resource = HelloResource('data')
self.api.add_route('/data', self.data_resource)
self.chunked_resource = HelloResource('stream')
self.api.add_route('/chunked-stream', self.chunked_resource)
self.stream_resource = HelloResource('stream, stream_len')
self.api.add_route('/stream', self.stream_resource)
self.no_status_resource = NoStatusResource()
self.api.add_route('/nostatus', self.no_status_resource)
self.root_resource = testing.TestResource()
self.api.add_route('/', self.root_resource)
def after(self):
pass
def test_env_headers_list_of_tuples(self):
env = testing.create_environ(headers=[('User-Agent', 'Falcon-Test')])
self.assertEqual(env['HTTP_USER_AGENT'], 'Falcon-Test')
def test_empty_route(self):
self.simulate_request('')
self.assertTrue(self.root_resource.called)
def test_route_negative(self):
bogus_route = self.test_route + 'x'
self.simulate_request(bogus_route)
# Ensure the request was NOT routed to resource
self.assertFalse(self.resource.called)
self.assertEqual(self.srmock.status, falcon.HTTP_404)
def test_body(self):
body = self.simulate_request(self.test_route)
resp = self.resource.resp
content_length = int(self.srmock.headers_dict['content-length'])
self.assertEqual(content_length, len(self.resource.sample_utf8))
self.assertEqual(self.srmock.status, self.resource.sample_status)
self.assertEqual(resp.status, self.resource.sample_status)
self.assertEqual(resp.body_encoded, self.resource.sample_utf8)
self.assertEqual(body, [self.resource.sample_utf8])
def test_body_bytes(self):
body = self.simulate_request('/bytes')
resp = self.bytes_resource.resp
content_length = int(self.srmock.headers_dict['content-length'])
self.assertEqual(content_length, len(self.resource.sample_utf8))
self.assertEqual(self.srmock.status, self.resource.sample_status)
self.assertEqual(resp.status, self.resource.sample_status)
self.assertEqual(resp.body_encoded, self.resource.sample_utf8)
self.assertEqual(body, [self.resource.sample_utf8])
def test_data(self):
body = self.simulate_request('/data')
resp = self.data_resource.resp
content_length = int(self.srmock.headers_dict['content-length'])
self.assertEqual(content_length, len(self.resource.sample_utf8))
self.assertEqual(self.srmock.status, self.resource.sample_status)
self.assertEqual(resp.status, self.resource.sample_status)
self.assertEqual(resp.data, self.resource.sample_utf8)
self.assertEqual(body, [self.resource.sample_utf8])
def test_no_body_on_head(self):
body = self.simulate_request(self.test_route, method='HEAD')
self.assertEqual(body, [])
self.assertEqual(self.srmock.status, falcon.HTTP_200)
def test_stream_chunked(self):
src = self.simulate_request('/chunked-stream')
dest = io.BytesIO()
for chunk in src:
dest.write(chunk)
self.assertEqual(dest.getvalue(), self.chunked_resource.sample_utf8)
for header in self.srmock.headers:
self.assertNotEqual(header[0].lower(), 'content-length')
def test_stream_known_len(self):
src = self.simulate_request('/stream')
dest = io.BytesIO()
for chunk in src:
dest.write(chunk)
expected_len = self.stream_resource.resp.stream_len
content_length = ('content-length', str(expected_len))
self.assertThat(self.srmock.headers, Contains(content_length))
self.assertEqual(dest.tell(), expected_len)
self.assertEqual(dest.getvalue(), self.chunked_resource.sample_utf8)
def test_status_not_set(self):
body = self.simulate_request('/nostatus')
self.assertEqual(body, [])
self.assertEqual(self.srmock.status, falcon.HTTP_200)
|
