Merge branch 'master' into add_register_sys_signals_flag
This commit is contained in:
@@ -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
32
tests/test_custom_protocol.py
Normal 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'
|
||||
@@ -1,51 +1,93 @@
|
||||
import pytest
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
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
|
||||
soup = BeautifulSoup(response.body, 'html.parser')
|
||||
assert soup.h1.text == 'Internal Server Error'
|
||||
|
||||
message = " ".join(soup.p.text.split())
|
||||
assert message == (
|
||||
"The server encountered an internal error and "
|
||||
"cannot complete your request.")
|
||||
|
||||
|
||||
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'
|
||||
|
||||
@@ -2,6 +2,7 @@ from sanic import Sanic
|
||||
from sanic.response import text
|
||||
from sanic.exceptions import InvalidUsage, ServerError, NotFound
|
||||
from sanic.utils import sanic_endpoint_test
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
exception_handler_app = Sanic('test_exception_handler')
|
||||
|
||||
@@ -21,6 +22,12 @@ def handler_3(request):
|
||||
raise NotFound("OK")
|
||||
|
||||
|
||||
@exception_handler_app.route('/4')
|
||||
def handler_4(request):
|
||||
foo = bar
|
||||
return text(foo)
|
||||
|
||||
|
||||
@exception_handler_app.exception(NotFound, ServerError)
|
||||
def handler_exception(request, exception):
|
||||
return text("OK")
|
||||
@@ -47,3 +54,20 @@ def test_text_exception__handler():
|
||||
exception_handler_app, uri='/random')
|
||||
assert response.status == 200
|
||||
assert response.text == 'OK'
|
||||
|
||||
|
||||
def test_html_traceback_output_in_debug_mode():
|
||||
request, response = sanic_endpoint_test(
|
||||
exception_handler_app, uri='/4', debug=True)
|
||||
assert response.status == 500
|
||||
soup = BeautifulSoup(response.body, 'html.parser')
|
||||
html = str(soup)
|
||||
|
||||
assert 'response = handler(request, *args, **kwargs)' in html
|
||||
assert 'handler_4' in html
|
||||
assert 'foo = bar' in html
|
||||
|
||||
summary_text = " ".join(soup.select('.summary')[0].text.split())
|
||||
assert (
|
||||
"NameError: name 'bar' "
|
||||
"is not defined while handling uri /4") == summary_text
|
||||
|
||||
@@ -19,7 +19,7 @@ def test_log():
|
||||
stream=log_stream
|
||||
)
|
||||
log = logging.getLogger()
|
||||
app = Sanic('test_logging', logger=True)
|
||||
app = Sanic('test_logging')
|
||||
@app.route('/')
|
||||
def handler(request):
|
||||
log.info('hello world')
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
from json import loads as json_loads, dumps as json_dumps
|
||||
from sanic import Sanic
|
||||
from sanic.response import json, text
|
||||
from sanic.response import json, text, redirect
|
||||
from sanic.utils import sanic_endpoint_test
|
||||
from sanic.exceptions import ServerError
|
||||
|
||||
import pytest
|
||||
|
||||
# ------------------------------------------------------------ #
|
||||
# GET
|
||||
@@ -33,6 +34,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 +73,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 +118,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
|
||||
# ------------------------------------------------------------ #
|
||||
@@ -145,3 +189,73 @@ def test_post_form_multipart_form_data():
|
||||
request, response = sanic_endpoint_test(app, data=payload, headers=headers)
|
||||
|
||||
assert request.form.get('test') == 'OK'
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def redirect_app():
|
||||
app = Sanic('test_redirection')
|
||||
|
||||
@app.route('/redirect_init')
|
||||
async def redirect_init(request):
|
||||
return redirect("/redirect_target")
|
||||
|
||||
@app.route('/redirect_init_with_301')
|
||||
async def redirect_init_with_301(request):
|
||||
return redirect("/redirect_target", status=301)
|
||||
|
||||
@app.route('/redirect_target')
|
||||
async def redirect_target(request):
|
||||
return text('OK')
|
||||
|
||||
return app
|
||||
|
||||
|
||||
def test_redirect_default_302(redirect_app):
|
||||
"""
|
||||
We expect a 302 default status code and the headers to be set.
|
||||
"""
|
||||
request, response = sanic_endpoint_test(
|
||||
redirect_app, method="get",
|
||||
uri="/redirect_init",
|
||||
allow_redirects=False)
|
||||
|
||||
assert response.status == 302
|
||||
assert response.headers["Location"] == "/redirect_target"
|
||||
assert response.headers["Content-Type"] == 'text/html; charset=utf-8'
|
||||
|
||||
|
||||
def test_redirect_headers_none(redirect_app):
|
||||
request, response = sanic_endpoint_test(
|
||||
redirect_app, method="get",
|
||||
uri="/redirect_init",
|
||||
headers=None,
|
||||
allow_redirects=False)
|
||||
|
||||
assert response.status == 302
|
||||
assert response.headers["Location"] == "/redirect_target"
|
||||
|
||||
|
||||
def test_redirect_with_301(redirect_app):
|
||||
"""
|
||||
Test redirection with a different status code.
|
||||
"""
|
||||
request, response = sanic_endpoint_test(
|
||||
redirect_app, method="get",
|
||||
uri="/redirect_init_with_301",
|
||||
allow_redirects=False)
|
||||
|
||||
assert response.status == 301
|
||||
assert response.headers["Location"] == "/redirect_target"
|
||||
|
||||
|
||||
def test_get_then_redirect_follow_redirect(redirect_app):
|
||||
"""
|
||||
With `allow_redirects` we expect a 200.
|
||||
"""
|
||||
response = sanic_endpoint_test(
|
||||
redirect_app, method="get",
|
||||
uri="/redirect_init", gather_request=False,
|
||||
allow_redirects=True)
|
||||
|
||||
assert response.status == 200
|
||||
assert response.text == 'OK'
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -10,6 +10,84 @@ from sanic.utils import sanic_endpoint_test
|
||||
# UTF-8
|
||||
# ------------------------------------------------------------ #
|
||||
|
||||
def test_shorthand_routes_get():
|
||||
app = Sanic('test_shorhand_routes_get')
|
||||
|
||||
@app.get('/get')
|
||||
def handler(request):
|
||||
return text('OK')
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/get', method='get')
|
||||
assert response.text == 'OK'
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/get', method='post')
|
||||
assert response.status == 405
|
||||
|
||||
def test_shorthand_routes_post():
|
||||
app = Sanic('test_shorhand_routes_post')
|
||||
|
||||
@app.post('/post')
|
||||
def handler(request):
|
||||
return text('OK')
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/post', method='post')
|
||||
assert response.text == 'OK'
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/post', method='get')
|
||||
assert response.status == 405
|
||||
|
||||
def test_shorthand_routes_put():
|
||||
app = Sanic('test_shorhand_routes_put')
|
||||
|
||||
@app.put('/put')
|
||||
def handler(request):
|
||||
return text('OK')
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/put', method='put')
|
||||
assert response.text == 'OK'
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/put', method='get')
|
||||
assert response.status == 405
|
||||
|
||||
def test_shorthand_routes_patch():
|
||||
app = Sanic('test_shorhand_routes_patch')
|
||||
|
||||
@app.patch('/patch')
|
||||
def handler(request):
|
||||
return text('OK')
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/patch', method='patch')
|
||||
assert response.text == 'OK'
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/patch', method='get')
|
||||
assert response.status == 405
|
||||
|
||||
def test_shorthand_routes_head():
|
||||
app = Sanic('test_shorhand_routes_head')
|
||||
|
||||
@app.head('/head')
|
||||
def handler(request):
|
||||
return text('OK')
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/head', method='head')
|
||||
assert response.status == 200
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/head', method='get')
|
||||
assert response.status == 405
|
||||
|
||||
def test_shorthand_routes_options():
|
||||
app = Sanic('test_shorhand_routes_options')
|
||||
|
||||
@app.options('/options')
|
||||
def handler(request):
|
||||
return text('OK')
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/options', method='options')
|
||||
assert response.status == 200
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/options', method='get')
|
||||
assert response.status == 405
|
||||
|
||||
def test_static_routes():
|
||||
app = Sanic('test_dynamic_route')
|
||||
|
||||
@@ -356,3 +434,174 @@ 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
|
||||
|
||||
|
||||
def test_overload_routes():
|
||||
app = Sanic('test_dynamic_route')
|
||||
|
||||
@app.route('/overload', methods=['GET'])
|
||||
async def handler1(request):
|
||||
return text('OK1')
|
||||
|
||||
@app.route('/overload', methods=['POST', 'PUT'])
|
||||
async def handler2(request):
|
||||
return text('OK2')
|
||||
|
||||
request, response = sanic_endpoint_test(app, 'get', uri='/overload')
|
||||
assert response.text == 'OK1'
|
||||
|
||||
request, response = sanic_endpoint_test(app, 'post', uri='/overload')
|
||||
assert response.text == 'OK2'
|
||||
|
||||
request, response = sanic_endpoint_test(app, 'put', uri='/overload')
|
||||
assert response.text == 'OK2'
|
||||
|
||||
request, response = sanic_endpoint_test(app, 'delete', uri='/overload')
|
||||
assert response.status == 405
|
||||
|
||||
with pytest.raises(RouteExists):
|
||||
@app.route('/overload', methods=['PUT', 'DELETE'])
|
||||
async def handler3(request):
|
||||
return text('Duplicated')
|
||||
|
||||
|
||||
def test_unmergeable_overload_routes():
|
||||
app = Sanic('test_dynamic_route')
|
||||
|
||||
@app.route('/overload_whole', methods=None)
|
||||
async def handler1(request):
|
||||
return text('OK1')
|
||||
|
||||
with pytest.raises(RouteExists):
|
||||
@app.route('/overload_whole', methods=['POST', 'PUT'])
|
||||
async def handler2(request):
|
||||
return text('Duplicated')
|
||||
|
||||
request, response = sanic_endpoint_test(app, 'get', uri='/overload_whole')
|
||||
assert response.text == 'OK1'
|
||||
|
||||
request, response = sanic_endpoint_test(app, 'post', uri='/overload_whole')
|
||||
assert response.text == 'OK1'
|
||||
|
||||
|
||||
@app.route('/overload_part', methods=['GET'])
|
||||
async def handler1(request):
|
||||
return text('OK1')
|
||||
|
||||
with pytest.raises(RouteExists):
|
||||
@app.route('/overload_part')
|
||||
async def handler2(request):
|
||||
return text('Duplicated')
|
||||
|
||||
request, response = sanic_endpoint_test(app, 'get', uri='/overload_part')
|
||||
assert response.text == 'OK1'
|
||||
|
||||
request, response = sanic_endpoint_test(app, 'post', uri='/overload_part')
|
||||
assert response.status == 405
|
||||
|
||||
59
tests/test_server_events.py
Normal file
59
tests/test_server_events.py
Normal 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()
|
||||
39
tests/test_vhosts.py
Normal file
39
tests/test_vhosts.py
Normal file
@@ -0,0 +1,39 @@
|
||||
from sanic import Sanic
|
||||
from sanic.response import json, text
|
||||
from sanic.utils import sanic_endpoint_test
|
||||
|
||||
|
||||
def test_vhosts():
|
||||
app = Sanic('test_vhosts')
|
||||
|
||||
@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!"
|
||||
|
||||
|
||||
def test_vhosts_with_list():
|
||||
app = Sanic('test_vhosts')
|
||||
|
||||
@app.route('/', host=["hello.com", "world.com"])
|
||||
async def handler(request):
|
||||
return text("Hello, world!")
|
||||
|
||||
headers = {"Host": "hello.com"}
|
||||
request, response = sanic_endpoint_test(app, headers=headers)
|
||||
assert response.text == "Hello, world!"
|
||||
|
||||
headers = {"Host": "world.com"}
|
||||
request, response = sanic_endpoint_test(app, headers=headers)
|
||||
assert response.text == "Hello, world!"
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user