Vasanko/sanic
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Add tests (#1433)

Browse Source 
* Add tests for remove_route()

* Add tests for sanic/router.py

* Add tests for sanic/cookies.py

* Disable reset logging in test_logging.py

* Add tests for sanic/request.py

* Add tests for ContentRangeHandler

* Add tests for exception at response middleware

* Fix cached_handlers for ErrorHandler.lookup()

* Add test for websocket request timeout

* Add tests for getting cookies of StreamResponse, Remove some unused variables in tests/test_cookies.py

* Add tests for nested error handle
This commit is contained in:
Jacob 2018-12-22 23:21:45 +08:00 committed by Stephen Sadowski
parent d2670664ba
commit 4efd450b32
12 changed files with 554 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
sanic/handlers.py
View File

@ -61,7 +61,7 @@ class ErrorHandler:
self.handlers.append((exception, handler))
def lookup(self, exception):
handler = self.cached_handlers.get(exception, self._missing)
handler = self.cached_handlers.get(type(exception), self._missing)
if handler is self._missing:
for exception_class, handler in self.handlers:
if isinstance(exception, exception_class):

4
sanic/request.py
View File

@ -362,8 +362,8 @@ def parse_multipart_form(body, boundary):
fields[field_name] = [value]
else:
logger.debug(
"Form-data field does not have a 'name' parameter \
in the Content-Disposition header"
"Form-data field does not have a 'name' parameter "
"in the Content-Disposition header"
)
return fields, files

69
tests/test_app.py
View File

@ -1,4 +1,5 @@
import asyncio
import logging
import pytest
@ -69,7 +70,7 @@ def test_app_handle_request_handler_is_none(app, monkeypatch):
@pytest.mark.parametrize('websocket_enabled', [True, False])
@pytest.mark.parametrize('enable', [True, False])
def test_enable_websocket(app, websocket_enabled, enable):
def test_app_enable_websocket(app, websocket_enabled, enable):
app.websocket_enabled = websocket_enabled
app.enable_websocket(enable=enable)
@ -80,3 +81,69 @@ def test_enable_websocket(app, websocket_enabled, enable):
await ws.send('test')
assert app.websocket_enabled == True
def test_handle_request_with_nested_exception(app, monkeypatch):
err_msg = 'Mock Exception'
# Not sure how to raise an exception in app.error_handler.response(), use mock here
def mock_error_handler_response(*args, **kwargs):
raise Exception(err_msg)
monkeypatch.setattr(app.error_handler, 'response', mock_error_handler_response)
@app.get('/')
def handler(request):
raise Exception
return text('OK')
request, response = app.test_client.get('/')
assert response.status == 500
assert response.text == 'An error occurred while handling an error'
def test_handle_request_with_nested_exception_debug(app, monkeypatch):
err_msg = 'Mock Exception'
# Not sure how to raise an exception in app.error_handler.response(), use mock here
def mock_error_handler_response(*args, **kwargs):
raise Exception(err_msg)
monkeypatch.setattr(app.error_handler, 'response', mock_error_handler_response)
@app.get('/')
def handler(request):
raise Exception
return text('OK')
request, response = app.test_client.get('/', debug=True)
assert response.status == 500
assert response.text.startswith(
'Error while handling error: {}\nStack: Traceback (most recent call last):\n'.format(err_msg)
)
def test_handle_request_with_nested_sanic_exception(app, monkeypatch, caplog):
# Not sure how to raise an exception in app.error_handler.response(), use mock here
def mock_error_handler_response(*args, **kwargs):
raise SanicException('Mock SanicException')
monkeypatch.setattr(app.error_handler, 'response', mock_error_handler_response)
@app.get('/')
def handler(request):
raise Exception
return text('OK')
with caplog.at_level(logging.ERROR):
request, response = app.test_client.get('/')
assert response.status == 500
assert response.text == 'Error: Mock SanicException'
assert caplog.record_tuples[0] == (
'sanic.root',
logging.ERROR,
"Exception occurred while handling uri: 'http://127.0.0.1:42101/'"
)

59
tests/test_cookies.py
View File

@ -130,3 +130,62 @@ def test_cookie_set_unknown_property():
with pytest.raises(expected_exception=KeyError) as e:
c["invalid"] = "value"
assert e.message == "Unknown cookie property"
def test_cookie_set_same_key(app):
cookies = {'test': 'wait'}
@app.get('/')
def handler(request):
response = text('pass')
response.cookies['test'] = 'modified'
response.cookies['test'] = 'pass'
return response
request, response = app.test_client.get('/', cookies=cookies)
assert response.status == 200
assert response.cookies['test'].value == 'pass'
@pytest.mark.parametrize('max_age', ['0', 30, '30'])
def test_cookie_max_age(app, max_age):
cookies = {'test': 'wait'}
@app.get('/')
def handler(request):
response = text('pass')
response.cookies['test'] = 'pass'
response.cookies['test']['max-age'] = max_age
return response
request, response = app.test_client.get('/', cookies=cookies)
assert response.status == 200
assert response.cookies['test'].value == 'pass'
assert response.cookies['test']['max-age'] == str(max_age)
@pytest.mark.parametrize('expires', [
datetime.now() + timedelta(seconds=60),
'Fri, 21-Dec-2018 15:30:00 GMT'
])
def test_cookie_expires(app, expires):
cookies = {'test': 'wait'}
@app.get('/')
def handler(request):
response = text('pass')
response.cookies['test'] = 'pass'
response.cookies['test']['expires'] = expires
return response
request, response = app.test_client.get('/', cookies=cookies)
assert response.status == 200
assert response.cookies['test'].value == 'pass'
if isinstance(expires, datetime):
expires = expires.strftime("%a, %d-%b-%Y %T GMT")
assert response.cookies['test']['expires'] == expires

2
tests/test_logging.py
View File

@ -125,7 +125,7 @@ def test_logger(caplog):
def test_logging_modified_root_logger_config():
reset_logging()
# reset_logging()
modified_config = LOGGING_CONFIG_DEFAULTS
modified_config['loggers']['sanic.root']['level'] = 'DEBUG'

51
tests/test_middleware.py
View File

@ -1,7 +1,9 @@
from sanic.request import Request
from sanic.response import text, HTTPResponse
from sanic.exceptions import NotFound
import logging
from asyncio import CancelledError
from sanic.exceptions import NotFound
from sanic.request import Request
from sanic.response import HTTPResponse, text
# ------------------------------------------------------------ #
# GET
@ -69,6 +71,49 @@ def test_middleware_response_exception(app):
assert result['status_code'] == 404
def test_middleware_response_raise_cancelled_error(app, caplog):
@app.middleware('response')
async def process_response(request, response):
raise CancelledError('CancelledError at response middleware')
@app.get('/')
def handler(request):
return text('OK')
with caplog.at_level(logging.ERROR):
reqrequest, response = app.test_client.get('/')
assert response.status == 503
assert caplog.record_tuples[0] == (
'sanic.root',
logging.ERROR,
'Exception occurred while handling uri: \'http://127.0.0.1:42101/\''
)
def test_middleware_response_raise_exception(app, caplog):
@app.middleware('response')
async def process_response(request, response):
raise Exception('Exception at response middleware')
with caplog.at_level(logging.ERROR):
reqrequest, response = app.test_client.get('/')
assert response.status == 404
assert caplog.record_tuples[0] == (
'sanic.root',
logging.ERROR,
'Exception occurred while handling uri: \'http://127.0.0.1:42101/\''
)
assert caplog.record_tuples[1] == (
'sanic.error',
logging.ERROR,
'Exception occurred in one of response middleware handlers'
)
def test_middleware_override_request(app):
@app.middleware

21
tests/test_request_timeout.py
View File

@ -188,6 +188,11 @@ async def handler2(request):
return text('OK')
@request_timeout_default_app.websocket('/ws1')
async def ws_handler1(request, ws):
await ws.send('OK')
def test_default_server_error_request_timeout():
client = DelayableSanicTestClient(request_timeout_default_app, None, 2)
request, response = client.get('/1')
@ -200,3 +205,19 @@ def test_default_server_error_request_dont_timeout():
request, response = client.get('/1')
assert response.status == 200
assert response.text == 'OK'
def test_default_server_error_websocket_request_timeout():
headers={
'Upgrade': 'websocket',
'Connection': 'upgrade',
'Sec-WebSocket-Key': 'dGhlIHNhbXBsZSBub25jZQ==',
'Sec-WebSocket-Version': '13'
}
client = DelayableSanicTestClient(request_timeout_default_app, None, 2)
request, response = client.get('/ws1', headers=headers)
assert response.status == 408
assert response.text == 'Error: Request Timeout'

222
tests/test_requests.py
View File

@ -1,16 +1,17 @@
from json import loads as json_loads, dumps as json_dumps
from urllib.parse import urlparse
import logging
import os
import ssl
from json import dumps as json_dumps
from json import loads as json_loads
from urllib.parse import urlparse
import pytest
from sanic.exceptions import ServerError
from sanic.response import json, text
from sanic.request import DEFAULT_HTTP_CONTENT_TYPE
from sanic.response import json, text
from sanic.testing import HOST, PORT
# ------------------------------------------------------------ #
# GET
# ------------------------------------------------------------ #
@ -266,6 +267,7 @@ def test_post_json(app):
'/', data=json_dumps(payload), headers=headers)
assert request.json.get('test') == 'OK'
assert request.json.get('test') == 'OK' # for request.parsed_json
assert response.text == 'OK'
@ -282,6 +284,7 @@ def test_post_form_urlencoded(app):
headers=headers)
assert request.form.get('test') == 'OK'
assert request.form.get('test') == 'OK' # For request.parsed_form
@pytest.mark.parametrize(
@ -471,8 +474,15 @@ def test_request_multipart_file_with_json_content_type(app):
async def post(request):
return text("OK")
payload = '------sanic\r\nContent-Disposition: form-data; name="file"; filename="test.json"' \
'\r\nContent-Type: application/json\r\n\r\n\r\n------sanic--'
payload = (
'------sanic\r\n'
'Content-Disposition: form-data; name="file"; filename="test.json"\r\n'
'Content-Type: application/json\r\n'
'Content-Length: 0'
'\r\n'
'\r\n'
'------sanic--'
)
headers = {'content-type': 'multipart/form-data; boundary=------sanic'}
@ -480,6 +490,59 @@ def test_request_multipart_file_with_json_content_type(app):
assert request.files.get('file').type == 'application/json'
def test_request_multipart_file_without_field_name(app, caplog):
@app.route("/", methods=["POST"])
async def post(request):
return text("OK")
payload = (
'------sanic\r\nContent-Disposition: form-data; filename="test.json"'
'\r\nContent-Type: application/json\r\n\r\n\r\n------sanic--'
)
headers = {'content-type': 'multipart/form-data; boundary=------sanic'}
request, _ = app.test_client.post(data=payload, headers=headers, debug=True)
with caplog.at_level(logging.DEBUG):
request.form
assert caplog.record_tuples[-1] == ('sanic.root', logging.DEBUG,
"Form-data field does not have a 'name' parameter "
"in the Content-Disposition header"
)
def test_request_multipart_file_duplicate_filed_name(app):
@app.route("/", methods=["POST"])
async def post(request):
return text("OK")
payload = (
'--e73ffaa8b1b2472b8ec848de833cb05b\r\n'
'Content-Disposition: form-data; name="file"\r\n'
'Content-Type: application/octet-stream\r\n'
'Content-Length: 15\r\n'
'\r\n'
'{"test":"json"}\r\n'
'--e73ffaa8b1b2472b8ec848de833cb05b\r\n'
'Content-Disposition: form-data; name="file"\r\n'
'Content-Type: application/octet-stream\r\n'
'Content-Length: 15\r\n'
'\r\n'
'{"test":"json2"}\r\n'
'--e73ffaa8b1b2472b8ec848de833cb05b--\r\n'
)
headers = {
'Content-Type': 'multipart/form-data; boundary=e73ffaa8b1b2472b8ec848de833cb05b'
}
request, _ = app.test_client.post(data=payload, headers=headers, debug=True)
assert request.form.getlist('file') == ['{"test":"json"}', '{"test":"json2"}']
def test_request_multipart_with_multiple_files_and_type(app):
@app.route("/", methods=["POST"])
async def post(request):
@ -495,3 +558,150 @@ def test_request_multipart_with_multiple_files_and_type(app):
assert len(request.files.getlist('file')) == 2
assert request.files.getlist('file')[0].type == 'application/json'
assert request.files.getlist('file')[1].type == 'application/pdf'
def test_request_repr(app):
@app.get('/')
def handler(request):
return text('pass')
request, response = app.test_client.get('/')
assert repr(request) == '<Request: GET />'
request.method = None
assert repr(request) == '<Request>'
def test_request_bool(app):
@app.get('/')
def handler(request):
return text('pass')
request, response = app.test_client.get('/')
assert bool(request)
request.transport = False
assert not bool(request)
def test_request_parsing_form_failed(app, caplog):
@app.route('/', methods=['POST'])
async def handler(request):
return text('OK')
payload = 'test=OK'
headers = {'content-type': 'multipart/form-data'}
request, response = app.test_client.post('/', data=payload, headers=headers)
with caplog.at_level(logging.ERROR):
request.form
assert caplog.record_tuples[-1] == ('sanic.error', logging.ERROR, 'Failed when parsing form')
def test_request_args_no_query_string(app):
@app.get('/')
def handler(request):
return text('pass')
request, response = app.test_client.get('/')
assert request.args == {}
def test_request_raw_args(app):
params = {'test': 'OK'}
@app.get('/')
def handler(request):
return text('pass')
request, response = app.test_client.get('/', params=params)
assert request.raw_args == params
def test_request_cookies(app):
cookies = {'test': 'OK'}
@app.get('/')
def handler(request):
return text('OK')
request, response = app.test_client.get('/', cookies=cookies)
assert request.cookies == cookies
assert request.cookies == cookies # For request._cookies
def test_request_cookies_without_cookies(app):
@app.get('/')
def handler(request):
return text('OK')
request, response = app.test_client.get('/')
assert request.cookies == {}
def test_request_port(app):
@app.get('/')
def handler(request):
return text('OK')
request, response = app.test_client.get('/')
port = request.port
assert isinstance(port, int)
delattr(request, '_socket')
delattr(request, '_port')
port = request.port
assert isinstance(port, int)
assert hasattr(request, '_socket')
assert hasattr(request, '_port')
def test_request_socket(app):
@app.get('/')
def handler(request):
return text('OK')
request, response = app.test_client.get('/')
socket = request.socket
assert isinstance(socket, tuple)
ip = socket[0]
port = socket[1]
assert ip == request.ip
assert port == request.port
delattr(request, '_socket')
socket = request.socket
assert isinstance(socket, tuple)
assert hasattr(request, '_socket')
def test_request_form_invalid_content_type(app):
@app.route("/", methods=["POST"])
async def post(request):
return text("OK")
request, response = app.test_client.post('/', json={'test': 'OK'})
assert request.form == {}

24
tests/test_response.py
View File

@ -264,6 +264,29 @@ def test_stream_response_writes_correct_content_to_transport(streaming_app):
streaming_app.run(host=HOST, port=PORT)
def test_stream_response_with_cookies(app):
@app.route("/")
async def test(request):
response = stream(sample_streaming_fn, content_type='text/csv')
response.cookies['test'] = 'modified'
response.cookies['test'] = 'pass'
return response
request, response = app.test_client.get('/')
assert response.cookies['test'].value == 'pass'
def test_stream_response_without_cookies(app):
@app.route("/")
async def test(request):
return stream(sample_streaming_fn, content_type='text/csv')
request, response = app.test_client.get('/')
assert response.cookies == {}
@pytest.fixture
def static_file_directory():
"""The static directory to serve"""
@ -451,6 +474,7 @@ def test_file_stream_response_range(app, file_name, static_file_directory, size,
assert 'Content-Range' in response.headers
assert response.headers['Content-Range'] == 'bytes {}-{}/{}'.format(range.start, range.end, range.total)
def test_raw_response(app):
@app.get('/test')

22
tests/test_routes.py
View File

@ -468,6 +468,7 @@ def test_websocket_route(app, url):
@app.websocket(url)
async def handler(request, ws):
assert request.scheme == 'ws'
assert ws.subprotocol is None
ev.set()
@ -785,8 +786,11 @@ def test_remove_dynamic_route(app):
def test_remove_inexistent_route(app):
with pytest.raises(RouteDoesNotExist):
app.remove_route('/test')
uri = '/test'
with pytest.raises(RouteDoesNotExist) as excinfo:
app.remove_route(uri)
assert str(excinfo.value) == 'Route was not registered: {}'.format(uri)
def test_removing_slash(app):
@ -963,3 +967,17 @@ def test_route_raise_ParameterNameConflicts(app):
@app.get('/api/v1/<user>/<user>/')
def handler(request, user):
return text('OK')
def test_route_invalid_host(app):
host = 321
with pytest.raises(ValueError) as excinfo:
@app.get('/test', host=host)
def handler(request):
return text('pass')
assert str(excinfo.value) == (
"Expected either string or Iterable of "
"host strings, not {!r}"
).format(host)

79
tests/test_static.py
View File

@ -205,6 +205,69 @@ def test_static_content_range_error(app, file_name, static_file_directory):
len(get_file_content(static_file_directory, file_name)),)
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_content_range_invalid_unit(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)
unit = 'bit'
headers = {
'Range': '{}=1-0'.format(unit)
}
request, response = app.test_client.get('/testing.file', headers=headers)
assert response.status == 416
assert response.text == "Error: {} is not a valid Range Type".format(unit)
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_content_range_invalid_start(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)
start = 'start'
headers = {
'Range': 'bytes={}-0'.format(start)
}
request, response = app.test_client.get('/testing.file', headers=headers)
assert response.status == 416
assert response.text == "Error: '{}' is invalid for Content Range".format(start)
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_content_range_invalid_end(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)
end = 'end'
headers = {
'Range': 'bytes=1-{}'.format(end)
}
request, response = app.test_client.get('/testing.file', headers=headers)
assert response.status == 416
assert response.text == "Error: '{}' is invalid for Content Range".format(end)
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_content_range_invalid_parameters(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)
headers = {
'Range': 'bytes=-'
}
request, response = app.test_client.get('/testing.file', headers=headers)
assert response.status == 416
assert response.text == "Error: Invalid for Content Range parameters"
@pytest.mark.parametrize('file_name',
['test.file', 'decode me.txt', 'python.png'])
def test_static_file_specified_host(app, static_file_directory, file_name):
@ -274,3 +337,19 @@ def test_static_name(app, static_file_directory, static_name, file_name):
request, response = app.test_client.get('/static/{}'.format(file_name))
assert response.status == 200
@pytest.mark.parametrize('file_name',
['test.file'])
def test_static_remove_route(app, static_file_directory, file_name):
app.static(
'/testing.file',
get_file_path(static_file_directory, file_name)
)
request, response = app.test_client.get('/testing.file')
assert response.status == 200
app.remove_route('/testing.file')
request, response = app.test_client.get('/testing.file')
assert response.status == 404

15
tests/test_vhosts.py
View File

@ -51,3 +51,18 @@ def test_vhosts_with_defaults(app):
request, response = app.test_client.get('/')
assert response.text == "default"
def test_remove_vhost_route(app):
@app.route('/', host="example.com")
async def handler1(request):
return text("You're at example.com!")
headers = {"Host": "example.com"}
request, response = app.test_client.get('/', headers=headers)
assert response.status == 200
app.remove_route('/', host="example.com")
request, response = app.test_client.get('/', headers=headers)
assert response.status == 404