d4f1a950e4
resp.body decodes strings now, which slows it down a bit. Use resp.data, instead, for byte strings.
154 lines
4.8 KiB
Python
154 lines
4.8 KiB
Python
from testtools.matchers import Contains
|
|
|
|
import falcon
|
|
import io
|
|
from . import helpers
|
|
|
|
import six
|
|
|
|
|
|
class HelloResource:
|
|
sample_status = '200 OK'
|
|
sample_unicode = (u'Hello World! \x80' +
|
|
six.text_type(helpers.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(helpers.TestSuite):
|
|
|
|
def prepare(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 = helpers.TestResource()
|
|
self.api.add_route('', self.root_resource)
|
|
|
|
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.assertEquals(self.srmock.status, falcon.HTTP_404)
|
|
|
|
def test_body(self):
|
|
body = self._simulate_request(self.test_route)
|
|
resp = self.resource.resp
|
|
|
|
self.assertEquals(self.srmock.status, self.resource.sample_status)
|
|
self.assertEquals(resp.status, self.resource.sample_status)
|
|
self.assertEquals(resp.body, self.resource.sample_unicode)
|
|
self.assertEquals(body, [self.resource.sample_utf8])
|
|
|
|
if not six.PY3:
|
|
# On Python 3, strings are always Unicode,
|
|
# so only perform this test under Python 2.
|
|
def test_body_bytes(self):
|
|
body = self._simulate_request('/bytes')
|
|
resp = self.bytes_resource.resp
|
|
|
|
self.assertEquals(self.srmock.status, self.resource.sample_status)
|
|
self.assertEquals(resp.status, self.resource.sample_status)
|
|
self.assertEquals(resp.body, self.resource.sample_utf8)
|
|
self.assertEquals(body, [self.resource.sample_utf8])
|
|
|
|
def test_data(self):
|
|
body = self._simulate_request('/data')
|
|
resp = self.data_resource.resp
|
|
|
|
self.assertEquals(self.srmock.status, self.resource.sample_status)
|
|
self.assertEquals(resp.status, self.resource.sample_status)
|
|
self.assertEquals(resp.data, self.resource.sample_utf8)
|
|
self.assertEquals(body, [self.resource.sample_utf8])
|
|
|
|
def test_no_body_on_head(self):
|
|
body = self._simulate_request(self.test_route, method='HEAD')
|
|
self.assertEquals(body, [])
|
|
self.assertEquals(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)
|