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

Merge branch 'master' into convert_dict_to_set

Browse Source
This commit is contained in:
Eli Uriegas 2016-12-14 11:26:31 -06:00 committed by GitHub
commit ddfb7f2861
23 changed files with 830 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
README.md
View File

@ -33,13 +33,17 @@ All tests were run on an AWS medium instance running ubuntu, using 1 process. E
from sanic import Sanic
from sanic.response import json
app = Sanic()
@app.route("/")
async def test(request):
return json({"hello": "world"})
app.run(host="0.0.0.0", port=8000)
if __name__ == '__main__':
app.run(host="0.0.0.0", port=8000)
```
## Installation
@ -52,6 +56,7 @@ app.run(host="0.0.0.0", port=8000)
* [Middleware](docs/middleware.md)
* [Exceptions](docs/exceptions.md)
* [Blueprints](docs/blueprints.md)
* [Class Based Views](docs/class_based_views.md)
* [Cookies](docs/cookies.md)
* [Static Files](docs/static_files.md)
* [Deploying](docs/deploying.md)
@ -72,7 +77,7 @@ app.run(host="0.0.0.0", port=8000)
▄▄▄▄▄
▀▀▀██████▄▄▄ _______________
▄▄▄▄▄ █████████▄ / \
▀▀▀▀█████▌ ▀▐▄ ▀▐█ | Gotta go fast! |
▀▀▀▀█████▌ ▀▐▄ ▀▐█ | Gotta go fast! |
▀▀█████▄▄ ▀██████▄██ | _________________/
▀▄▄▄▄▄ ▀▀█▄▀█════█▀ |/
▀▀▀▄ ▀▀███ ▀ ▄▄

44
docs/class_based_views.md Normal file
View File

@ -0,0 +1,44 @@
# Class based views
Sanic has simple class based implementation. You should implement methods(get, post, put, patch, delete) for the class to every HTTP method you want to support. If someone tries to use a method that has not been implemented, there will be 405 response.
## Examples
```python
from sanic import Sanic
from sanic.views import HTTPMethodView
app = Sanic('some_name')
class SimpleView(HTTPMethodView):
def get(self, request):
return text('I am get method')
def post(self, request):
return text('I am post method')
def put(self, request):
return text('I am put method')
def patch(self, request):
return text('I am patch method')
def delete(self, request):
return text('I am delete method')
app.add_route(SimpleView(), '/')
```
If you need any url params just mention them in method definition:
```python
class NameView(HTTPMethodView):
def get(self, request, name):
return text('Hello {}'.format(name))
app.add_route(NameView(), '/<name>')
```

12
docs/routing.md
View File

@ -29,4 +29,16 @@ async def person_handler(request, name):
async def folder_handler(request, folder_id):
return text('Folder - {}'.format(folder_id))
async def handler1(request):
return text('OK')
app.add_route(handler1, '/test')
async def handler(request, name):
return text('Folder - {}'.format(name))
app.add_route(handler, '/folder/<name>')
async def person_handler(request, name):
return text('Person - {}'.format(name))
app.add_route(handler, '/person/<name:[A-z]>')
```

18
examples/jinja_example.py Normal file
View File

@ -0,0 +1,18 @@
## To use this example:
# curl -d '{"name": "John Doe"}' localhost:8000
from sanic import Sanic
from sanic.response import html
from jinja2 import Template
template = Template('Hello {{ name }}!')
app = Sanic(__name__)
@app.route('/')
async def test(request):
data = request.json
return html(template.render(**data))
app.run(host="0.0.0.0", port=8000)

21
examples/request_timeout.py Normal file
View File

@ -0,0 +1,21 @@
from sanic import Sanic
import asyncio
from sanic.response import text
from sanic.config import Config
from sanic.exceptions import RequestTimeout
Config.REQUEST_TIMEOUT = 1
app = Sanic(__name__)
@app.route('/')
async def test(request):
await asyncio.sleep(3)
return text('Hello, world!')
@app.exception(RequestTimeout)
def timeout(request, exception):
return text('RequestTimeout from error_handler.', 408)
app.run(host='0.0.0.0', port=8000)

65
examples/sanic_aiopg_example.py Normal file
View File

@ -0,0 +1,65 @@
""" To run this example you need additional aiopg package
"""
import os
import asyncio
import uvloop
import aiopg
from sanic import Sanic
from sanic.response import json
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
database_name = os.environ['DATABASE_NAME']
database_host = os.environ['DATABASE_HOST']
database_user = os.environ['DATABASE_USER']
database_password = os.environ['DATABASE_PASSWORD']
connection = 'postgres://{0}:{1}@{2}/{3}'.format(database_user,
database_password,
database_host,
database_name)
loop = asyncio.get_event_loop()
async def get_pool():
return await aiopg.create_pool(connection)
app = Sanic(name=__name__)
pool = loop.run_until_complete(get_pool())
async def prepare_db():
""" Let's create some table and add some data
"""
async with pool.acquire() as conn:
async with conn.cursor() as cur:
await cur.execute('DROP TABLE IF EXISTS sanic_polls')
await cur.execute("""CREATE TABLE sanic_polls (
id serial primary key,
question varchar(50),
pub_date timestamp
);""")
for i in range(0, 100):
await cur.execute("""INSERT INTO sanic_polls
(id, question, pub_date) VALUES ({}, {}, now())
""".format(i, i))
@app.route("/")
async def handle(request):
async with pool.acquire() as conn:
async with conn.cursor() as cur:
result = []
await cur.execute("SELECT question, pub_date FROM sanic_polls")
async for row in cur:
result.append({"question": row[0], "pub_date": row[1]})
return json({"polls": result})
if __name__ == '__main__':
loop.run_until_complete(prepare_db())
app.run(host='0.0.0.0', port=8000, loop=loop)

73
examples/sanic_aiopg_sqlalchemy_example.py Normal file
View File

@ -0,0 +1,73 @@
""" To run this example you need additional aiopg package
"""
import os
import asyncio
import datetime
import uvloop
from aiopg.sa import create_engine
import sqlalchemy as sa
from sanic import Sanic
from sanic.response import json
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
database_name = os.environ['DATABASE_NAME']
database_host = os.environ['DATABASE_HOST']
database_user = os.environ['DATABASE_USER']
database_password = os.environ['DATABASE_PASSWORD']
connection = 'postgres://{0}:{1}@{2}/{3}'.format(database_user,
database_password,
database_host,
database_name)
loop = asyncio.get_event_loop()
metadata = sa.MetaData()
polls = sa.Table('sanic_polls', metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('question', sa.String(50)),
sa.Column("pub_date", sa.DateTime))
async def get_engine():
return await create_engine(connection)
app = Sanic(name=__name__)
engine = loop.run_until_complete(get_engine())
async def prepare_db():
""" Let's add some data
"""
async with engine.acquire() as conn:
await conn.execute('DROP TABLE IF EXISTS sanic_polls')
await conn.execute("""CREATE TABLE sanic_polls (
id serial primary key,
question varchar(50),
pub_date timestamp
);""")
for i in range(0, 100):
await conn.execute(
polls.insert().values(question=i,
pub_date=datetime.datetime.now())
)
@app.route("/")
async def handle(request):
async with engine.acquire() as conn:
result = []
async for row in conn.execute(polls.select()):
result.append({"question": row.question, "pub_date": row.pub_date})
return json({"polls": result})
if __name__ == '__main__':
loop.run_until_complete(prepare_db())
app.run(host='0.0.0.0', port=8000, loop=loop)

1
requirements-dev.txt
View File

@ -11,3 +11,4 @@ bottle
kyoukai
falcon
tornado
aiofiles

2
sanic/__init__.py
View File

@ -1,6 +1,6 @@
from .sanic import Sanic
from .blueprints import Blueprint
__version__ = '0.1.7'
__version__ = '0.1.8'
__all__ = ['Sanic', 'Blueprint']

6
sanic/blueprints.py
View File

@ -91,6 +91,12 @@ class Blueprint:
return handler
return decorator
def add_route(self, handler, uri, methods=None):
"""
"""
self.record(lambda s: s.add_route(handler, uri, methods))
return handler
def listener(self, event):
"""
"""

8
sanic/exceptions.py
View File

@ -30,6 +30,14 @@ class FileNotFound(NotFound):
self.relative_url = relative_url
class RequestTimeout(SanicException):
status_code = 408
class PayloadTooLarge(SanicException):
status_code = 413
class Handler:
handlers = None

10
sanic/request.py
View File

@ -4,6 +4,7 @@ from http.cookies import SimpleCookie
from httptools import parse_url
from urllib.parse import parse_qs
from ujson import loads as json_loads
from sanic.exceptions import InvalidUsage
from .log import log
@ -67,7 +68,7 @@ class Request(dict):
try:
self.parsed_json = json_loads(self.body)
except Exception:
log.exception("failed when parsing body as json")
raise InvalidUsage("Failed when parsing body as json")
return self.parsed_json
@ -89,7 +90,7 @@ class Request(dict):
self.parsed_form, self.parsed_files = (
parse_multipart_form(self.body, boundary))
except Exception:
log.exception("failed when parsing form")
log.exception("Failed when parsing form")
return self.parsed_form
@ -114,9 +115,10 @@ class Request(dict):
@property
def cookies(self):
if self._cookies is None:
if 'Cookie' in self.headers:
cookie = self.headers.get('Cookie') or self.headers.get('cookie')
if cookie is not None:
cookies = SimpleCookie()
cookies.load(self.headers['Cookie'])
cookies.load(cookie)
self._cookies = {name: cookie.value
for name, cookie in cookies.items()}
else:

12
sanic/router.py
View File

@ -30,11 +30,17 @@ class Router:
@sanic.route('/my/url/<my_parameter>', methods=['GET', 'POST', ...])
def my_route(request, my_parameter):
do stuff...
or
@sanic.route('/my/url/<my_paramter>:type', methods['GET', 'POST', ...])
def my_route_with_type(request, my_parameter):
do stuff...
Parameters will be passed as keyword arguments to the request handling
function provided Parameters can also have a type by appending :type to
the <parameter>. If no type is provided, a string is expected. A regular
expression can also be passed in as the type
function. Provided parameters can also have a type by appending :type to
the <parameter>. Given parameter must be able to be type-casted to this.
If no type is provided, a string is expected. A regular expression can
also be passed in as the type. The argument given to the function will
always be a string, independent of the type.
"""
routes_static = None
routes_dynamic = None

14
sanic/sanic.py
View File

@ -60,6 +60,19 @@ class Sanic:
return response
def add_route(self, handler, uri, methods=None):
"""
A helper method to register class instance or
functions as a handler to the application url
routes.
:param handler: function or class instance
:param uri: path of the URL
:param methods: list or tuple of methods allowed
:return: function or class instance
"""
self.route(uri=uri, methods=methods)(handler)
return handler
# Decorator
def exception(self, *exceptions):
"""
@ -250,6 +263,7 @@ class Sanic:
'sock': sock,
'debug': debug,
'request_handler': self.handle_request,
'error_handler': self.error_handler,
'request_timeout': self.config.REQUEST_TIMEOUT,
'request_max_size': self.config.REQUEST_MAX_SIZE,
'loop': loop

60
sanic/server.py
View File

@ -4,7 +4,8 @@ from inspect import isawaitable
from multidict import CIMultiDict
from signal import SIGINT, SIGTERM
from time import time
import httptools
from httptools import HttpRequestParser
from httptools.parser.errors import HttpParserError
try:
import uvloop as async_loop
@ -13,6 +14,7 @@ except ImportError:
from .log import log
from .request import Request
from .exceptions import RequestTimeout, PayloadTooLarge
class Signal:
@ -33,8 +35,9 @@ class HttpProtocol(asyncio.Protocol):
# connection management
'_total_request_size', '_timeout_handler', '_last_communication_time')
def __init__(self, *, loop, request_handler, signal=Signal(),
connections=set(), request_timeout=60,
def __init__(self, *, loop, request_handler, error_handler,
signal=Signal(), connections={}, request_timeout=60,
request_max_size=None):
self.loop = loop
self.transport = None
@ -45,11 +48,13 @@ class HttpProtocol(asyncio.Protocol):
self.signal = signal
self.connections = connections
self.request_handler = request_handler
self.error_handler = error_handler
self.request_timeout = request_timeout
self.request_max_size = request_max_size
self._total_request_size = 0
self._timeout_handler = None
self._last_request_time = None
self._request_handler_task = None
# -------------------------------------------- #
# Connection
@ -75,7 +80,10 @@ class HttpProtocol(asyncio.Protocol):
self._timeout_handler = \
self.loop.call_later(time_left, self.connection_timeout)
else:
self.bail_out("Request timed out, connection closed")
if self._request_handler_task:
self._request_handler_task.cancel()
exception = RequestTimeout('Request Timeout')
self.write_error(exception)
# -------------------------------------------- #
# Parsing
@ -86,20 +94,19 @@ class HttpProtocol(asyncio.Protocol):
# memory limits
self._total_request_size += len(data)
if self._total_request_size > self.request_max_size:
return self.bail_out(
"Request too large ({}), connection closed".format(
self._total_request_size))
exception = PayloadTooLarge('Payload Too Large')
self.write_error(exception)
# Create parser if this is the first time we're receiving data
if self.parser is None:
assert self.request is None
self.headers = []
self.parser = httptools.HttpRequestParser(self)
self.parser = HttpRequestParser(self)
# Parse request chunk or close connection
try:
self.parser.feed_data(data)
except httptools.parser.errors.HttpParserError as e:
except HttpParserError as e:
self.bail_out(
"Invalid request data, connection closed ({})".format(e))
@ -108,8 +115,8 @@ class HttpProtocol(asyncio.Protocol):
def on_header(self, name, value):
if name == b'Content-Length' and int(value) > self.request_max_size:
return self.bail_out(
"Request body too large ({}), connection closed".format(value))
exception = PayloadTooLarge('Payload Too Large')
self.write_error(exception)
self.headers.append((name.decode(), value.decode('utf-8')))
@ -132,7 +139,7 @@ class HttpProtocol(asyncio.Protocol):
self.request.body = body
def on_message_complete(self):
self.loop.create_task(
self._request_handler_task = self.loop.create_task(
self.request_handler(self.request, self.write_response))
# -------------------------------------------- #
@ -156,6 +163,16 @@ class HttpProtocol(asyncio.Protocol):
self.bail_out(
"Writing response failed, connection closed {}".format(e))
def write_error(self, exception):
try:
response = self.error_handler.response(self.request, exception)
version = self.request.version if self.request else '1.1'
self.transport.write(response.output(version))
self.transport.close()
except Exception as e:
self.bail_out(
"Writing error failed, connection closed {}".format(e))
def bail_out(self, message):
log.debug(message)
self.transport.close()
@ -165,6 +182,7 @@ class HttpProtocol(asyncio.Protocol):
self.request = None
self.url = None
self.headers = None
self._request_handler_task = None
self._total_request_size = 0
def close_if_idle(self):
@ -204,8 +222,8 @@ def trigger_events(events, loop):
loop.run_until_complete(result)
def serve(host, port, request_handler, before_start=None, after_start=None,
before_stop=None, after_stop=None,
def serve(host, port, request_handler, error_handler, before_start=None,
after_start=None, before_stop=None, after_stop=None,
debug=False, request_timeout=60, sock=None,
request_max_size=None, reuse_port=False, loop=None):
"""
@ -235,14 +253,24 @@ def serve(host, port, request_handler, before_start=None, after_start=None,
connections = set()
signal = Signal()
server_coroutine = loop.create_server(lambda: HttpProtocol(
server = partial(
HttpProtocol,
loop=loop,
connections=connections,
signal=signal,
request_handler=request_handler,
error_handler=error_handler,
request_timeout=request_timeout,
request_max_size=request_max_size,
), host, port, reuse_port=reuse_port, sock=sock)
)
server_coroutine = loop.create_server(
server,
host,
port,
reuse_port=reuse_port,
sock=sock
)
# Instead of pulling time at the end of every request,
# pull it once per minute

10
sanic/utils.py
View File

@ -16,7 +16,8 @@ async def local_request(method, uri, cookies=None, *args, **kwargs):
def sanic_endpoint_test(app, method='get', uri='/', gather_request=True,
loop=None, *request_args, **request_kwargs):
loop=None, debug=False, *request_args,
**request_kwargs):
results = []
exceptions = []
@ -34,7 +35,8 @@ def sanic_endpoint_test(app, method='get', uri='/', gather_request=True,
exceptions.append(e)
app.stop()
app.run(host=HOST, port=42101, after_start=_collect_response, loop=loop)
app.run(host=HOST, debug=debug, port=42101,
after_start=_collect_response, loop=loop)
if exceptions:
raise ValueError("Exception during request: {}".format(exceptions))
@ -45,11 +47,11 @@ def sanic_endpoint_test(app, method='get', uri='/', gather_request=True,
return request, response
except:
raise ValueError(
"request and response object expected, got ({})".format(
"Request and response object expected, got ({})".format(
results))
else:
try:
return results[0]
except:
raise ValueError(
"request object expected, got ({})".format(results))
"Request object expected, got ({})".format(results))

39
sanic/views.py Normal file
View File

@ -0,0 +1,39 @@
from .exceptions import InvalidUsage
class HTTPMethodView:
""" Simple class based implementation of view for the sanic.
You should implement methods (get, post, put, patch, delete) for the class
to every HTTP method you want to support.
For example:
class DummyView(View):
def get(self, request, *args, **kwargs):
return text('I am get method')
def put(self, request, *args, **kwargs):
return text('I am put method')
etc.
If someone tries to use a non-implemented method, there will be a
405 response.
If you need any url params just mention them in method definition:
class DummyView(View):
def get(self, request, my_param_here, *args, **kwargs):
return text('I am get method with %s' % my_param_here)
To add the view into the routing you could use
1) app.add_route(DummyView(), '/')
2) app.route('/')(DummyView())
"""
def __call__(self, request, *args, **kwargs):
handler = getattr(self, request.method.lower(), None)
if handler:
return handler(request, *args, **kwargs)
raise InvalidUsage(
'Method {} not allowed for URL {}'.format(
request.method, request.url), status_code=405)

13
tests/test_cookies.py
View File

@ -25,6 +25,19 @@ def test_cookies():
assert response.text == 'Cookies are: working!'
assert response_cookies['right_back'].value == 'at you'
def test_http2_cookies():
app = Sanic('test_http2_cookies')
@app.route('/')
async def handler(request):
response = text('Cookies are: {}'.format(request.cookies['test']))
return response
headers = {'cookie': 'test=working!'}
request, response = sanic_endpoint_test(app, headers=headers)
assert response.text == 'Cookies are: working!'
def test_cookie_options():
app = Sanic('test_text')

54
tests/test_payload_too_large.py Normal file
View File

@ -0,0 +1,54 @@
from sanic import Sanic
from sanic.response import text
from sanic.exceptions import PayloadTooLarge
from sanic.utils import sanic_endpoint_test
data_received_app = Sanic('data_received')
data_received_app.config.REQUEST_MAX_SIZE = 1
data_received_default_app = Sanic('data_received_default')
data_received_default_app.config.REQUEST_MAX_SIZE = 1
on_header_default_app = Sanic('on_header')
on_header_default_app.config.REQUEST_MAX_SIZE = 500
@data_received_app.route('/1')
async def handler1(request):
return text('OK')
@data_received_app.exception(PayloadTooLarge)
def handler_exception(request, exception):
return text('Payload Too Large from error_handler.', 413)
def test_payload_too_large_from_error_handler():
response = sanic_endpoint_test(
data_received_app, uri='/1', gather_request=False)
assert response.status == 413
assert response.text == 'Payload Too Large from error_handler.'
@data_received_default_app.route('/1')
async def handler2(request):
return text('OK')
def test_payload_too_large_at_data_received_default():
response = sanic_endpoint_test(
data_received_default_app, uri='/1', gather_request=False)
assert response.status == 413
assert response.text == 'Error: Payload Too Large'
@on_header_default_app.route('/1')
async def handler3(request):
return text('OK')
def test_payload_too_large_at_on_header_default():
data = 'a' * 1000
response = sanic_endpoint_test(
on_header_default_app, method='post', uri='/1',
gather_request=False, data=data)
assert response.status == 413
assert response.text == 'Error: Payload Too Large'

40
tests/test_request_timeout.py Normal file
View File

@ -0,0 +1,40 @@
from sanic import Sanic
import asyncio
from sanic.response import text
from sanic.exceptions import RequestTimeout
from sanic.utils import sanic_endpoint_test
from sanic.config import Config
Config.REQUEST_TIMEOUT = 1
request_timeout_app = Sanic('test_request_timeout')
request_timeout_default_app = Sanic('test_request_timeout_default')
@request_timeout_app.route('/1')
async def handler_1(request):
await asyncio.sleep(2)
return text('OK')
@request_timeout_app.exception(RequestTimeout)
def handler_exception(request, exception):
return text('Request Timeout from error_handler.', 408)
def test_server_error_request_timeout():
request, response = sanic_endpoint_test(request_timeout_app, uri='/1')
assert response.status == 408
assert response.text == 'Request Timeout from error_handler.'
@request_timeout_default_app.route('/1')
async def handler_2(request):
await asyncio.sleep(2)
return text('OK')
def test_default_server_error_request_timeout():
request, response = sanic_endpoint_test(
request_timeout_default_app, uri='/1')
assert response.status == 408
assert response.text == 'Error: Request Timeout'

13
tests/test_requests.py
View File

@ -49,6 +49,19 @@ def test_json():
assert results.get('test') == True
def test_invalid_json():
app = Sanic('test_json')
@app.route('/')
async def handler(request):
return json(request.json())
data = "I am not json"
request, response = sanic_endpoint_test(app, data=data)
assert response.status == 400
def test_query_string():
app = Sanic('test_query_string')

184
tests/test_routes.py
View File

@ -84,7 +84,7 @@ def test_dynamic_route_int():
def test_dynamic_route_number():
app = Sanic('test_dynamic_route_int')
app = Sanic('test_dynamic_route_number')
results = []
@ -105,7 +105,7 @@ def test_dynamic_route_number():
def test_dynamic_route_regex():
app = Sanic('test_dynamic_route_int')
app = Sanic('test_dynamic_route_regex')
@app.route('/folder/<folder_id:[A-Za-z0-9]{0,4}>')
async def handler(request, folder_id):
@ -145,7 +145,7 @@ def test_dynamic_route_unhashable():
def test_route_duplicate():
app = Sanic('test_dynamic_route')
app = Sanic('test_route_duplicate')
with pytest.raises(RouteExists):
@app.route('/test')
@ -178,3 +178,181 @@ def test_method_not_allowed():
request, response = sanic_endpoint_test(app, method='post', uri='/test')
assert response.status == 405
def test_static_add_route():
app = Sanic('test_static_add_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.text == 'OK1'
request, response = sanic_endpoint_test(app, uri='/test2')
assert response.text == 'OK2'
def test_dynamic_add_route():
app = Sanic('test_dynamic_add_route')
results = []
async def handler(request, name):
results.append(name)
return text('OK')
app.add_route(handler, '/folder/<name>')
request, response = sanic_endpoint_test(app, uri='/folder/test123')
assert response.text == 'OK'
assert results[0] == 'test123'
def test_dynamic_add_route_string():
app = Sanic('test_dynamic_add_route_string')
results = []
async def handler(request, name):
results.append(name)
return text('OK')
app.add_route(handler, '/folder/<name:string>')
request, response = sanic_endpoint_test(app, uri='/folder/test123')
assert response.text == 'OK'
assert results[0] == 'test123'
request, response = sanic_endpoint_test(app, uri='/folder/favicon.ico')
assert response.text == 'OK'
assert results[1] == 'favicon.ico'
def test_dynamic_add_route_int():
app = Sanic('test_dynamic_add_route_int')
results = []
async def handler(request, folder_id):
results.append(folder_id)
return text('OK')
app.add_route(handler, '/folder/<folder_id:int>')
request, response = sanic_endpoint_test(app, uri='/folder/12345')
assert response.text == 'OK'
assert type(results[0]) is int
request, response = sanic_endpoint_test(app, uri='/folder/asdf')
assert response.status == 404
def test_dynamic_add_route_number():
app = Sanic('test_dynamic_add_route_number')
results = []
async def handler(request, weight):
results.append(weight)
return text('OK')
app.add_route(handler, '/weight/<weight:number>')
request, response = sanic_endpoint_test(app, uri='/weight/12345')
assert response.text == 'OK'
assert type(results[0]) is float
request, response = sanic_endpoint_test(app, uri='/weight/1234.56')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/weight/1234-56')
assert response.status == 404
def test_dynamic_add_route_regex():
app = Sanic('test_dynamic_route_int')
async def handler(request, folder_id):
return text('OK')
app.add_route(handler, '/folder/<folder_id:[A-Za-z0-9]{0,4}>')
request, response = sanic_endpoint_test(app, uri='/folder/test')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test1')
assert response.status == 404
request, response = sanic_endpoint_test(app, uri='/folder/test-123')
assert response.status == 404
request, response = sanic_endpoint_test(app, uri='/folder/')
assert response.status == 200
def test_dynamic_add_route_unhashable():
app = Sanic('test_dynamic_add_route_unhashable')
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
request, response = sanic_endpoint_test(app, uri='/folder/test/nope/')
assert response.status == 404
def test_add_route_duplicate():
app = Sanic('test_add_route_duplicate')
with pytest.raises(RouteExists):
async def handler1(request):
pass
async def handler2(request):
pass
app.add_route(handler1, '/test')
app.add_route(handler2, '/test')
with pytest.raises(RouteExists):
async def handler1(request, dynamic):
pass
async def handler2(request, dynamic):
pass
app.add_route(handler1, '/test/<dynamic>/')
app.add_route(handler2, '/test/<dynamic>/')
def test_add_route_method_not_allowed():
app = Sanic('test_add_route_method_not_allowed')
async def handler(request):
return text('OK')
app.add_route(handler, '/test', methods=['GET'])
request, response = sanic_endpoint_test(app, uri='/test')
assert response.status == 200
request, response = sanic_endpoint_test(app, method='post', uri='/test')
assert response.status == 405

155
tests/test_views.py Normal file
View File

@ -0,0 +1,155 @@
from sanic import Sanic
from sanic.response import text, HTTPResponse
from sanic.views import HTTPMethodView
from sanic.blueprints import Blueprint
from sanic.request import Request
from sanic.utils import sanic_endpoint_test
def test_methods():
app = Sanic('test_methods')
class DummyView(HTTPMethodView):
def get(self, request):
return text('I am get method')
def post(self, request):
return text('I am post method')
def put(self, request):
return text('I am put method')
def patch(self, request):
return text('I am patch method')
def delete(self, request):
return text('I am delete method')
app.add_route(DummyView(), '/')
request, response = sanic_endpoint_test(app, method="get")
assert response.text == 'I am get method'
request, response = sanic_endpoint_test(app, method="post")
assert response.text == 'I am post method'
request, response = sanic_endpoint_test(app, method="put")
assert response.text == 'I am put method'
request, response = sanic_endpoint_test(app, method="patch")
assert response.text == 'I am patch method'
request, response = sanic_endpoint_test(app, method="delete")
assert response.text == 'I am delete method'
def test_unexisting_methods():
app = Sanic('test_unexisting_methods')
class DummyView(HTTPMethodView):
def get(self, request):
return text('I am get method')
app.add_route(DummyView(), '/')
request, response = sanic_endpoint_test(app, method="get")
assert response.text == 'I am get method'
request, response = sanic_endpoint_test(app, method="post")
assert response.text == 'Error: Method POST not allowed for URL /'
def test_argument_methods():
app = Sanic('test_argument_methods')
class DummyView(HTTPMethodView):
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>')
request, response = sanic_endpoint_test(app, uri='/test123')
assert response.text == 'I am get method with test123'
def test_with_bp():
app = Sanic('test_with_bp')
bp = Blueprint('test_text')
class DummyView(HTTPMethodView):
def get(self, request):
return text('I am get method')
bp.add_route(DummyView(), '/')
app.blueprint(bp)
request, response = sanic_endpoint_test(app)
assert response.text == 'I am get method'
def test_with_bp_with_url_prefix():
app = Sanic('test_with_bp_with_url_prefix')
bp = Blueprint('test_text', url_prefix='/test1')
class DummyView(HTTPMethodView):
def get(self, request):
return text('I am get method')
bp.add_route(DummyView(), '/')
app.blueprint(bp)
request, response = sanic_endpoint_test(app, uri='/test1/')
assert response.text == 'I am get method'
def test_with_middleware():
app = Sanic('test_with_middleware')
class DummyView(HTTPMethodView):
def get(self, request):
return text('I am get method')
app.add_route(DummyView(), '/')
results = []
@app.middleware
async def handler(request):
results.append(request)
request, response = sanic_endpoint_test(app)
assert response.text == 'I am get method'
assert type(results[0]) is Request
def test_with_middleware_response():
app = Sanic('test_with_middleware_response')
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)
class DummyView(HTTPMethodView):
def get(self, request):
return text('I am get method')
app.add_route(DummyView(), '/')
request, response = sanic_endpoint_test(app)
assert response.text == 'I am get method'
assert type(results[0]) is Request
assert type(results[1]) is Request
assert issubclass(type(results[2]), HTTPResponse)