Reformatted code to use spaces instead of tabs

This commit is contained in:
Channel Cat 2016-10-14 19:53:49 -07:00
parent 67db0bcbf3
commit 254861bc37
21 changed files with 344 additions and 266 deletions

View File

@ -3,8 +3,10 @@ from sanic.response import json
app = Sanic(__name__) app = Sanic(__name__)
@app.route("/") @app.route("/")
async def test(request): async def test(request):
return json({ "test": True }) return json({"test": True})
app.run(host="0.0.0.0", port=8000)
app.run(host="0.0.0.0", port=8000)

View File

@ -5,29 +5,35 @@ from sanic.exceptions import ServerError
app = Sanic(__name__) app = Sanic(__name__)
@app.route("/") @app.route("/")
async def test_async(request): async def test_async(request):
return json({ "test": True }) return json({"test": True})
@app.route("/sync", methods=['GET', 'POST']) @app.route("/sync", methods=['GET', 'POST'])
def test_sync(request): def test_sync(request):
return json({ "test": True }) return json({"test": True})
@app.route("/dynamic/<name>/<id:int>") @app.route("/dynamic/<name>/<id:int>")
def test_params(request, name, id): def test_params(request, name, id):
return text("yeehaww {} {}".format(name, id)) return text("yeehaww {} {}".format(name, id))
@app.route("/exception") @app.route("/exception")
def exception(request): def exception(request):
raise ServerError("It's dead jim") raise ServerError("It's dead jim")
# ----------------------------------------------- # # ----------------------------------------------- #
# Exceptions # Exceptions
# ----------------------------------------------- # # ----------------------------------------------- #
@app.exception(ServerError) @app.exception(ServerError)
async def test(request, exception): async def test(request, exception):
return json({ "exception": "{}".format(exception), "status": exception.status_code }, status=exception.status_code) return json({"exception": "{}".format(exception), "status": exception.status_code}, status=exception.status_code)
# ----------------------------------------------- # # ----------------------------------------------- #
# Read from request # Read from request
@ -35,15 +41,18 @@ async def test(request, exception):
@app.route("/json") @app.route("/json")
def post_json(request): def post_json(request):
return json({ "received": True, "message": request.json }) return json({"received": True, "message": request.json})
@app.route("/form") @app.route("/form")
def post_json(request): def post_json(request):
return json({ "received": True, "form_data": request.form, "test": request.form.get('test') }) return json({"received": True, "form_data": request.form, "test": request.form.get('test')})
@app.route("/query_string") @app.route("/query_string")
def query_string(request): def query_string(request):
return json({ "parsed": True, "args": request.args, "url": request.url, "query_string": request.query_string }) return json({"parsed": True, "args": request.args, "url": request.url, "query_string": request.query_string})
# ----------------------------------------------- # # ----------------------------------------------- #
# Run Server # Run Server
@ -51,9 +60,10 @@ def query_string(request):
def after_start(loop): def after_start(loop):
log.info("OH OH OH OH OHHHHHHHH") log.info("OH OH OH OH OHHHHHHHH")
def before_stop(loop): def before_stop(loop):
log.info("TRIED EVERYTHING") log.info("TRIED EVERYTHING")
app.run(host="0.0.0.0", port=8000, debug=True, after_start=after_start, before_stop=before_stop) app.run(host="0.0.0.0", port=8000, debug=True, after_start=after_start, before_stop=before_stop)

View File

@ -1 +1 @@
from .sanic import Sanic from .sanic import Sanic

View File

@ -20,5 +20,5 @@ class Config:
""" """
REQUEST_MAX_SIZE = 100000000 # 100 megababies REQUEST_MAX_SIZE = 100000000 # 100 megababies
REQUEST_TIMEOUT = 60 # 60 seconds REQUEST_TIMEOUT = 60 # 60 seconds

View File

@ -1,43 +1,51 @@
from .response import text from .response import text
from traceback import format_exc from traceback import format_exc
class SanicException(Exception): class SanicException(Exception):
def __init__(self, message, status_code=None): def __init__(self, message, status_code=None):
super().__init__(message) super().__init__(message)
if status_code is not None: if status_code is not None:
self.status_code = status_code self.status_code = status_code
class NotFound(SanicException): class NotFound(SanicException):
status_code = 404 status_code = 404
class InvalidUsage(SanicException): class InvalidUsage(SanicException):
status_code = 400 status_code = 400
class ServerError(SanicException): class ServerError(SanicException):
status_code = 500 status_code = 500
class Handler: class Handler:
handlers = None handlers = None
def __init__(self, sanic):
self.handlers = {}
self.sanic = sanic
def add(self, exception, handler): def __init__(self, sanic):
self.handlers[exception] = handler self.handlers = {}
self.sanic = sanic
def response(self, request, exception): def add(self, exception, handler):
""" self.handlers[exception] = handler
Fetches and executes an exception handler and returns a reponse object
:param request: Request
:param exception: Exception to handle
:return: Response object
"""
handler = self.handlers.get(type(exception), self.default)
response = handler(request=request, exception=exception)
return response
def default(self, request, exception): def response(self, request, exception):
if issubclass(type(exception), SanicException): """
return text("Error: {}".format(exception), status=getattr(exception, 'status_code', 500)) Fetches and executes an exception handler and returns a reponse object
elif self.sanic.debug: :param request: Request
return text("Error: {}\nException: {}".format(exception, format_exc()), status=500) :param exception: Exception to handle
else: :return: Response object
return text("An error occurred while generating the request", status=500) """
handler = self.handlers.get(type(exception), self.default)
response = handler(request=request, exception=exception)
return response
def default(self, request, exception):
if issubclass(type(exception), SanicException):
return text("Error: {}".format(exception), status=getattr(exception, 'status_code', 500))
elif self.sanic.debug:
return text("Error: {}\nException: {}".format(exception, format_exc()), status=500)
else:
return text("An error occurred while generating the request", status=500)

View File

@ -1,4 +1,4 @@
import logging import logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s: %(levelname)s: %(message)s") logging.basicConfig(level=logging.INFO, format="%(asctime)s: %(levelname)s: %(message)s")
log = logging.getLogger(__name__) log = logging.getLogger(__name__)

View File

@ -1,4 +1,4 @@
class Middleware: class Middleware:
def __init__(self, process_request=None, process_response=None): def __init__(self, process_request=None, process_response=None):
self.process_request = process_request self.process_request = process_request
self.process_response = process_response self.process_response = process_response

View File

@ -6,20 +6,25 @@ from ujson import loads as json_loads
from .log import log from .log import log
class RequestParameters(dict): class RequestParameters(dict):
""" """
Hosts a dict with lists as values where get returns the first Hosts a dict with lists as values where get returns the first
value of the list and getlist returns the whole shebang value of the list and getlist returns the whole shebang
""" """
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
self.super = super() self.super = super()
self.super.__init__(*args, **kwargs) self.super.__init__(*args, **kwargs)
def get(self, name, default=None): def get(self, name, default=None):
values = self.super.get(name) values = self.super.get(name)
return values[0] if values else default return values[0] if values else default
def getlist(self, name, default=None): def getlist(self, name, default=None):
return self.super.get(name, default) return self.super.get(name, default)
class Request: class Request:
__slots__ = ( __slots__ = (
'url', 'headers', 'version', 'method', 'url', 'headers', 'version', 'method',
@ -75,7 +80,7 @@ class Request:
@property @property
def files(self): def files(self):
if self.parsed_files is None: if self.parsed_files is None:
_ = self.form # compute form to get files _ = self.form # compute form to get files
return self.parsed_files return self.parsed_files
@ -89,7 +94,10 @@ class Request:
return self.parsed_args return self.parsed_args
File = namedtuple('File', ['type', 'body', 'name']) File = namedtuple('File', ['type', 'body', 'name'])
def parse_multipart_form(body, boundary): def parse_multipart_form(body, boundary):
""" """
Parses a request body and returns fields and files Parses a request body and returns fields and files
@ -117,7 +125,7 @@ def parse_multipart_form(body, boundary):
colon_index = form_line.index(':') colon_index = form_line.index(':')
form_header_field = form_line[0:colon_index] form_header_field = form_line[0:colon_index]
form_header_value, form_parameters = parse_header(form_line[colon_index+2:]) form_header_value, form_parameters = parse_header(form_line[colon_index + 2:])
if form_header_field == 'Content-Disposition': if form_header_field == 'Content-Disposition':
if 'filename' in form_parameters: if 'filename' in form_parameters:
@ -126,11 +134,10 @@ def parse_multipart_form(body, boundary):
elif form_header_field == 'Content-Type': elif form_header_field == 'Content-Type':
file_type = form_header_value file_type = form_header_value
post_data = form_part[line_index:-4] post_data = form_part[line_index:-4]
if file_name or file_type: if file_name or file_type:
files[field_name] = File(type=file_type, name=file_name, body=post_data) files[field_name] = File(type=file_type, name=file_name, body=post_data)
else: else:
fields[field_name] = post_data.decode('utf-8') fields[field_name] = post_data.decode('utf-8')
return fields, files return fields, files

View File

@ -18,6 +18,7 @@ STATUS_CODES = {
504: 'Gateway Timeout', 504: 'Gateway Timeout',
} }
class HTTPResponse: class HTTPResponse:
__slots__ = ('body', 'status', 'content_type', 'headers') __slots__ = ('body', 'status', 'content_type', 'headers')
@ -43,18 +44,25 @@ class HTTPResponse:
additional_headers.append('{}: {}\r\n'.format(name, value).encode('utf-8')) additional_headers.append('{}: {}\r\n'.format(name, value).encode('utf-8'))
return b''.join([ return b''.join([
'HTTP/{} {} {}\r\n'.format(version, self.status, STATUS_CODES.get(self.status, 'FAIL')).encode(), 'HTTP/{} {} {}\r\n'.format(version, self.status,
b'Content-Type: ', self.content_type.encode(), b'\r\n', STATUS_CODES.get(self.status, 'FAIL')).encode(),
b'Content-Length: ', str(len(self.body)).encode(), b'\r\n', b'Content-Type: ', self.content_type.encode(), b'\r\n',
b'Connection: ', ('keep-alive' if keep_alive else 'close').encode(), b'\r\n', b'Content-Length: ', str(len(self.body)).encode(), b'\r\n',
] + additional_headers + [ b'Connection: ', ('keep-alive' if keep_alive else 'close').encode(), b'\r\n',
b'\r\n', ] + additional_headers + [
self.body, b'\r\n',
]) self.body,
])
def json(body, status=200, headers=None): def json(body, status=200, headers=None):
return HTTPResponse(ujson.dumps(body), headers=headers, status=status, content_type="application/json; charset=utf-8") return HTTPResponse(ujson.dumps(body), headers=headers, status=status,
content_type="application/json; charset=utf-8")
def text(body, status=200, headers=None): def text(body, status=200, headers=None):
return HTTPResponse(body, status=status, headers=headers, content_type="text/plain; charset=utf-8") return HTTPResponse(body, status=status, headers=headers, content_type="text/plain; charset=utf-8")
def html(body, status=200, headers=None): def html(body, status=200, headers=None):
return HTTPResponse(body, status=status, headers=headers, content_type="text/html; charset=utf-8") return HTTPResponse(body, status=status, headers=headers, content_type="text/html; charset=utf-8")

View File

@ -5,6 +5,7 @@ from .exceptions import NotFound, InvalidUsage
Route = namedtuple("Route", ['handler', 'methods', 'pattern', 'parameters']) Route = namedtuple("Route", ['handler', 'methods', 'pattern', 'parameters'])
Parameter = namedtuple("Parameter", ['name', 'cast']) Parameter = namedtuple("Parameter", ['name', 'cast'])
class Router: class Router:
""" """
Router supports basic routing with parameters and method checks Router supports basic routing with parameters and method checks
@ -42,9 +43,10 @@ class Router:
""" """
# Dict for faster lookups of if method allowed # Dict for faster lookups of if method allowed
methods_dict = { method: True for method in methods } if methods else None methods_dict = {method: True for method in methods} if methods else None
parameters = [] parameters = []
def add_parameter(match): def add_parameter(match):
# We could receive NAME or NAME:PATTERN # We could receive NAME or NAME:PATTERN
parts = match.group(1).split(':') parts = match.group(1).split(':')
@ -93,11 +95,13 @@ class Router:
if route: if route:
if route.methods and not request.method in route.methods: if route.methods and not request.method in route.methods:
raise InvalidUsage("Method {} not allowed for URL {}".format(request.method, request.url), status_code=405) raise InvalidUsage("Method {} not allowed for URL {}".format(request.method, request.url),
status_code=405)
return route.handler, args, kwargs return route.handler, args, kwargs
else: else:
raise NotFound("Requested URL {} not found".format(request.url)) raise NotFound("Requested URL {} not found".format(request.url))
class SimpleRouter: class SimpleRouter:
""" """
Simple router records and reads all routes from a dictionary Simple router records and reads all routes from a dictionary
@ -110,14 +114,15 @@ class SimpleRouter:
def add(self, uri, methods, handler): def add(self, uri, methods, handler):
# Dict for faster lookups of method allowed # Dict for faster lookups of method allowed
methods_dict = { method: True for method in methods } if methods else None methods_dict = {method: True for method in methods} if methods else None
self.routes[uri] = Route(handler=handler, methods=methods_dict, pattern=uri, parameters=None) self.routes[uri] = Route(handler=handler, methods=methods_dict, pattern=uri, parameters=None)
def get(self, request): def get(self, request):
route = self.routes.get(request.url) route = self.routes.get(request.url)
if route: if route:
if route.methods and not request.method in route.methods: if route.methods and not request.method in route.methods:
raise InvalidUsage("Method {} not allowed for URL {}".format(request.method, request.url), status_code=405) raise InvalidUsage("Method {} not allowed for URL {}".format(request.method, request.url),
status_code=405)
return route.handler, [], {} return route.handler, [], {}
else: else:
raise NotFound("Requested URL {} not found".format(request.url)) raise NotFound("Requested URL {} not found".format(request.url))

View File

@ -12,6 +12,7 @@ from .router import Router
from .server import serve from .server import serve
from .exceptions import ServerError from .exceptions import ServerError
class Sanic: class Sanic:
def __init__(self, name, router=None, error_handler=None): def __init__(self, name, router=None, error_handler=None):
self.name = name self.name = name
@ -34,6 +35,7 @@ class Sanic:
:param methods: list or tuple of methods allowed :param methods: list or tuple of methods allowed
:return: decorated function :return: decorated function
""" """
def response(handler): def response(handler):
self.router.add(uri=uri, methods=methods, handler=handler) self.router.add(uri=uri, methods=methods, handler=handler)
return handler return handler
@ -48,6 +50,7 @@ class Sanic:
:param methods: list or tuple of methods allowed :param methods: list or tuple of methods allowed
:return: decorated function :return: decorated function
""" """
def response(handler): def response(handler):
for exception in exceptions: for exception in exceptions:
self.error_handler.add(exception, handler) self.error_handler.add(exception, handler)
@ -63,6 +66,7 @@ class Sanic:
""" """
middleware = None middleware = None
attach_to = 'request' attach_to = 'request'
def register_middleware(middleware): def register_middleware(middleware):
if attach_to == 'request': if attach_to == 'request':
self.request_middleware.append(middleware) self.request_middleware.append(middleware)
@ -156,7 +160,7 @@ class Sanic:
:param before_stop: Function to be executed when a stop signal is received before it is respected :param before_stop: Function to be executed when a stop signal is received before it is respected
:return: Nothing :return: Nothing
""" """
self.error_handler.debug=True self.error_handler.debug = True
self.debug = debug self.debug = debug
if debug: if debug:

View File

@ -3,6 +3,7 @@ from inspect import isawaitable
from signal import SIGINT, SIGTERM from signal import SIGINT, SIGTERM
import httptools import httptools
try: try:
import uvloop as async_loop import uvloop as async_loop
except: except:
@ -11,17 +12,19 @@ except:
from .log import log from .log import log
from .request import Request from .request import Request
class Signal: class Signal:
stopped = False stopped = False
class HttpProtocol(asyncio.Protocol): class HttpProtocol(asyncio.Protocol):
__slots__ = ('loop', 'transport', 'connections', 'signal', # event loop, connection
'parser', 'request', 'url', 'headers', # request params
'request_handler', 'request_timeout', 'request_max_size', # request config
'_total_request_size', '_timeout_handler') # connection management
__slots__ = ('loop', 'transport', 'connections', 'signal', # event loop, connection def __init__(self, *, loop, request_handler, signal=Signal(), connections={}, request_timeout=60,
'parser', 'request', 'url', 'headers', # request params request_max_size=None):
'request_handler', 'request_timeout', 'request_max_size', # request config
'_total_request_size', '_timeout_handler') # connection management
def __init__(self, *, loop, request_handler, signal=Signal(), connections={}, request_timeout=60, request_max_size=None):
self.loop = loop self.loop = loop
self.transport = None self.transport = None
self.request = None self.request = None
@ -37,6 +40,7 @@ class HttpProtocol(asyncio.Protocol):
self._timeout_handler = None self._timeout_handler = None
# -------------------------------------------- # # -------------------------------------------- #
# Connection # Connection
# -------------------------------------------- # # -------------------------------------------- #
@ -51,9 +55,10 @@ class HttpProtocol(asyncio.Protocol):
self.cleanup() self.cleanup()
def connection_timeout(self): def connection_timeout(self):
self.bail_out("Request timed out, connection closed") self.bail_out("Request timed out, connection closed")
# -------------------------------------------- #
# -------------------------------------------- #
# Parsing # Parsing
# -------------------------------------------- # # -------------------------------------------- #
@ -86,7 +91,7 @@ class HttpProtocol(asyncio.Protocol):
def on_headers_complete(self): def on_headers_complete(self):
self.request = Request( self.request = Request(
url_bytes=self.url, url_bytes=self.url,
headers=dict(self.headers), headers=dict(self.headers),
version=self.parser.get_http_version(), version=self.parser.get_http_version(),
method=self.parser.get_method().decode() method=self.parser.get_method().decode()
@ -94,6 +99,7 @@ class HttpProtocol(asyncio.Protocol):
def on_body(self, body): def on_body(self, body):
self.request.body = body self.request.body = body
def on_message_complete(self): def on_message_complete(self):
self.loop.create_task(self.request_handler(self.request, self.write_response)) self.loop.create_task(self.request_handler(self.request, self.write_response))
@ -133,20 +139,22 @@ class HttpProtocol(asyncio.Protocol):
return True return True
return False return False
def serve(host, port, request_handler, after_start=None, before_stop=None, debug=False, request_timeout=60, request_max_size=None):
def serve(host, port, request_handler, after_start=None, before_stop=None, debug=False, request_timeout=60,
request_max_size=None):
# Create Event Loop # Create Event Loop
loop = async_loop.new_event_loop() loop = async_loop.new_event_loop()
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)
# I don't think we take advantage of this # I don't think we take advantage of this
# And it slows everything waaayyy down # And it slows everything waaayyy down
#loop.set_debug(debug) # loop.set_debug(debug)
connections = {} connections = {}
signal = Signal() signal = Signal()
server_coroutine = loop.create_server(lambda: HttpProtocol( server_coroutine = loop.create_server(lambda: HttpProtocol(
loop=loop, loop=loop,
connections = connections, connections=connections,
signal = signal, signal=signal,
request_handler=request_handler, request_handler=request_handler,
request_timeout=request_timeout, request_timeout=request_timeout,
request_max_size=request_max_size, request_max_size=request_max_size,

View File

@ -22,4 +22,4 @@ setup(
'Development Status :: 1 - Alpha', 'Development Status :: 1 - Alpha',
'Environment :: Web Environment', 'Environment :: Web Environment',
], ],
) )

View File

@ -4,13 +4,13 @@ import os
import inspect import inspect
currentdir = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe()))) currentdir = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
sys.path.insert(0,currentdir + '/../../../') sys.path.insert(0, currentdir + '/../../../')
import timeit import timeit
from sanic.response import json from sanic.response import json
print(json({ "test":True }).output()) print(json({"test": True}).output())
print("Running New 100,000 times") print("Running New 100,000 times")
times = 0 times = 0
@ -20,7 +20,7 @@ for n in range(6):
print("Took {} seconds".format(time)) print("Took {} seconds".format(time))
total_time += time total_time += time
times += 1 times += 1
print("Average: {}".format(total_time/times)) print("Average: {}".format(total_time / times))
print("Running Old 100,000 times") print("Running Old 100,000 times")
times = 0 times = 0
@ -30,4 +30,4 @@ for n in range(6):
print("Took {} seconds".format(time)) print("Took {} seconds".format(time))
total_time += time total_time += time
times += 1 times += 1
print("Average: {}".format(total_time/times)) print("Average: {}".format(total_time / times))

View File

@ -3,15 +3,17 @@ import os
import inspect import inspect
currentdir = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe()))) currentdir = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
sys.path.insert(0,currentdir + '/../../../') sys.path.insert(0, currentdir + '/../../../')
from sanic import Sanic from sanic import Sanic
from sanic.response import json from sanic.response import json
app = Sanic("test") app = Sanic("test")
@app.route("/") @app.route("/")
async def test(request): async def test(request):
return json({ "test": True }) return json({"test": True})
app.run(host="0.0.0.0", port=sys.argv[1])
app.run(host="0.0.0.0", port=sys.argv[1])

View File

@ -3,7 +3,7 @@ import os
import inspect import inspect
currentdir = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe()))) currentdir = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
sys.path.insert(0,currentdir + '/../../../') sys.path.insert(0, currentdir + '/../../../')
from sanic import Sanic from sanic import Sanic
from sanic.response import json, text from sanic.response import json, text
@ -11,36 +11,44 @@ from sanic.exceptions import ServerError
app = Sanic("test") app = Sanic("test")
@app.route("/") @app.route("/")
async def test(request): async def test(request):
return json({ "test": True }) return json({"test": True})
@app.route("/sync", methods=['GET', 'POST']) @app.route("/sync", methods=['GET', 'POST'])
def test(request): def test(request):
return json({ "test": True }) return json({"test": True})
@app.route("/text/<name>/<butt:int>") @app.route("/text/<name>/<butt:int>")
def rtext(request, name, butt): def rtext(request, name, butt):
return text("yeehaww {} {}".format(name, butt)) return text("yeehaww {} {}".format(name, butt))
@app.route("/exception") @app.route("/exception")
def exception(request): def exception(request):
raise ServerError("yep") raise ServerError("yep")
@app.route("/exception/async") @app.route("/exception/async")
async def test(request): async def test(request):
raise ServerError("asunk") raise ServerError("asunk")
@app.route("/post_json") @app.route("/post_json")
def post_json(request): def post_json(request):
return json({ "received": True, "message": request.json }) return json({"received": True, "message": request.json})
@app.route("/query_string") @app.route("/query_string")
def query_string(request): def query_string(request):
return json({ "parsed": True, "args": request.args, "url": request.url, "query_string": request.query_string }) return json({"parsed": True, "args": request.args, "url": request.url, "query_string": request.query_string})
import sys import sys
app.run(host="0.0.0.0", port=sys.argv[1]) app.run(host="0.0.0.0", port=sys.argv[1])
@ -55,7 +63,7 @@ app.run(host="0.0.0.0", port=sys.argv[1])
# for n in range(30): # for n in range(30):
# connection = await asyncio_redis.Connection.create(host='192.168.99.100', port=6379) # connection = await asyncio_redis.Connection.create(host='192.168.99.100', port=6379)
# sanic.redis.append(connection) # sanic.redis.append(connection)
# c=0 # c=0
# @app.route("/postgres") # @app.route("/postgres")

View File

@ -49,5 +49,3 @@ def test_invalid_usage_exception():
def test_not_found_exception(): def test_not_found_exception():
request, response = sanic_endpoint_test(exception_app, uri='/404') request, response = sanic_endpoint_test(exception_app, uri='/404')
assert response.status == 404 assert response.status == 404

View File

@ -4,78 +4,85 @@ from sanic.request import Request
from sanic.response import json, text, HTTPResponse from sanic.response import json, text, HTTPResponse
from sanic.utils import sanic_endpoint_test from sanic.utils import sanic_endpoint_test
# ------------------------------------------------------------ # # ------------------------------------------------------------ #
# GET # GET
# ------------------------------------------------------------ # # ------------------------------------------------------------ #
def test_middleware_request(): def test_middleware_request():
app = Sanic('test_middleware_request') app = Sanic('test_middleware_request')
results = [] results = []
@app.middleware
async def handler(request):
results.append(request)
@app.route('/') @app.middleware
async def handler(request): async def handler(request):
return text('OK') results.append(request)
request, response = sanic_endpoint_test(app) @app.route('/')
async def handler(request):
return text('OK')
request, response = sanic_endpoint_test(app)
assert response.text == 'OK'
assert type(results[0]) is Request
assert response.text == 'OK'
assert type(results[0]) is Request
def test_middleware_response(): def test_middleware_response():
app = Sanic('test_middleware_response') app = Sanic('test_middleware_response')
results = [] results = []
@app.middleware('request')
async def process_response(request):
results.append(request)
@app.middleware('response')
async def process_response(request, response):
results.append(request)
results.append(response)
@app.route('/') @app.middleware('request')
async def handler(request): async def process_response(request):
return text('OK') results.append(request)
request, response = sanic_endpoint_test(app) @app.middleware('response')
async def process_response(request, response):
results.append(request)
results.append(response)
@app.route('/')
async def handler(request):
return text('OK')
request, response = sanic_endpoint_test(app)
assert response.text == 'OK'
assert type(results[0]) is Request
assert type(results[1]) is Request
assert issubclass(type(results[2]), HTTPResponse)
assert response.text == 'OK'
assert type(results[0]) is Request
assert type(results[1]) is Request
assert issubclass(type(results[2]), HTTPResponse)
def test_middleware_override_request(): def test_middleware_override_request():
app = Sanic('test_middleware_override_request') app = Sanic('test_middleware_override_request')
@app.middleware @app.middleware
async def halt_request(request): async def halt_request(request):
return text('OK') return text('OK')
@app.route('/') @app.route('/')
async def handler(request): async def handler(request):
return text('FAIL') return text('FAIL')
response = sanic_endpoint_test(app, gather_request=False) response = sanic_endpoint_test(app, gather_request=False)
assert response.status == 200
assert response.text == 'OK'
assert response.status == 200
assert response.text == 'OK'
def test_middleware_override_response(): def test_middleware_override_response():
app = Sanic('test_middleware_override_response') app = Sanic('test_middleware_override_response')
@app.middleware('response') @app.middleware('response')
async def process_response(request, response): async def process_response(request, response):
return text('OK') return text('OK')
@app.route('/') @app.route('/')
async def handler(request): async def handler(request):
return text('FAIL') return text('FAIL')
request, response = sanic_endpoint_test(app) request, response = sanic_endpoint_test(app)
assert response.status == 200 assert response.status == 200
assert response.text == 'OK' assert response.text == 'OK'

View File

@ -3,77 +3,80 @@ from sanic import Sanic
from sanic.response import json, text from sanic.response import json, text
from sanic.utils import sanic_endpoint_test from sanic.utils import sanic_endpoint_test
# ------------------------------------------------------------ # # ------------------------------------------------------------ #
# GET # GET
# ------------------------------------------------------------ # # ------------------------------------------------------------ #
def test_sync(): def test_sync():
app = Sanic('test_text') app = Sanic('test_text')
@app.route('/') @app.route('/')
def handler(request): def handler(request):
return text('Hello') return text('Hello')
request, response = sanic_endpoint_test(app) request, response = sanic_endpoint_test(app)
assert response.text == 'Hello'
assert response.text == 'Hello'
def test_text(): def test_text():
app = Sanic('test_text') app = Sanic('test_text')
@app.route('/') @app.route('/')
async def handler(request): async def handler(request):
return text('Hello') return text('Hello')
request, response = sanic_endpoint_test(app) request, response = sanic_endpoint_test(app)
assert response.text == 'Hello' assert response.text == 'Hello'
def test_json(): def test_json():
app = Sanic('test_json') app = Sanic('test_json')
@app.route('/') @app.route('/')
async def handler(request): async def handler(request):
return json({"test":True}) return json({"test": True})
request, response = sanic_endpoint_test(app) request, response = sanic_endpoint_test(app)
try: try:
results = json_loads(response.text) results = json_loads(response.text)
except: except:
raise ValueError("Expected JSON response but got '{}'".format(response)) raise ValueError("Expected JSON response but got '{}'".format(response))
assert results.get('test') == True assert results.get('test') == True
def test_query_string(): def test_query_string():
app = Sanic('test_query_string') app = Sanic('test_query_string')
@app.route('/') @app.route('/')
async def handler(request): async def handler(request):
return text('OK') return text('OK')
request, response = sanic_endpoint_test(app, params=[("test1", 1), ("test2", "false"), ("test2", "true")]) 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'
assert request.args.get('test1') == '1'
assert request.args.get('test2') == 'false'
# ------------------------------------------------------------ # # ------------------------------------------------------------ #
# POST # POST
# ------------------------------------------------------------ # # ------------------------------------------------------------ #
def test_post_json(): def test_post_json():
app = Sanic('test_post_json') app = Sanic('test_post_json')
@app.route('/') @app.route('/')
async def handler(request): async def handler(request):
return text('OK') return text('OK')
payload = {'test': 'OK'} payload = {'test': 'OK'}
headers = {'content-type': 'application/json'} headers = {'content-type': 'application/json'}
request, response = sanic_endpoint_test(app, data=json_dumps(payload), headers=headers) request, response = sanic_endpoint_test(app, data=json_dumps(payload), headers=headers)
assert request.json.get('test') == 'OK' assert request.json.get('test') == 'OK'
assert response.text == 'OK' assert response.text == 'OK'

View File

@ -3,93 +3,97 @@ from sanic import Sanic
from sanic.response import json, text from sanic.response import json, text
from sanic.utils import sanic_endpoint_test from sanic.utils import sanic_endpoint_test
# ------------------------------------------------------------ # # ------------------------------------------------------------ #
# UTF-8 # UTF-8
# ------------------------------------------------------------ # # ------------------------------------------------------------ #
def test_dynamic_route(): def test_dynamic_route():
app = Sanic('test_dynamic_route') app = Sanic('test_dynamic_route')
results = [] results = []
@app.route('/folder/<name>') @app.route('/folder/<name>')
async def handler(request, name): async def handler(request, name):
results.append(name) results.append(name)
return text('OK') return text('OK')
request, response = sanic_endpoint_test(app, uri='/folder/test123') request, response = sanic_endpoint_test(app, uri='/folder/test123')
assert response.text == 'OK'
assert results[0] == 'test123'
assert response.text == 'OK'
assert results[0] == 'test123'
def test_dynamic_route_string(): def test_dynamic_route_string():
app = Sanic('test_dynamic_route_string') app = Sanic('test_dynamic_route_string')
results = [] results = []
@app.route('/folder/<name:string>') @app.route('/folder/<name:string>')
async def handler(request, name): async def handler(request, name):
results.append(name) results.append(name)
return text('OK') return text('OK')
request, response = sanic_endpoint_test(app, uri='/folder/test123') request, response = sanic_endpoint_test(app, uri='/folder/test123')
assert response.text == 'OK'
assert results[0] == 'test123'
assert response.text == 'OK'
assert results[0] == 'test123'
def test_dynamic_route_int(): def test_dynamic_route_int():
app = Sanic('test_dynamic_route_int') app = Sanic('test_dynamic_route_int')
results = [] results = []
@app.route('/folder/<folder_id:int>') @app.route('/folder/<folder_id:int>')
async def handler(request, folder_id): async def handler(request, folder_id):
results.append(folder_id) results.append(folder_id)
return text('OK') return text('OK')
request, response = sanic_endpoint_test(app, uri='/folder/12345') request, response = sanic_endpoint_test(app, uri='/folder/12345')
assert response.text == 'OK' assert response.text == 'OK'
assert type(results[0]) is int assert type(results[0]) is int
request, response = sanic_endpoint_test(app, uri='/folder/asdf') request, response = sanic_endpoint_test(app, uri='/folder/asdf')
assert response.status == 404 assert response.status == 404
def test_dynamic_route_number(): def test_dynamic_route_number():
app = Sanic('test_dynamic_route_int') app = Sanic('test_dynamic_route_int')
results = [] results = []
@app.route('/weight/<weight:number>') @app.route('/weight/<weight:number>')
async def handler(request, weight): async def handler(request, weight):
results.append(weight) results.append(weight)
return text('OK') return text('OK')
request, response = sanic_endpoint_test(app, uri='/weight/12345') request, response = sanic_endpoint_test(app, uri='/weight/12345')
assert response.text == 'OK' assert response.text == 'OK'
assert type(results[0]) is float assert type(results[0]) is float
request, response = sanic_endpoint_test(app, uri='/weight/1234.56') request, response = sanic_endpoint_test(app, uri='/weight/1234.56')
assert response.status == 200 assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/weight/1234-56')
assert response.status == 404
request, response = sanic_endpoint_test(app, uri='/weight/1234-56')
assert response.status == 404
def test_dynamic_route_regex(): def test_dynamic_route_regex():
app = Sanic('test_dynamic_route_int') app = Sanic('test_dynamic_route_int')
@app.route('/folder/<folder_id:[A-Za-z0-9]{0,4}>') @app.route('/folder/<folder_id:[A-Za-z0-9]{0,4}>')
async def handler(request, folder_id): async def handler(request, folder_id):
return text('OK') return text('OK')
request, response = sanic_endpoint_test(app, uri='/folder/test') request, response = sanic_endpoint_test(app, uri='/folder/test')
assert response.status == 200 assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test1') request, response = sanic_endpoint_test(app, uri='/folder/test1')
assert response.status == 404 assert response.status == 404
request, response = sanic_endpoint_test(app, uri='/folder/test-123') request, response = sanic_endpoint_test(app, uri='/folder/test-123')
assert response.status == 404 assert response.status == 404
request, response = sanic_endpoint_test(app, uri='/folder/') request, response = sanic_endpoint_test(app, uri='/folder/')
assert response.status == 200 assert response.status == 200

View File

@ -3,52 +3,56 @@ from sanic import Sanic
from sanic.response import json, text from sanic.response import json, text
from sanic.utils import sanic_endpoint_test from sanic.utils import sanic_endpoint_test
# ------------------------------------------------------------ # # ------------------------------------------------------------ #
# UTF-8 # UTF-8
# ------------------------------------------------------------ # # ------------------------------------------------------------ #
def test_utf8_query_string(): def test_utf8_query_string():
app = Sanic('test_utf8_query_string') app = Sanic('test_utf8_query_string')
@app.route('/') @app.route('/')
async def handler(request): async def handler(request):
return text('OK') return text('OK')
request, response = sanic_endpoint_test(app, params=[("utf8", '')])
assert request.args.get('utf8') == ''
request, response = sanic_endpoint_test(app, params=[("utf8", '')])
assert request.args.get('utf8') == ''
def test_utf8_response(): def test_utf8_response():
app = Sanic('test_utf8_response') app = Sanic('test_utf8_response')
@app.route('/') @app.route('/')
async def handler(request): async def handler(request):
return text('') return text('')
request, response = sanic_endpoint_test(app)
assert response.text == ''
request, response = sanic_endpoint_test(app)
assert response.text == ''
def skip_test_utf8_route(): def skip_test_utf8_route():
app = Sanic('skip_test_utf8_route') app = Sanic('skip_test_utf8_route')
@app.route('/') @app.route('/')
async def handler(request): async def handler(request):
return text('OK') return text('OK')
# UTF-8 Paths are not supported
request, response = sanic_endpoint_test(app, route='/✓', uri='/✓')
assert response.text == 'OK'
# UTF-8 Paths are not supported
request, response = sanic_endpoint_test(app, route='/✓', uri='/✓')
assert response.text == 'OK'
def test_utf8_post_json(): def test_utf8_post_json():
app = Sanic('test_utf8_post_json') app = Sanic('test_utf8_post_json')
@app.route('/') @app.route('/')
async def handler(request): async def handler(request):
return text('OK') return text('OK')
payload = {'test': ''} payload = {'test': ''}
headers = {'content-type': 'application/json'} headers = {'content-type': 'application/json'}
request, response = sanic_endpoint_test(app, data=json_dumps(payload), headers=headers) request, response = sanic_endpoint_test(app, data=json_dumps(payload), headers=headers)
assert request.json.get('test') == '' assert request.json.get('test') == ''
assert response.text == 'OK' assert response.text == 'OK'