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

Merge branch 'master' into improved_config

Browse Source
This commit is contained in:
Tim Mundt
2017-01-13 12:34:56 +01:00
parent 2d4512cd1c cf60ebd988
commit 0b9094d348
29 changed files with 870 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
tests/test_blueprints.py
View File

@@ -59,6 +59,71 @@ def test_several_bp_with_url_prefix():
request, response = sanic_endpoint_test(app, uri='/test2/')
assert response.text == 'Hello2'
def test_bp_with_host():
app = Sanic('test_bp_host')
bp = Blueprint('test_bp_host', url_prefix='/test1', host="example.com")
@bp.route('/')
def handler(request):
return text('Hello')
@bp.route('/', host="sub.example.com")
def handler(request):
return text('Hello subdomain!')
app.blueprint(bp)
headers = {"Host": "example.com"}
request, response = sanic_endpoint_test(app, uri='/test1/',
headers=headers)
assert response.text == 'Hello'
headers = {"Host": "sub.example.com"}
request, response = sanic_endpoint_test(app, uri='/test1/',
headers=headers)
assert response.text == 'Hello subdomain!'
def test_several_bp_with_host():
app = Sanic('test_text')
bp = Blueprint('test_text',
url_prefix='/test',
host="example.com")
bp2 = Blueprint('test_text2',
url_prefix='/test',
host="sub.example.com")
@bp.route('/')
def handler(request):
return text('Hello')
@bp2.route('/')
def handler2(request):
return text('Hello2')
@bp2.route('/other/')
def handler2(request):
return text('Hello3')
app.blueprint(bp)
app.blueprint(bp2)
assert bp.host == "example.com"
headers = {"Host": "example.com"}
request, response = sanic_endpoint_test(app, uri='/test/',
headers=headers)
assert response.text == 'Hello'
assert bp2.host == "sub.example.com"
headers = {"Host": "sub.example.com"}
request, response = sanic_endpoint_test(app, uri='/test/',
headers=headers)
assert response.text == 'Hello2'
request, response = sanic_endpoint_test(app, uri='/test/other/',
headers=headers)
assert response.text == 'Hello3'
def test_bp_middleware():
app = Sanic('test_middleware')
@@ -162,4 +227,4 @@ def test_bp_static():
request, response = sanic_endpoint_test(app, uri='/testing.file')
assert response.status == 200
assert response.body == current_file_contents
assert response.body == current_file_contents

32
tests/test_custom_protocol.py Normal file
View File

@@ -0,0 +1,32 @@
from sanic import Sanic
from sanic.server import HttpProtocol
from sanic.response import text
from sanic.utils import sanic_endpoint_test
app = Sanic('test_custom_porotocol')
class CustomHttpProtocol(HttpProtocol):
def write_response(self, response):
if isinstance(response, str):
response = text(response)
self.transport.write(
response.output(self.request.version)
)
self.transport.close()
@app.route('/1')
async def handler_1(request):
return 'OK'
def test_use_custom_protocol():
server_kwargs = {
'protocol': CustomHttpProtocol
}
request, response = sanic_endpoint_test(app, uri='/1',
server_kwargs=server_kwargs)
assert response.status == 200
assert response.text == 'OK'

87
tests/test_exceptions.py
View File

@@ -1,51 +1,86 @@
import pytest
from sanic import Sanic
from sanic.response import text
from sanic.exceptions import InvalidUsage, ServerError, NotFound
from sanic.utils import sanic_endpoint_test
# ------------------------------------------------------------ #
# GET
# ------------------------------------------------------------ #
exception_app = Sanic('test_exceptions')
class SanicExceptionTestException(Exception):
pass
@exception_app.route('/')
def handler(request):
return text('OK')
@pytest.fixture(scope='module')
def exception_app():
app = Sanic('test_exceptions')
@app.route('/')
def handler(request):
return text('OK')
@app.route('/error')
def handler_error(request):
raise ServerError("OK")
@app.route('/404')
def handler_404(request):
raise NotFound("OK")
@app.route('/invalid')
def handler_invalid(request):
raise InvalidUsage("OK")
@app.route('/divide_by_zero')
def handle_unhandled_exception(request):
1 / 0
@app.route('/error_in_error_handler_handler')
def custom_error_handler(request):
raise SanicExceptionTestException('Dummy message!')
@app.exception(SanicExceptionTestException)
def error_in_error_handler_handler(request, exception):
1 / 0
return app
@exception_app.route('/error')
def handler_error(request):
raise ServerError("OK")
@exception_app.route('/404')
def handler_404(request):
raise NotFound("OK")
@exception_app.route('/invalid')
def handler_invalid(request):
raise InvalidUsage("OK")
def test_no_exception():
def test_no_exception(exception_app):
"""Test that a route works without an exception"""
request, response = sanic_endpoint_test(exception_app)
assert response.status == 200
assert response.text == 'OK'
def test_server_error_exception():
def test_server_error_exception(exception_app):
"""Test the built-in ServerError exception works"""
request, response = sanic_endpoint_test(exception_app, uri='/error')
assert response.status == 500
def test_invalid_usage_exception():
def test_invalid_usage_exception(exception_app):
"""Test the built-in InvalidUsage exception works"""
request, response = sanic_endpoint_test(exception_app, uri='/invalid')
assert response.status == 400
def test_not_found_exception():
def test_not_found_exception(exception_app):
"""Test the built-in NotFound exception works"""
request, response = sanic_endpoint_test(exception_app, uri='/404')
assert response.status == 404
def test_handled_unhandled_exception(exception_app):
"""Test that an exception not built into sanic is handled"""
request, response = sanic_endpoint_test(
exception_app, uri='/divide_by_zero')
assert response.status == 500
assert response.body == b'An error occurred while generating the response'
def test_exception_in_exception_handler(exception_app):
"""Test that an exception thrown in an error handler is handled"""
request, response = sanic_endpoint_test(
exception_app, uri='/error_in_error_handler_handler')
assert response.status == 500
assert response.body == b'An error occurred while handling an error'

34
tests/test_multiprocessing.py
View File

@@ -1,7 +1,9 @@
from multiprocessing import Array, Event, Process
from time import sleep
from time import sleep, time
from ujson import loads as json_loads
import pytest
from sanic import Sanic
from sanic.response import json
from sanic.utils import local_request, HOST, PORT
@@ -13,8 +15,9 @@ from sanic.utils import local_request, HOST, PORT
# TODO: Figure out why this freezes on pytest but not when
# executed via interpreter
def skip_test_multiprocessing():
@pytest.mark.skip(
reason="Freezes with pytest not on interpreter")
def test_multiprocessing():
app = Sanic('test_json')
response = Array('c', 50)
@@ -51,3 +54,28 @@ def skip_test_multiprocessing():
raise ValueError("Expected JSON response but got '{}'".format(response))
assert results.get('test') == True
@pytest.mark.skip(
reason="Freezes with pytest not on interpreter")
def test_drain_connections():
app = Sanic('test_json')
@app.route('/')
async def handler(request):
return json({"test": True})
stop_event = Event()
async def after_start(*args, **kwargs):
http_response = await local_request('get', '/')
stop_event.set()
start = time()
app.serve_multiple({
'host': HOST,
'port': PORT,
'after_start': after_start,
'request_handler': app.handle_request,
}, workers=2, stop_event=stop_event)
end = time()
assert end - start < 0.05

47
tests/test_requests.py
View File

@@ -33,6 +33,31 @@ def test_text():
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')
@@ -47,8 +72,8 @@ def test_invalid_response():
request, response = sanic_endpoint_test(app)
assert response.status == 500
assert response.text == "Internal Server Error."
def test_json():
app = Sanic('test_json')
@@ -92,6 +117,24 @@ def test_query_string():
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
# ------------------------------------------------------------ #

109
tests/test_routes.py
View File

@@ -2,7 +2,7 @@ import pytest
from sanic import Sanic
from sanic.response import text
from sanic.router import RouteExists
from sanic.router import RouteExists, RouteDoesNotExist
from sanic.utils import sanic_endpoint_test
@@ -356,3 +356,110 @@ def test_add_route_method_not_allowed():
request, response = sanic_endpoint_test(app, method='post', uri='/test')
assert response.status == 405
def test_remove_static_route():
app = Sanic('test_remove_static_route')
async def handler1(request):
return text('OK1')
async def handler2(request):
return text('OK2')
app.add_route(handler1, '/test')
app.add_route(handler2, '/test2')
request, response = sanic_endpoint_test(app, uri='/test')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/test2')
assert response.status == 200
app.remove_route('/test')
app.remove_route('/test2')
request, response = sanic_endpoint_test(app, uri='/test')
assert response.status == 404
request, response = sanic_endpoint_test(app, uri='/test2')
assert response.status == 404
def test_remove_dynamic_route():
app = Sanic('test_remove_dynamic_route')
async def handler(request, name):
return text('OK')
app.add_route(handler, '/folder/<name>')
request, response = sanic_endpoint_test(app, uri='/folder/test123')
assert response.status == 200
app.remove_route('/folder/<name>')
request, response = sanic_endpoint_test(app, uri='/folder/test123')
assert response.status == 404
def test_remove_inexistent_route():
app = Sanic('test_remove_inexistent_route')
with pytest.raises(RouteDoesNotExist):
app.remove_route('/test')
def test_remove_unhashable_route():
app = Sanic('test_remove_unhashable_route')
async def handler(request, unhashable):
return text('OK')
app.add_route(handler, '/folder/<unhashable:[A-Za-z0-9/]+>/end/')
request, response = sanic_endpoint_test(app, uri='/folder/test/asdf/end/')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test///////end/')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test/end/')
assert response.status == 200
app.remove_route('/folder/<unhashable:[A-Za-z0-9/]+>/end/')
request, response = sanic_endpoint_test(app, uri='/folder/test/asdf/end/')
assert response.status == 404
request, response = sanic_endpoint_test(app, uri='/folder/test///////end/')
assert response.status == 404
request, response = sanic_endpoint_test(app, uri='/folder/test/end/')
assert response.status == 404
def test_remove_route_without_clean_cache():
app = Sanic('test_remove_static_route')
async def handler(request):
return text('OK')
app.add_route(handler, '/test')
request, response = sanic_endpoint_test(app, uri='/test')
assert response.status == 200
app.remove_route('/test', clean_cache=True)
request, response = sanic_endpoint_test(app, uri='/test')
assert response.status == 404
app.add_route(handler, '/test')
request, response = sanic_endpoint_test(app, uri='/test')
assert response.status == 200
app.remove_route('/test', clean_cache=False)
request, response = sanic_endpoint_test(app, uri='/test')
assert response.status == 200

59
tests/test_server_events.py Normal file
View File

@@ -0,0 +1,59 @@
from io import StringIO
from random import choice
from string import ascii_letters
import signal
import pytest
from sanic import Sanic
AVAILABLE_LISTENERS = [
'before_start',
'after_start',
'before_stop',
'after_stop'
]
def create_listener(listener_name, in_list):
async def _listener(app, loop):
print('DEBUG MESSAGE FOR PYTEST for {}'.format(listener_name))
in_list.insert(0, app.name + listener_name)
return _listener
def start_stop_app(random_name_app, **run_kwargs):
def stop_on_alarm(signum, frame):
raise KeyboardInterrupt('SIGINT for sanic to stop gracefully')
signal.signal(signal.SIGALRM, stop_on_alarm)
signal.alarm(1)
try:
random_name_app.run(**run_kwargs)
except KeyboardInterrupt:
pass
@pytest.mark.parametrize('listener_name', AVAILABLE_LISTENERS)
def test_single_listener(listener_name):
"""Test that listeners on their own work"""
random_name_app = Sanic(''.join(
[choice(ascii_letters) for _ in range(choice(range(5, 10)))]))
output = list()
start_stop_app(
random_name_app,
**{listener_name: create_listener(listener_name, output)})
assert random_name_app.name + listener_name == output.pop()
def test_all_listeners():
random_name_app = Sanic(''.join(
[choice(ascii_letters) for _ in range(choice(range(5, 10)))]))
output = list()
start_stop_app(
random_name_app,
**{listener_name: create_listener(listener_name, output)
for listener_name in AVAILABLE_LISTENERS})
for listener_name in AVAILABLE_LISTENERS:
assert random_name_app.name + listener_name == output.pop()

23
tests/test_vhosts.py Normal file
View File

@@ -0,0 +1,23 @@
from sanic import Sanic
from sanic.response import json, text
from sanic.utils import sanic_endpoint_test
def test_vhosts():
app = Sanic('test_text')
@app.route('/', host="example.com")
async def handler(request):
return text("You're at example.com!")
@app.route('/', host="subdomain.example.com")
async def handler(request):
return text("You're at subdomain.example.com!")
headers = {"Host": "example.com"}
request, response = sanic_endpoint_test(app, headers=headers)
assert response.text == "You're at example.com!"
headers = {"Host": "subdomain.example.com"}
request, response = sanic_endpoint_test(app, headers=headers)
assert response.text == "You're at subdomain.example.com!"

55
tests/test_views.py
View File

@@ -26,7 +26,7 @@ def test_methods():
def delete(self, request):
return text('I am delete method')
app.add_route(DummyView(), '/')
app.add_route(DummyView.as_view(), '/')
request, response = sanic_endpoint_test(app, method="get")
assert response.text == 'I am get method'
@@ -48,7 +48,7 @@ def test_unexisting_methods():
def get(self, request):
return text('I am get method')
app.add_route(DummyView(), '/')
app.add_route(DummyView.as_view(), '/')
request, response = sanic_endpoint_test(app, method="get")
assert response.text == 'I am get method'
request, response = sanic_endpoint_test(app, method="post")
@@ -63,7 +63,7 @@ def test_argument_methods():
def get(self, request, my_param_here):
return text('I am get method with %s' % my_param_here)
app.add_route(DummyView(), '/<my_param_here>')
app.add_route(DummyView.as_view(), '/<my_param_here>')
request, response = sanic_endpoint_test(app, uri='/test123')
@@ -79,7 +79,7 @@ def test_with_bp():
def get(self, request):
return text('I am get method')
bp.add_route(DummyView(), '/')
bp.add_route(DummyView.as_view(), '/')
app.blueprint(bp)
request, response = sanic_endpoint_test(app)
@@ -96,7 +96,7 @@ def test_with_bp_with_url_prefix():
def get(self, request):
return text('I am get method')
bp.add_route(DummyView(), '/')
bp.add_route(DummyView.as_view(), '/')
app.blueprint(bp)
request, response = sanic_endpoint_test(app, uri='/test1/')
@@ -112,7 +112,7 @@ def test_with_middleware():
def get(self, request):
return text('I am get method')
app.add_route(DummyView(), '/')
app.add_route(DummyView.as_view(), '/')
results = []
@@ -145,7 +145,7 @@ def test_with_middleware_response():
def get(self, request):
return text('I am get method')
app.add_route(DummyView(), '/')
app.add_route(DummyView.as_view(), '/')
request, response = sanic_endpoint_test(app)
@@ -153,3 +153,44 @@ def test_with_middleware_response():
assert type(results[0]) is Request
assert type(results[1]) is Request
assert issubclass(type(results[2]), HTTPResponse)
def test_with_custom_class_methods():
app = Sanic('test_with_custom_class_methods')
class DummyView(HTTPMethodView):
global_var = 0
def _iternal_method(self):
self.global_var += 10
def get(self, request):
self._iternal_method()
return text('I am get method and global var is {}'.format(self.global_var))
app.add_route(DummyView.as_view(), '/')
request, response = sanic_endpoint_test(app, method="get")
assert response.text == 'I am get method and global var is 10'
def test_with_decorator():
app = Sanic('test_with_decorator')
results = []
def stupid_decorator(view):
def decorator(*args, **kwargs):
results.append(1)
return view(*args, **kwargs)
return decorator
class DummyView(HTTPMethodView):
decorators = [stupid_decorator]
def get(self, request):
return text('I am get method')
app.add_route(DummyView.as_view(), '/')
request, response = sanic_endpoint_test(app, method="get")
assert response.text == 'I am get method'
assert results[0] == 1