def test_order_independent_mw_executed_when_exception_in_resp(self):
"""Test that error in inner middleware leaves"""
global context
class RaiseErrorMiddleware(object):
def process_response(self, req, resp, resource):
raise Exception('Always fail')
app = falcon.API(independent_middleware=True,
middleware=[ExecutedFirstMiddleware(),
RaiseErrorMiddleware(),
ExecutedLastMiddleware()])
def handler(ex, req, resp, params):
pass
app.add_error_handler(Exception, handler)
app.add_route(TEST_ROUTE, MiddlewareClassResource())
client = testing.TestClient(app)
client.simulate_request(path=TEST_ROUTE)
# Any mw is executed now...
expectedExecutedMethods = [
'ExecutedFirstMiddleware.process_request',
'ExecutedLastMiddleware.process_request',
'ExecutedFirstMiddleware.process_resource',
'ExecutedLastMiddleware.process_resource',
'ExecutedLastMiddleware.process_response',
'ExecutedFirstMiddleware.process_response'
]
assert expectedExecutedMethods == context['executed_methods']
python类testing()的实例源码
def test_order_mw_executed_when_exception_in_req(self):
"""Test that error in inner middleware leaves"""
global context
class RaiseErrorMiddleware(object):
def process_request(self, req, resp):
raise Exception('Always fail')
app = falcon.API(middleware=[ExecutedFirstMiddleware(),
RaiseErrorMiddleware(),
ExecutedLastMiddleware()])
def handler(ex, req, resp, params):
pass
app.add_error_handler(Exception, handler)
app.add_route(TEST_ROUTE, MiddlewareClassResource())
client = testing.TestClient(app)
client.simulate_request(path=TEST_ROUTE)
# Any mw is executed now...
expectedExecutedMethods = [
'ExecutedFirstMiddleware.process_request',
'ExecutedFirstMiddleware.process_response'
]
assert expectedExecutedMethods == context['executed_methods']
def test_order_independent_mw_executed_when_exception_in_req(self):
"""Test that error in inner middleware leaves"""
global context
class RaiseErrorMiddleware(object):
def process_request(self, req, resp):
raise Exception('Always fail')
app = falcon.API(independent_middleware=True,
middleware=[ExecutedFirstMiddleware(),
RaiseErrorMiddleware(),
ExecutedLastMiddleware()])
def handler(ex, req, resp, params):
pass
app.add_error_handler(Exception, handler)
app.add_route(TEST_ROUTE, MiddlewareClassResource())
client = testing.TestClient(app)
client.simulate_request(path=TEST_ROUTE)
# All response middleware still executed...
expectedExecutedMethods = [
'ExecutedFirstMiddleware.process_request',
'ExecutedLastMiddleware.process_response',
'ExecutedFirstMiddleware.process_response'
]
assert expectedExecutedMethods == context['executed_methods']
def test_order_mw_executed_when_exception_in_rsrc(self):
"""Test that error in inner middleware leaves"""
global context
class RaiseErrorMiddleware(object):
def process_resource(self, req, resp, resource):
raise Exception('Always fail')
app = falcon.API(middleware=[ExecutedFirstMiddleware(),
RaiseErrorMiddleware(),
ExecutedLastMiddleware()])
def handler(ex, req, resp, params):
pass
app.add_error_handler(Exception, handler)
app.add_route(TEST_ROUTE, MiddlewareClassResource())
client = testing.TestClient(app)
client.simulate_request(path=TEST_ROUTE)
# Any mw is executed now...
expectedExecutedMethods = [
'ExecutedFirstMiddleware.process_request',
'ExecutedLastMiddleware.process_request',
'ExecutedFirstMiddleware.process_resource',
'ExecutedLastMiddleware.process_response',
'ExecutedFirstMiddleware.process_response'
]
assert expectedExecutedMethods == context['executed_methods']
def test_base_path_is_removed_before_routing(self):
"""Test that RemoveBasePathMiddleware is executed before routing"""
app = falcon.API(middleware=RemoveBasePathMiddleware())
# We dont include /base_path as it will be removed in middleware
app.add_route('/sub_path', MiddlewareClassResource())
client = testing.TestClient(app)
response = client.simulate_request(path='/base_path/sub_path')
assert _EXPECTED_BODY == response.json
assert response.status == falcon.HTTP_200
response = client.simulate_request(path='/base_pathIncorrect/sub_path')
assert response.status == falcon.HTTP_404
def test_error_composed_before_resp_middleware_called(self):
mw = CaptureResponseMiddleware()
app = falcon.API(middleware=mw)
app.add_route('/', MiddlewareClassResource())
client = testing.TestClient(app)
response = client.simulate_request(path='/', method='POST')
assert response.status == falcon.HTTP_403
assert mw.resp.status == response.status
composed_body = json.loads(mw.resp.body)
assert composed_body['title'] == response.status
assert not mw.req_succeeded
# NOTE(kgriffs): Sanity-check the other params passed to
# process_response()
assert isinstance(mw.req, falcon.Request)
assert isinstance(mw.resource, MiddlewareClassResource)
def test_srmock(self):
mock = testing.StartResponseMock()
mock(falcon.HTTP_200, ())
assert mock.status == falcon.HTTP_200
assert mock.exc_info is None
mock = testing.StartResponseMock()
exc_info = sys.exc_info()
mock(falcon.HTTP_200, (), exc_info)
assert mock.exc_info == exc_info
def client():
app = falcon.API()
app.add_route('/', CookieResource())
app.add_route('/test-convert', CookieResourceMaxAgeFloatString())
return testing.TestClient(app)
# =====================================================================
# Response
# =====================================================================
def test_request_cookie_parsing():
# testing with a github-ish set of cookies
headers = [
(
'Cookie',
"""
logged_in=no;_gh_sess=eyJzZXXzaW9uX2lkIjoiN2;
tz=Europe/Berlin; _ga=GA1.2.332347814.1422308165;
_gat=1;
_octo=GH1.1.201722077.1422308165
"""
),
]
environ = testing.create_environ(headers=headers)
req = falcon.Request(environ)
assert req.cookies['logged_in'] == 'no'
assert req.cookies['tz'] == 'Europe/Berlin'
assert req.cookies['_octo'] == 'GH1.1.201722077.1422308165'
assert 'logged_in' in req.cookies
assert '_gh_sess' in req.cookies
assert 'tz' in req.cookies
assert '_ga' in req.cookies
assert '_gat' in req.cookies
assert '_octo' in req.cookies
def client():
app = falcon.API()
resource = RedirectingResource()
app.add_route('/', resource)
return testing.TestClient(app)
def resource():
return testing.TestResource()
def test_bounded_stream_property_empty_body(self):
"""Test that we can get a bounded stream outside of wsgiref."""
environ = testing.create_environ()
req = falcon.Request(environ)
bounded_stream = req.bounded_stream
# NOTE(kgriffs): Verify that we aren't creating a new object
# each time the property is called. Also ensures branch
# coverage of the property implementation.
assert bounded_stream is req.bounded_stream
data = bounded_stream.read()
assert len(data) == 0
def test_request_repr(self):
environ = testing.create_environ()
req = falcon.Request(environ)
_repr = '<%s: %s %r>' % (req.__class__.__name__, req.method, req.url)
assert req.__repr__() == _repr
def client():
app = falcon.API()
resource = FaultyResource()
app.add_route('/fail', resource)
return testing.TestClient(app)
def client(resource):
app = falcon.API()
app.add_route('/', resource)
return testing.TestClient(app)
def test_post_read_bounded_stream(self):
body = testing.rand_string(_SIZE_1_KB / 2, _SIZE_1_KB)
resp = requests.post(_SERVER_BASE_URL + 'bucket', data=body)
assert resp.status_code == 200
assert resp.text == body
def client():
return testing.TestClient(falcon.API())
# NOTE(kgriffs): Concept from Gunicorn's source (wsgi.py)
def test_env_headers_list_of_tuples(self):
env = testing.create_environ(headers=[('User-Agent', 'Falcon-Test')])
assert env['HTTP_USER_AGENT'] == 'Falcon-Test'
def test_root_route(self, client):
doc = {u'message': u'Hello world!'}
resource = testing.SimpleTestResource(json=doc)
client.app.add_route('/', resource)
result = client.simulate_get()
assert result.json == doc
def client():
app = falcon.API()
app.add_route('/stonewall', Stonewall())
resource_things = ThingsResource()
app.add_route('/things', resource_things)
app.add_route('/things/{id}/stuff/{sid}', resource_things)
resource_misc = MiscResource()
app.add_route('/misc', resource_misc)
resource_get_with_faulty_put = GetWithFaultyPutResource()
app.add_route('/get_with_param/{param}', resource_get_with_faulty_put)
return testing.TestClient(app)
