a15ee3ad06
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
```
192 lines
4.6 KiB
Python
192 lines
4.6 KiB
Python
from json import loads as json_loads, dumps as json_dumps
|
|
from sanic import Sanic
|
|
from sanic.response import json, text, redirect
|
|
from sanic.utils import sanic_endpoint_test
|
|
from sanic.exceptions import ServerError
|
|
|
|
import pytest
|
|
|
|
# ------------------------------------------------------------ #
|
|
# 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'
|
|
|
|
|
|
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'
|
|
|
|
|
|
def test_non_str_headers():
|
|
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'
|
|
|
|
def test_invalid_response():
|
|
app = Sanic('test_invalid_response')
|
|
|
|
@app.exception(ServerError)
|
|
def handler_exception(request, exception):
|
|
return text('Internal Server Error.', 500)
|
|
|
|
@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."
|
|
|
|
|
|
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
|
|
|
|
|
|
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'
|
|
|
|
|
|
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'
|