sanic/tests/test_requests.py

196 lines
4.6 KiB
Python
Raw Normal View History

from json import loads as json_loads, dumps as json_dumps
import pytest
from sanic import Sanic
from sanic.exceptions import ServerError
2017-01-14 05:41:54 +00:00
from sanic.response import json, text, redirect
2016-10-14 23:38:43 +01:00
from sanic.utils import sanic_endpoint_test
# ------------------------------------------------------------ #
# GET
# ------------------------------------------------------------ #
def test_sync():
app = Sanic('test_text')
@app.route('/')
def handler(request):
return text('Hello')
request, response = sanic_endpoint_test(app)
assert response.text == 'Hello'
def test_text():
app = Sanic('test_text')
@app.route('/')
async def handler(request):
return text('Hello')
request, response = sanic_endpoint_test(app)
assert response.text == 'Hello'
2016-12-14 05:24:26 +00:00
def test_headers():
app = Sanic('test_text')
@app.route('/')
async def handler(request):
headers = {"spam": "great"}
return text('Hello', headers=headers)
request, response = sanic_endpoint_test(app)
assert response.headers.get('spam') == 'great'
2016-12-30 18:15:08 +00:00
def test_non_str_headers():
2016-12-14 05:24:26 +00:00
app = Sanic('test_text')
@app.route('/')
async def handler(request):
headers = {"answer": 42}
return text('Hello', headers=headers)
request, response = sanic_endpoint_test(app)
assert response.headers.get('answer') == '42'
Fix sanic_endpoint_test working with redirects Before fix, it raises error like: ``` tests/test_utils.py F ================================= FAILURES ================================= ______________________________ test_redirect _______________________________ app = <sanic.sanic.Sanic object at 0x1045fda20>, method = 'get', uri = '/1', gather_request = True, debug = False server_kwargs = {}, request_args = (), request_kwargs = {} _collect_request = <function sanic_endpoint_test.<locals>._collect_request at 0x1045ec950> _collect_response = <function sanic_endpoint_test.<locals>._collect_response at 0x1045ec7b8> def sanic_endpoint_test(app, method='get', uri='/', gather_request=True, debug=False, server_kwargs={}, *request_args, **request_kwargs): results = [] exceptions = [] if gather_request: def _collect_request(request): results.append(request) app.request_middleware.appendleft(_collect_request) async def _collect_response(sanic, loop): try: response = await local_request(method, uri, *request_args, **request_kwargs) results.append(response) except Exception as e: exceptions.append(e) app.stop() app.run(host=HOST, debug=debug, port=PORT, after_start=_collect_response, **server_kwargs) if exceptions: raise ValueError("Exception during request: {}".format(exceptions)) if gather_request: try: > request, response = results E ValueError: too many values to unpack (expected 2) sanic/utils.py:47: ValueError During handling of the above exception, another exception occurred: utils_app = <sanic.sanic.Sanic object at 0x1045fda20> def test_redirect(utils_app): """Test sanic_endpoint_test is working for redirection""" > request, response = sanic_endpoint_test(utils_app, uri='/1') tests/test_utils.py:33: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ app = <sanic.sanic.Sanic object at 0x1045fda20>, method = 'get', uri = '/1', gather_request = True, debug = False server_kwargs = {}, request_args = (), request_kwargs = {} _collect_request = <function sanic_endpoint_test.<locals>._collect_request at 0x1045ec950> _collect_response = <function sanic_endpoint_test.<locals>._collect_response at 0x1045ec7b8> def sanic_endpoint_test(app, method='get', uri='/', gather_request=True, debug=False, server_kwargs={}, *request_args, **request_kwargs): results = [] exceptions = [] if gather_request: def _collect_request(request): results.append(request) app.request_middleware.appendleft(_collect_request) async def _collect_response(sanic, loop): try: response = await local_request(method, uri, *request_args, **request_kwargs) results.append(response) except Exception as e: exceptions.append(e) app.stop() app.run(host=HOST, debug=debug, port=PORT, after_start=_collect_response, **server_kwargs) if exceptions: raise ValueError("Exception during request: {}".format(exceptions)) if gather_request: try: request, response = results return request, response except: raise ValueError( "Request and response object expected, got ({})".format( > results)) E ValueError: Request and response object expected, got ([{}, {}, {}, <ClientResponse(http://127.0.0.1:42101/3) [200 OK]> E <CIMultiDictProxy('Content-Type': 'text/plain; charset=utf-8', 'Content-Length': '2', 'Connection': 'keep-alive', 'Keep-Alive': 'timeout=1')> E ]) sanic/utils.py:52: ValueError ```
2017-02-02 09:51:33 +00:00
def test_invalid_response():
app = Sanic('test_invalid_response')
2016-12-14 05:24:26 +00:00
@app.exception(ServerError)
def handler_exception(request, exception):
return text('Internal Server Error.', 500)
2016-12-14 05:24:26 +00:00
@app.route('/')
async def handler(request):
return 'This should fail'
request, response = sanic_endpoint_test(app)
assert response.status == 500
assert response.text == "Internal Server Error."
Fix sanic_endpoint_test working with redirects Before fix, it raises error like: ``` tests/test_utils.py F ================================= FAILURES ================================= ______________________________ test_redirect _______________________________ app = <sanic.sanic.Sanic object at 0x1045fda20>, method = 'get', uri = '/1', gather_request = True, debug = False server_kwargs = {}, request_args = (), request_kwargs = {} _collect_request = <function sanic_endpoint_test.<locals>._collect_request at 0x1045ec950> _collect_response = <function sanic_endpoint_test.<locals>._collect_response at 0x1045ec7b8> def sanic_endpoint_test(app, method='get', uri='/', gather_request=True, debug=False, server_kwargs={}, *request_args, **request_kwargs): results = [] exceptions = [] if gather_request: def _collect_request(request): results.append(request) app.request_middleware.appendleft(_collect_request) async def _collect_response(sanic, loop): try: response = await local_request(method, uri, *request_args, **request_kwargs) results.append(response) except Exception as e: exceptions.append(e) app.stop() app.run(host=HOST, debug=debug, port=PORT, after_start=_collect_response, **server_kwargs) if exceptions: raise ValueError("Exception during request: {}".format(exceptions)) if gather_request: try: > request, response = results E ValueError: too many values to unpack (expected 2) sanic/utils.py:47: ValueError During handling of the above exception, another exception occurred: utils_app = <sanic.sanic.Sanic object at 0x1045fda20> def test_redirect(utils_app): """Test sanic_endpoint_test is working for redirection""" > request, response = sanic_endpoint_test(utils_app, uri='/1') tests/test_utils.py:33: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ app = <sanic.sanic.Sanic object at 0x1045fda20>, method = 'get', uri = '/1', gather_request = True, debug = False server_kwargs = {}, request_args = (), request_kwargs = {} _collect_request = <function sanic_endpoint_test.<locals>._collect_request at 0x1045ec950> _collect_response = <function sanic_endpoint_test.<locals>._collect_response at 0x1045ec7b8> def sanic_endpoint_test(app, method='get', uri='/', gather_request=True, debug=False, server_kwargs={}, *request_args, **request_kwargs): results = [] exceptions = [] if gather_request: def _collect_request(request): results.append(request) app.request_middleware.appendleft(_collect_request) async def _collect_response(sanic, loop): try: response = await local_request(method, uri, *request_args, **request_kwargs) results.append(response) except Exception as e: exceptions.append(e) app.stop() app.run(host=HOST, debug=debug, port=PORT, after_start=_collect_response, **server_kwargs) if exceptions: raise ValueError("Exception during request: {}".format(exceptions)) if gather_request: try: request, response = results return request, response except: raise ValueError( "Request and response object expected, got ({})".format( > results)) E ValueError: Request and response object expected, got ([{}, {}, {}, <ClientResponse(http://127.0.0.1:42101/3) [200 OK]> E <CIMultiDictProxy('Content-Type': 'text/plain; charset=utf-8', 'Content-Length': '2', 'Connection': 'keep-alive', 'Keep-Alive': 'timeout=1')> E ]) sanic/utils.py:52: ValueError ```
2017-02-02 09:51:33 +00:00
def test_json():
app = Sanic('test_json')
@app.route('/')
async def handler(request):
return json({"test": True})
request, response = sanic_endpoint_test(app)
try:
results = json_loads(response.text)
except:
raise ValueError("Expected JSON response but got '{}'".format(response))
assert results.get('test') == True
2016-12-08 04:33:56 +00:00
def test_invalid_json():
app = Sanic('test_json')
@app.route('/')
async def handler(request):
return json(request.json())
data = "I am not json"
request, response = sanic_endpoint_test(app, data=data)
assert response.status == 400
def test_query_string():
app = Sanic('test_query_string')
@app.route('/')
async def handler(request):
return text('OK')
request, response = sanic_endpoint_test(
app, params=[("test1", "1"), ("test2", "false"), ("test2", "true")])
assert request.args.get('test1') == '1'
assert request.args.get('test2') == 'false'
def test_token():
app = Sanic('test_post_token')
@app.route('/')
async def handler(request):
return text('OK')
# uuid4 generated token.
token = 'a1d895e0-553a-421a-8e22-5ff8ecb48cbf'
headers = {
'content-type': 'application/json',
'Authorization': 'Token {}'.format(token)
}
request, response = sanic_endpoint_test(app, headers=headers)
assert request.token == token
# ------------------------------------------------------------ #
# POST
# ------------------------------------------------------------ #
def test_post_json():
app = Sanic('test_post_json')
@app.route('/')
async def handler(request):
return text('OK')
payload = {'test': 'OK'}
headers = {'content-type': 'application/json'}
request, response = sanic_endpoint_test(
app, data=json_dumps(payload), headers=headers)
assert request.json.get('test') == 'OK'
assert response.text == 'OK'
2016-10-19 08:23:44 +01:00
def test_post_form_urlencoded():
app = Sanic('test_post_form_urlencoded')
@app.route('/')
async def handler(request):
return text('OK')
payload = 'test=OK'
headers = {'content-type': 'application/x-www-form-urlencoded'}
request, response = sanic_endpoint_test(app, data=payload, headers=headers)
assert request.form.get('test') == 'OK'
def test_post_form_multipart_form_data():
app = Sanic('test_post_form_multipart_form_data')
@app.route('/')
async def handler(request):
return text('OK')
payload = '------sanic\r\n' \
'Content-Disposition: form-data; name="test"\r\n' \
'\r\n' \
'OK\r\n' \
'------sanic--\r\n'
headers = {'content-type': 'multipart/form-data; boundary=----sanic'}
request, response = sanic_endpoint_test(app, data=payload, headers=headers)
assert request.form.get('test') == 'OK'