Merge pull request #24 from huge-success/master

merge upstream master branch
This commit is contained in:
7 2018-10-06 21:22:54 -07:00 committed by GitHub
commit cc83c1f0cf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
55 changed files with 812 additions and 706 deletions

View File

@ -1,7 +1,7 @@
[run]
branch = True
source = sanic
omit = site-packages, sanic/utils.py
omit = site-packages, sanic/utils.py, sanic/__main__.py
[html]
directory = coverage

View File

@ -31,7 +31,7 @@ deploy:
provider: pypi
user: channelcat
password:
secure: OgADRQH3+dTL5swGzXkeRJDNbLpFzwqYnXB4iLD0Npvzj9QnKyQVvkbaeq6VmV9dpEFb5ULaAKYQq19CrXYDm28yanUSn6jdJ4SukaHusi7xt07U6H7pmoX/uZ2WZYqCSLM8cSp8TXY/3oV3rY5Jfj/AibE5XTbim5/lrhsvW6NR+ALzxc0URRPAHDZEPpojTCjSTjpY0aDsaKWg4mXVRMFfY3O68j6KaIoukIZLuoHfePLKrbZxaPG5VxNhMHEaICdxVxE/dO+7pQmQxXuIsEOHK1QiVJ9YrSGcNqgEqhN36kYP8dqMeVB07sv8Xa6o/Uax2/wXS2HEJvuwP1YD6WkoZuo9ZB85bcMdg7BV9jJDbVFVPJwc75BnTLHrMa3Q1KrRlKRDBUXBUsQivPuWhFNwUgvEayq2qSI3aRQR4Z0O+DfboEhXYojSoD64/EWBTZ7vhgbvOTGEdukUQSYrKj9P8jc1s8exomTsAiqdFxTUpzfiammUSL+M93lP4urtahl1jjXFX7gd3DzdEEb0NsGkx5lm/qdsty8/TeAvKUmC+RVU6T856W6MqN0P+yGbpWUARcSE7fwztC3SPxwAuxvIN3BHmRhOUHoORPNG2VpfbnscIzBKJR4v0JKzbpi0IDa66K+tCGsCEvQuL4cxVOtoUySPWNSUAyUWWUrGM2k=
secure: h7oNDjA/ObDBGK7xt55SV0INHOclMJW/byxMrNxvCZ0JxiRk7WBNtWYt0WJjyf5lO/L0/sfgiAk0GIdFon57S24njSLPAq/a4ptkWZ68s2A+TaF6ezJSZvE9V8khivjoeub90TzfX6P5aukRja1CSxXKJm+v0V8hGE4CZGyCgEDvK3JqIakpXllSDl19DhVftCS/lQZD7AXrZlg1kZnPCMtB5IbCVR4L2bfrSJVNptBi2CqqxacY2MOLu+jv5FzJ2BGVIJ2zoIJS2T+JmGJzpiamF6y8Amv0667i9lg2DXWCtI3PsQzCmwa3F/ZsI+ohUAvJC5yvzP7SyTJyXifRBdJ9O137QkNAHFoJOOY3B4GSnTo8/boajKXEqGiV4h2EgwNjBaR0WJl0pB7HHUCBMkNRWqo6ACB8eCr04tXWXPvkGIc+wPjq960hsUZea1O31MuktYc9Ot6eiFqm7OKoItdi7LxCen1eTj93ePgkiEnVZ+p/04Hh1U7CX31UJMNu5kCvZPIANnAuDsS2SK7Qkr88OAuWL0wmrBcXKOcnVkJtZ5mzx8T54bI1RrSYtFDBLFfOPb0GucSziMBtQpE76qPEauVwIXBk3RnR8N57xBR/lvTaIk758tf+haO0llEO5rVls1zLNZ+VlTzXy7hX0OZbdopIAcCFBFWqWMAdXQc=
on:
tags: true
distributions: "sdist bdist_wheel"

View File

@ -1,3 +1,80 @@
Version 0.8
-----------
0.8.3
- Changes:
- Ownership changed to org 'huge-success'
0.8.0
- Changes:
- Add Server-Sent Events extension (Innokenty Lebedev)
- Graceful handling of request_handler_task cancellation (Ashley Sommer)
- Sanitize URL before redirection (aveao)
- Add url_bytes to request (johndoe46)
- py37 support for travisci (yunstanford)
- Auto reloader support for OSX (garyo)
- Add UUID route support (Volodymyr Maksymiv)
- Add pausable response streams (Ashley Sommer)
- Add weakref to request slots (vopankov)
- remove ubuntu 12.04 from test fixture due to deprecation (yunstanford)
- Allow streaming handlers in add_route (kinware)
- use travis_retry for tox (Raphael Deem)
- update aiohttp version for test client (yunstanford)
- add redirect import for clarity (yingshaoxo)
- Update HTTP Entity headers (Arnulfo Solís)
- Add register_listener method (Stephan Fitzpatrick)
- Remove uvloop/ujson dependencies for Windows (abuckenheimer)
- Content-length header on 204/304 responses (Arnulfo Solís)
- Extend WebSocketProtocol arguments and add docs (Bob Olde Hampsink, yunstanford)
- Update development status from pre-alpha to beta (Maksim Anisenkov)
- KeepAlive Timout log level changed to debug (Arnulfo Solís)
- Pin pytest to 3.3.2 because of pytest-dev/pytest#3170 (Maksim Aniskenov)
- Install Python 3.5 and 3.6 on docker container for tests (Shahin Azad)
- Add support for blueprint groups and nesting (Elias Tarhini)
- Remove uvloop for windows setup (Aleksandr Kurlov)
- Auto Reload (Yaser Amari)
- Documentation updates/fixups (multiple contributors)
- Fixes:
- Fix: auto_reload in Linux (Ashley Sommer)
- Fix: broken tests for aiohttp >= 3.3.0 (Ashley Sommer)
- Fix: disable auto_reload by default on windows (abuckenheimer)
- Fix (1143): Turn off access log with gunicorn (hqy)
- Fix (1268): Support status code for file response (Cosmo Borsky)
- Fix (1266): Add content_type flag to Sanic.static (Cosmo Borsky)
- Fix: subprotocols parameter missing from add_websocket_route (ciscorn)
- Fix (1242): Responses for CI header (yunstanford)
- Fix (1237): add version constraint for websockets (yunstanford)
- Fix (1231): memory leak - always release resource (Phillip Xu)
- Fix (1221): make request truthy if transport exists (Raphael Deem)
- Fix failing tests for aiohttp>=3.1.0 (Ashley Sommer)
- Fix try_everything examples (PyManiacGR, kot83)
- Fix (1158): default to auto_reload in debug mode (Raphael Deem)
- Fix (1136): ErrorHandler.response handler call too restrictive (Julien Castiaux)
- Fix: raw requires bytes-like object (cloudship)
- Fix (1120): passing a list in to a route decorator's host arg (Timothy Ebiuwhe)
- Fix: Bug in multipart/form-data parser (DirkGuijt)
- Fix: Exception for missing parameter when value is null (NyanKiyoshi)
- Fix: Parameter check (Howie Hu)
- Fix (1089): Routing issue with named parameters and different methods (yunstanford)
- Fix (1085): Signal handling in multi-worker mode (yunstanford)
- Fix: single quote in readme.rst (Cosven)
- Fix: method typos (Dmitry Dygalo)
- Fix: log_response correct output for ip and port (Wibowo Arindrarto)
- Fix (1042): Exception Handling (Raphael Deem)
- Fix: Chinese URIs (Howie Hu)
- Fix (1079): timeout bug when self.transport is None (Raphael Deem)
- Fix (1074): fix strict_slashes when route has slash (Raphael Deem)
- Fix (1050): add samesite cookie to cookie keys (Raphael Deem)
- Fix (1065): allow add_task after server starts (Raphael Deem)
- Fix (1061): double quotes in unauthorized exception (Raphael Deem)
- Fix (1062): inject the app in add_task method (Raphael Deem)
- Fix: update environment.yml for readthedocs (Eli Uriegas)
- Fix: Cancel request task when response timeout is triggered (Jeong YunWon)
- Fix (1052): Method not allowed response for RFC7231 compliance (Raphael Deem)
- Fix: IPv6 Address and Socket Data Format (Dan Palmer)
Note: Changelog was unmaintained between 0.1 and 0.7
Version 0.1
-----------
- 0.1.7

View File

@ -7,9 +7,9 @@ Sanic is a Flask-like Python 3.5+ web server that's written to go fast. It's ba
On top of being Flask-like, Sanic supports async request handlers. This means you can use the new shiny async/await syntax from Python 3.5, making your code non-blocking and speedy.
Sanic is developed `on GitHub <https://github.com/channelcat/sanic/>`_. Contributions are welcome!
Sanic is developed `on GitHub <https://github.com/huge-success/sanic/>`_. We also have `a community discussion board <https://community.sanicframework.org/>`_. Contributions are welcome!
If you have a project that utilizes Sanic make sure to comment on the `issue <https://github.com/channelcat/sanic/issues/396>`_ that we use to track those projects!
If you have a project that utilizes Sanic make sure to comment on the `issue <https://github.com/huge-success/sanic/issues/396>`_ that we use to track those projects!
Hello World Example
-------------------
@ -47,8 +47,8 @@ Documentation
.. |Join the chat at https://gitter.im/sanic-python/Lobby| image:: https://badges.gitter.im/sanic-python/Lobby.svg
:target: https://gitter.im/sanic-python/Lobby?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge
.. |Build Status| image:: https://travis-ci.org/channelcat/sanic.svg?branch=master
:target: https://travis-ci.org/channelcat/sanic
.. |Build Status| image:: https://travis-ci.org/huge-success/sanic.svg?branch=master
:target: https://travis-ci.org/huge-success/sanic
.. |Documentation| image:: https://readthedocs.org/projects/sanic/badge/?version=latest
:target: http://sanic.readthedocs.io/en/latest/?badge=latest
.. |PyPI| image:: https://img.shields.io/pypi/v/sanic.svg
@ -57,23 +57,21 @@ Documentation
:target: https://pypi.python.org/pypi/sanic/
Questions and Discussion
------------------------
`Ask a question or join the conversation <https://community.sanicframework.org/>`_.
Examples
--------
`Non-Core examples <https://github.com/channelcat/sanic/wiki/Examples/>`_. Examples of plugins and Sanic that are outside the scope of Sanic core.
`Non-Core examples <https://github.com/huge-success/sanic/wiki/Examples/>`_. Examples of plugins and Sanic that are outside the scope of Sanic core.
`Extensions <https://github.com/channelcat/sanic/wiki/Extensions/>`_. Sanic extensions created by the community.
`Extensions <https://github.com/huge-success/sanic/wiki/Extensions/>`_. Sanic extensions created by the community.
`Projects <https://github.com/channelcat/sanic/wiki/Projects/>`_. Sanic in production use.
`Projects <https://github.com/huge-success/sanic/wiki/Projects/>`_. Sanic in production use.
TODO
----
* http2
Limitations
-----------
* No wheels for uvloop and httptools on Windows :(
Final Thoughts
--------------

View File

@ -31,3 +31,4 @@ A list of Sanic extensions created by the community.
- [Sanic-Auth](https://github.com/pyx/sanic-auth): A minimal backend agnostic session-based user authentication mechanism for Sanic.
- [Sanic-CookieSession](https://github.com/pyx/sanic-cookiesession): A client-side only, cookie-based session, similar to the built-in session in Flask.
- [Sanic-WTF](https://github.com/pyx/sanic-wtf): Sanic-WTF makes using WTForms with Sanic and CSRF (Cross-Site Request Forgery) protection a little bit easier.
- [sanic-sse](https://github.com/inn0kenty/sanic_sse): [Server-Sent Events](https://en.wikipedia.org/wiki/Server-sent_events) implementation for Sanic.

View File

@ -37,7 +37,7 @@ async def handler(request):
if body is None:
break
body = body.decode('utf-8').replace('1', 'A')
response.write(body)
await response.write(body)
return stream(streaming)
@ -85,8 +85,8 @@ app = Sanic(__name__)
@app.route("/")
async def test(request):
async def sample_streaming_fn(response):
response.write('foo,')
response.write('bar')
await response.write('foo,')
await response.write('bar')
return stream(sample_streaming_fn, content_type='text/csv')
```
@ -100,7 +100,7 @@ async def index(request):
conn = await asyncpg.connect(database='test')
async with conn.transaction():
async for record in conn.cursor('SELECT generate_series(0, 10)'):
response.write(record[0])
await response.write(record[0])
return stream(stream_from_db)
```

View File

@ -12,9 +12,10 @@ dependencies:
- zlib=1.2.8=0
- pip:
- uvloop>=0.5.3
- httptools>=0.0.9
- httptools>=0.0.10
- ujson>=1.35
- aiofiles>=0.3.0
- websockets>=3.2
- websockets>=6.0
- sphinxcontrib-asyncio>=0.2.0
- multidict>=4.0<5.0
- https://github.com/channelcat/docutils-fork/zipball/master

View File

@ -1,7 +1,5 @@
from sanic import Sanic
from sanic import Blueprint
from sanic.response import json
from sanic import Blueprint, Sanic
from sanic.response import file, json
app = Sanic(__name__)
blueprint = Blueprint('name', url_prefix='/my_blueprint')
@ -19,7 +17,12 @@ async def foo2(request):
return json({'msg': 'hi from blueprint2'})
@blueprint3.websocket('/foo')
@blueprint3.route('/foo')
async def index(request):
return await file('websocket.html')
@app.websocket('/feed')
async def foo3(request, ws):
while True:
data = 'hello!'

View File

@ -30,7 +30,7 @@ async def handler(request):
if body is None:
break
body = body.decode('utf-8').replace('1', 'A')
response.write(body)
await response.write(body)
return stream(streaming)

View File

@ -0,0 +1,42 @@
from sanic import Sanic
from sanic.views import HTTPMethodView
from sanic.response import text
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')
class SimpleAsyncView(HTTPMethodView):
async def get(self, request):
return text('I am async get method')
async def post(self, request):
return text('I am async post method')
async def put(self, request):
return text('I am async put method')
app.add_route(SimpleView.as_view(), '/')
app.add_route(SimpleAsyncView.as_view(), '/async')
if __name__ == '__main__':
app.run(host="0.0.0.0", port=8000, debug=True)

View File

@ -1,12 +1,13 @@
aiofiles
aiohttp>=2.3.0
aiohttp>=2.3.0,<=3.2.1
chardet<=2.3.0
beautifulsoup4
coverage
httptools
httptools>=0.0.10
flake8
pytest==3.3.2
tox
ujson; sys_platform != "win32" and implementation_name == "cpython"
uvloop; sys_platform != "win32" and implementation_name == "cpython"
gunicorn
multidict>=4.0,<5.0

View File

@ -1,5 +1,6 @@
aiofiles
httptools
httptools>=0.0.10
ujson; sys_platform != "win32" and implementation_name == "cpython"
uvloop; sys_platform != "win32" and implementation_name == "cpython"
websockets>=5.0,<6.0
websockets>=6.0,<7.0
multidict>=4.0,<5.0

View File

@ -1,6 +1,6 @@
from sanic.app import Sanic
from sanic.blueprints import Blueprint
__version__ = '0.7.0'
__version__ = '0.8.3'
__all__ = ['Sanic', 'Blueprint']

View File

@ -315,13 +315,13 @@ class Sanic:
return response
def add_websocket_route(self, handler, uri, host=None,
strict_slashes=None, name=None):
strict_slashes=None, subprotocols=None, name=None):
"""A helper method to register a function as a websocket route."""
if strict_slashes is None:
strict_slashes = self.strict_slashes
return self.websocket(uri, host=host, strict_slashes=strict_slashes,
name=name)(handler)
subprotocols=subprotocols, name=name)(handler)
def enable_websocket(self, enable=True):
"""Enable or disable the support for websocket.
@ -386,13 +386,14 @@ class Sanic:
def static(self, uri, file_or_directory, pattern=r'/?.+',
use_modified_since=True, use_content_range=False,
stream_large_files=False, name='static', host=None,
strict_slashes=None):
strict_slashes=None, content_type=None):
"""Register a root to serve files from. The input can either be a
file or a directory. See
"""
static_register(self, uri, file_or_directory, pattern,
use_modified_since, use_content_range,
stream_large_files, name, host, strict_slashes)
stream_large_files, name, host, strict_slashes,
content_type)
def blueprint(self, blueprint, **options):
"""Register a blueprint on the application.
@ -570,6 +571,10 @@ class Sanic:
:return: Nothing
"""
# Define `response` var here to remove warnings about
# allocation before assignment below.
response = None
cancelled = False
try:
# -------------------------------------------- #
# Request Middleware
@ -596,6 +601,13 @@ class Sanic:
response = handler(request, *args, **kwargs)
if isawaitable(response):
response = await response
except CancelledError:
# If response handler times out, the server handles the error
# and cancels the handle_request job.
# In this case, the transport is already closed and we cannot
# issue a response.
response = None
cancelled = True
except Exception as e:
# -------------------------------------------- #
# Response Generation Failed
@ -621,13 +633,22 @@ class Sanic:
# -------------------------------------------- #
# Response Middleware
# -------------------------------------------- #
# Don't run response middleware if response is None
if response is not None:
try:
response = await self._run_response_middleware(request,
response)
except CancelledError:
# Response middleware can timeout too, as above.
response = None
cancelled = True
except BaseException:
error_logger.exception(
'Exception occurred in one of response middleware handlers'
'Exception occurred in one of response '
'middleware handlers'
)
if cancelled:
raise CancelledError()
# pass the response to the correct callback
if isinstance(response, StreamingHTTPResponse):
@ -670,8 +691,8 @@ class Sanic:
"""
# Default auto_reload to false
auto_reload = False
# If debug is set, default it to true
if debug:
# If debug is set, default it to true (unless on windows)
if debug and os.name == 'posix':
auto_reload = True
# Allow for overriding either of the defaults
auto_reload = kwargs.get("auto_reload", auto_reload)
@ -687,11 +708,12 @@ class Sanic:
warnings.simplefilter('default')
warnings.warn("stop_event will be removed from future versions.",
DeprecationWarning)
# compatibility old access_log params
self.config.ACCESS_LOG = access_log
server_settings = self._helper(
host=host, port=port, debug=debug, ssl=ssl, sock=sock,
workers=workers, protocol=protocol, backlog=backlog,
register_sys_signals=register_sys_signals,
access_log=access_log, auto_reload=auto_reload)
register_sys_signals=register_sys_signals, auto_reload=auto_reload)
try:
self.is_running = True
@ -745,12 +767,12 @@ class Sanic:
warnings.simplefilter('default')
warnings.warn("stop_event will be removed from future versions.",
DeprecationWarning)
# compatibility old access_log params
self.config.ACCESS_LOG = access_log
server_settings = self._helper(
host=host, port=port, debug=debug, ssl=ssl, sock=sock,
loop=get_event_loop(), protocol=protocol,
backlog=backlog, run_async=True,
access_log=access_log)
backlog=backlog, run_async=True)
# Trigger before_start events
await self.trigger_events(
@ -795,8 +817,7 @@ class Sanic:
def _helper(self, host=None, port=None, debug=False,
ssl=None, sock=None, workers=1, loop=None,
protocol=HttpProtocol, backlog=100, stop_event=None,
register_sys_signals=True, run_async=False, access_log=True,
auto_reload=False):
register_sys_signals=True, run_async=False, auto_reload=False):
"""Helper function used by `run` and `create_server`."""
if isinstance(ssl, dict):
# try common aliaseses
@ -837,7 +858,7 @@ class Sanic:
'loop': loop,
'register_sys_signals': register_sys_signals,
'backlog': backlog,
'access_log': access_log,
'access_log': self.config.ACCESS_LOG,
'websocket_max_size': self.config.WEBSOCKET_MAX_SIZE,
'websocket_max_queue': self.config.WEBSOCKET_MAX_QUEUE,
'websocket_read_limit': self.config.WEBSOCKET_READ_LIMIT,

View File

@ -39,6 +39,7 @@ class Config(dict):
self.WEBSOCKET_READ_LIMIT = 2 ** 16
self.WEBSOCKET_WRITE_LIMIT = 2 ** 16
self.GRACEFUL_SHUTDOWN_TIMEOUT = 15.0 # 15 sec
self.ACCESS_LOG = True
if load_env:
prefix = SANIC_PREFIX if load_env is True else load_env

View File

@ -47,16 +47,15 @@ class CookieJar(dict):
super().__init__()
self.headers = headers
self.cookie_headers = {}
self.header_key = "Set-Cookie"
def __setitem__(self, key, value):
# If this cookie doesn't exist, add it to the header keys
cookie_header = self.cookie_headers.get(key)
if not cookie_header:
if not self.cookie_headers.get(key):
cookie = Cookie(key, value)
cookie['path'] = '/'
cookie_header = MultiHeader("Set-Cookie")
self.cookie_headers[key] = cookie_header
self.headers[cookie_header] = cookie
self.cookie_headers[key] = self.header_key
self.headers.add(self.header_key, cookie)
return super().__setitem__(key, cookie)
else:
self[key].value = value
@ -67,7 +66,11 @@ class CookieJar(dict):
self[key]['max-age'] = 0
else:
cookie_header = self.cookie_headers[key]
del self.headers[cookie_header]
# remove it from header
cookies = self.headers.popall(cookie_header)
for cookie in cookies:
if cookie.key != key:
self.headers.add(cookie_header, cookie)
del self.cookie_headers[key]
return super().__delitem__(key)
@ -124,18 +127,3 @@ class Cookie(dict):
output.append('%s=%s' % (self._keys[key], value))
return "; ".join(output).encode(encoding)
# ------------------------------------------------------------ #
# Header Trickery
# ------------------------------------------------------------ #
class MultiHeader:
"""String-holding object which allow us to set a header within response
that has a unique key, but may contain duplicate header names
"""
def __init__(self, name):
self.name = name
def encode(self):
return self.name.encode()

View File

@ -74,7 +74,14 @@ def kill_process_children_unix(pid):
with open(children_proc_path) as children_list_file_2:
children_list_pid_2 = children_list_file_2.read().split()
for _pid in children_list_pid_2:
try:
os.kill(int(_pid), signal.SIGTERM)
except ProcessLookupError:
continue
try:
os.kill(int(child_pid), signal.SIGTERM)
except ProcessLookupError:
continue
def kill_process_children_osx(pid):
@ -94,7 +101,7 @@ def kill_process_children(pid):
"""
if sys.platform == 'darwin':
kill_process_children_osx(pid)
elif sys.platform == 'posix':
elif sys.platform == 'linux':
kill_process_children_unix(pid)
else:
pass # should signal error here
@ -136,8 +143,8 @@ def watchdog(sleep_interval):
continue
elif mtime > old_time:
kill_process_children(worker_process.pid)
worker_process.terminate()
worker_process = restart_with_reloader()
mtimes[filename] = mtime
break

View File

@ -48,10 +48,11 @@ class Request(dict):
'app', 'headers', 'version', 'method', '_cookies', 'transport',
'body', 'parsed_json', 'parsed_args', 'parsed_form', 'parsed_files',
'_ip', '_parsed_url', 'uri_template', 'stream', '_remote_addr',
'_socket', '_port', '__weakref__'
'_socket', '_port', '__weakref__', 'raw_url'
)
def __init__(self, url_bytes, headers, version, method, transport):
self.raw_url = url_bytes
# TODO: Content-Encoding detection
self._parsed_url = parse_url(url_bytes)
self.app = None

View File

@ -1,5 +1,6 @@
from mimetypes import guess_type
from os import path
from urllib.parse import quote_plus
try:
from ujson import dumps as json_dumps
@ -7,6 +8,7 @@ except BaseException:
from json import dumps as json_dumps
from aiofiles import open as open_async
from multidict import CIMultiDict
from sanic import http
from sanic.cookies import CookieJar
@ -44,7 +46,7 @@ class BaseHTTPResponse:
class StreamingHTTPResponse(BaseHTTPResponse):
__slots__ = (
'transport', 'streaming_fn', 'status',
'protocol', 'streaming_fn', 'status',
'content_type', 'headers', '_cookies'
)
@ -53,10 +55,10 @@ class StreamingHTTPResponse(BaseHTTPResponse):
self.content_type = content_type
self.streaming_fn = streaming_fn
self.status = status
self.headers = headers or {}
self.headers = CIMultiDict(headers or {})
self._cookies = None
def write(self, data):
async def write(self, data):
"""Writes a chunk of data to the streaming response.
:param data: bytes-ish data to be written.
@ -64,8 +66,9 @@ class StreamingHTTPResponse(BaseHTTPResponse):
if type(data) != bytes:
data = self._encode_body(data)
self.transport.write(
self.protocol.push_data(
b"%x\r\n%b\r\n" % (len(data), data))
await self.protocol.drain()
async def stream(
self, version="1.1", keep_alive=False, keep_alive_timeout=None):
@ -75,10 +78,12 @@ class StreamingHTTPResponse(BaseHTTPResponse):
headers = self.get_headers(
version, keep_alive=keep_alive,
keep_alive_timeout=keep_alive_timeout)
self.transport.write(headers)
self.protocol.push_data(headers)
await self.protocol.drain()
await self.streaming_fn(self)
self.transport.write(b'0\r\n\r\n')
self.protocol.push_data(b'0\r\n\r\n')
# no need to await drain here after this write, because it is the
# very last thing we write and nothing needs to wait for it.
def get_headers(
self, version="1.1", keep_alive=False, keep_alive_timeout=None):
@ -124,7 +129,7 @@ class HTTPResponse(BaseHTTPResponse):
self.body = body_bytes
self.status = status
self.headers = headers or {}
self.headers = CIMultiDict(headers or {})
self._cookies = None
def output(
@ -231,8 +236,8 @@ def html(body, status=200, headers=None):
content_type="text/html; charset=utf-8")
async def file(
location, mime_type=None, headers=None, filename=None, _range=None):
async def file(location, status=200, mime_type=None, headers=None,
filename=None, _range=None):
"""Return a response object with file data.
:param location: Location of file on system.
@ -258,15 +263,14 @@ async def file(
out_stream = await _file.read()
mime_type = mime_type or guess_type(filename)[0] or 'text/plain'
return HTTPResponse(status=200,
return HTTPResponse(status=status,
headers=headers,
content_type=mime_type,
body_bytes=out_stream)
async def file_stream(
location, chunk_size=4096, mime_type=None, headers=None,
filename=None, _range=None):
async def file_stream(location, status=200, chunk_size=4096, mime_type=None,
headers=None, filename=None, _range=None):
"""Return a streaming response object with file data.
:param location: Location of file on system.
@ -297,13 +301,13 @@ async def file_stream(
if len(content) < 1:
break
to_send -= len(content)
response.write(content)
await response.write(content)
else:
while True:
content = await _file.read(chunk_size)
if len(content) < 1:
break
response.write(content)
await response.write(content)
finally:
await _file.close()
return # Returning from this fn closes the stream
@ -313,7 +317,7 @@ async def file_stream(
headers['Content-Range'] = 'bytes %s-%s/%s' % (
_range.start, _range.end, _range.total)
return StreamingHTTPResponse(streaming_fn=_streaming_fn,
status=200,
status=status,
headers=headers,
content_type=mime_type)
@ -359,8 +363,11 @@ def redirect(to, headers=None, status=302,
"""
headers = headers or {}
# URL Quote the URL before redirecting
safe_to = quote_plus(to, safe=":/#?&=@[]!$&'()*+,;")
# According to RFC 7231, a relative URI is now permitted.
headers['Location'] = to
headers['Location'] = safe_to
return HTTPResponse(
status=status,

View File

@ -18,6 +18,7 @@ from time import time
from httptools import HttpRequestParser
from httptools.parser.errors import HttpParserError
from multidict import CIMultiDict
try:
import uvloop
@ -39,25 +40,6 @@ class Signal:
stopped = False
class CIDict(dict):
"""Case Insensitive dict where all keys are converted to lowercase
This does not maintain the inputted case when calling items() or keys()
in favor of speed, since headers are case insensitive
"""
def get(self, key, default=None):
return super().get(key.casefold(), default)
def __getitem__(self, key):
return super().__getitem__(key.casefold())
def __setitem__(self, key, value):
return super().__setitem__(key.casefold(), value)
def __contains__(self, key):
return super().__contains__(key.casefold())
class HttpProtocol(asyncio.Protocol):
__slots__ = (
# event loop, connection
@ -73,7 +55,8 @@ class HttpProtocol(asyncio.Protocol):
# connection management
'_total_request_size', '_request_timeout_handler',
'_response_timeout_handler', '_keep_alive_timeout_handler',
'_last_request_time', '_last_response_time', '_is_stream_handler')
'_last_request_time', '_last_response_time', '_is_stream_handler',
'_not_paused')
def __init__(self, *, loop, request_handler, error_handler,
signal=Signal(), connections=set(), request_timeout=60,
@ -100,6 +83,7 @@ class HttpProtocol(asyncio.Protocol):
self.request_class = request_class or Request
self.is_request_stream = is_request_stream
self._is_stream_handler = False
self._not_paused = asyncio.Event(loop=loop)
self._total_request_size = 0
self._request_timeout_handler = None
self._response_timeout_handler = None
@ -114,6 +98,7 @@ class HttpProtocol(asyncio.Protocol):
if 'requests_count' not in self.state:
self.state['requests_count'] = 0
self._debug = debug
self._not_paused.set()
@property
def keep_alive(self):
@ -142,6 +127,12 @@ class HttpProtocol(asyncio.Protocol):
if self._keep_alive_timeout_handler:
self._keep_alive_timeout_handler.cancel()
def pause_writing(self):
self._not_paused.clear()
def resume_writing(self):
self._not_paused.set()
def request_timeout_callback(self):
# See the docstring in the RequestTimeout exception, to see
# exactly what this timeout is checking for.
@ -159,10 +150,7 @@ class HttpProtocol(asyncio.Protocol):
self._request_stream_task.cancel()
if self._request_handler_task:
self._request_handler_task.cancel()
try:
raise RequestTimeout('Request Timeout')
except RequestTimeout as exception:
self.write_error(exception)
self.write_error(RequestTimeout('Request Timeout'))
def response_timeout_callback(self):
# Check if elapsed time since response was initiated exceeds our
@ -179,10 +167,7 @@ class HttpProtocol(asyncio.Protocol):
self._request_stream_task.cancel()
if self._request_handler_task:
self._request_handler_task.cancel()
try:
raise ServiceUnavailable('Response Timeout')
except ServiceUnavailable as exception:
self.write_error(exception)
self.write_error(ServiceUnavailable('Response Timeout'))
def keep_alive_timeout_callback(self):
# Check if elapsed time since last response exceeds our configured
@ -208,8 +193,7 @@ class HttpProtocol(asyncio.Protocol):
# memory limits
self._total_request_size += len(data)
if self._total_request_size > self.request_max_size:
exception = PayloadTooLarge('Payload Too Large')
self.write_error(exception)
self.write_error(PayloadTooLarge('Payload Too Large'))
# Create parser if this is the first time we're receiving data
if self.parser is None:
@ -227,8 +211,7 @@ class HttpProtocol(asyncio.Protocol):
message = 'Bad Request'
if self._debug:
message += '\n' + traceback.format_exc()
exception = InvalidUsage(message)
self.write_error(exception)
self.write_error(InvalidUsage(message))
def on_url(self, url):
if not self.url:
@ -242,8 +225,7 @@ class HttpProtocol(asyncio.Protocol):
if value is not None:
if self._header_fragment == b'Content-Length' \
and int(value) > self.request_max_size:
exception = PayloadTooLarge('Payload Too Large')
self.write_error(exception)
self.write_error(PayloadTooLarge('Payload Too Large'))
try:
value = value.decode()
except UnicodeDecodeError:
@ -256,7 +238,7 @@ class HttpProtocol(asyncio.Protocol):
def on_headers_complete(self):
self.request = self.request_class(
url_bytes=self.url,
headers=CIDict(self.headers),
headers=CIMultiDict(self.headers),
version=self.parser.get_http_version(),
method=self.parser.get_method().decode(),
transport=self.transport
@ -369,6 +351,12 @@ class HttpProtocol(asyncio.Protocol):
self._last_response_time = current_time
self.cleanup()
async def drain(self):
await self._not_paused.wait()
def push_data(self, data):
self.transport.write(data)
async def stream_response(self, response):
"""
Streams a response to the client asynchronously. Attaches
@ -378,9 +366,10 @@ class HttpProtocol(asyncio.Protocol):
if self._response_timeout_handler:
self._response_timeout_handler.cancel()
self._response_timeout_handler = None
try:
keep_alive = self.keep_alive
response.transport = self.transport
response.protocol = self
await response.stream(
self.request.version, keep_alive, self.keep_alive_timeout)
self.log_response(response)
@ -435,7 +424,7 @@ class HttpProtocol(asyncio.Protocol):
self.log_response(response)
try:
self.transport.close()
except AttributeError as e:
except AttributeError:
logger.debug('Connection lost before server could close it.')
def bail_out(self, message, from_error=False):
@ -445,8 +434,7 @@ class HttpProtocol(asyncio.Protocol):
self.transport.get_extra_info('peername'))
logger.debug('Exception:\n%s', traceback.format_exc())
else:
exception = ServerError(message)
self.write_error(exception)
self.write_error(ServerError(message))
logger.error(message)
def cleanup(self):

View File

@ -19,7 +19,7 @@ from sanic.response import file, file_stream, HTTPResponse
def register(app, uri, file_or_directory, pattern,
use_modified_since, use_content_range,
stream_large_files, name='static', host=None,
strict_slashes=None):
strict_slashes=None, content_type=None):
# TODO: Though sanic is not a file server, I feel like we should at least
# make a good effort here. Modified-since is nice, but we could
# also look into etags, expires, and caching
@ -41,6 +41,7 @@ def register(app, uri, file_or_directory, pattern,
If this is an integer, this represents the
threshold size to switch to file_stream()
:param name: user defined name used for url_for
:param content_type: user defined content type for header
"""
# If we're not trying to match a file directly,
# serve from the folder
@ -95,10 +96,10 @@ def register(app, uri, file_or_directory, pattern,
del headers['Content-Length']
for key, value in _range.headers.items():
headers[key] = value
headers['Content-Type'] = content_type \
or guess_type(file_path)[0] or 'text/plain'
if request.method == 'HEAD':
return HTTPResponse(
headers=headers,
content_type=guess_type(file_path)[0] or 'text/plain')
return HTTPResponse(headers=headers)
else:
if stream_large_files:
if isinstance(stream_large_files, int):

View File

@ -57,17 +57,11 @@ class WebSocketProtocol(HttpProtocol):
async def websocket_handshake(self, request, subprotocols=None):
# let the websockets package do the handshake with the client
headers = []
def get_header(k):
return request.headers.get(k, '')
def set_header(k, v):
headers.append((k, v))
headers = {}
try:
key = handshake.check_request(get_header)
handshake.build_response(set_header, key)
key = handshake.check_request(request.headers)
handshake.build_response(headers, key)
except InvalidHandshake:
raise InvalidUsage('Invalid websocket request')
@ -79,12 +73,12 @@ class WebSocketProtocol(HttpProtocol):
for p in client_subprotocols:
if p in subprotocols:
subprotocol = p
set_header('Sec-Websocket-Protocol', subprotocol)
headers['Sec-Websocket-Protocol'] = subprotocol
break
# write the 101 response back to the client
rv = b'HTTP/1.1 101 Switching Protocols\r\n'
for k, v in headers:
for k, v in headers.items():
rv += k.encode('utf-8') + b': ' + v.encode('utf-8') + b'\r\n'
rv += b'\r\n'
request.transport.write(rv)

View File

@ -56,11 +56,12 @@ ujson = 'ujson>=1.35' + env_dependency
uvloop = 'uvloop>=0.5.3' + env_dependency
requirements = [
'httptools>=0.0.9',
'httptools>=0.0.10',
uvloop,
ujson,
'aiofiles>=0.3.0',
'websockets>=5.0,<6.0',
'websockets>=6.0,<7.0',
'multidict>=4.0,<5.0',
]
if strtobool(os.environ.get("SANIC_NO_UJSON", "no")):
print("Installing without uJSON")

8
tests/conftest.py Normal file
View File

@ -0,0 +1,8 @@
import pytest
from sanic import Sanic
@pytest.fixture
def app(request):
return Sanic(request.node.name)

26
tests/static/test.html Normal file
View File

@ -0,0 +1,26 @@
<html>
<body>
<pre>
▄▄▄▄▄
▀▀▀██████▄▄▄ _______________
▄▄▄▄▄ █████████▄ / \
▀▀▀▀█████▌ ▀▐▄ ▀▐█ | Gotta go fast! |
▀▀█████▄▄ ▀██████▄██ | _________________/
▀▄▄▄▄▄ ▀▀█▄▀█════█▀ |/
▀▀▀▄ ▀▀███ ▀ ▄▄
▄███▀▀██▄████████▄ ▄▀▀▀▀▀▀█▌
██▀▄▄▄██▀▄███▀ ▀▀████ ▄██
▄▀▀▀▄██▄▀▀▌████▒▒▒▒▒▒███ ▌▄▄▀
▌ ▐▀████▐███▒▒▒▒▒▐██▌
▀▄▄▄▄▀ ▀▀████▒▒▒▒▄██▀
▀▀█████████▀
▄▄██▀██████▀█
▄██▀ ▀▀▀ █
▄█ ▐▌
▄▄▄▄█▌ ▀█▄▄▄▄▀▀▄
▌ ▐ ▀▀▄▄▄▀
▀▀▄▄▀
</pre>
</body>
</html>

View File

@ -1,9 +1,7 @@
import asyncio
from sanic import Sanic
def test_bad_request_response():
app = Sanic('test_bad_request_response')
def test_bad_request_response(app):
lines = []
@app.listener('after_server_start')
async def _request(sanic, loop):

View File

@ -1,10 +1,10 @@
import asyncio
import inspect
import os
import pytest
from sanic import Sanic
from sanic.blueprints import Blueprint
from sanic.response import json, text
from sanic.response import text
from sanic.exceptions import NotFound, ServerError, InvalidUsage
from sanic.constants import HTTP_METHODS
@ -13,9 +13,16 @@ from sanic.constants import HTTP_METHODS
# GET
# ------------------------------------------------------------ #
def get_file_path(static_file_directory, file_name):
return os.path.join(static_file_directory, file_name)
def get_file_content(static_file_directory, file_name):
"""The content of the static file to check"""
with open(get_file_path(static_file_directory, file_name), 'rb') as file:
return file.read()
@pytest.mark.parametrize('method', HTTP_METHODS)
def test_versioned_routes_get(method):
app = Sanic('test_shorhand_routes_get')
def test_versioned_routes_get(app, method):
bp = Blueprint('test_text')
method = method.lower()
@ -37,8 +44,7 @@ def test_versioned_routes_get(method):
assert response.status == 200
def test_bp():
app = Sanic('test_text')
def test_bp(app):
bp = Blueprint('test_text')
@bp.route('/')
@ -51,8 +57,7 @@ def test_bp():
assert response.text == 'Hello'
def test_bp_strict_slash():
app = Sanic('test_route_strict_slash')
def test_bp_strict_slash(app):
bp = Blueprint('test_text')
@bp.get('/get', strict_slashes=True)
@ -78,8 +83,7 @@ def test_bp_strict_slash():
request, response = app.test_client.post('/post')
assert response.status == 404
def test_bp_strict_slash_default_value():
app = Sanic('test_route_strict_slash')
def test_bp_strict_slash_default_value(app):
bp = Blueprint('test_text', strict_slashes=True)
@bp.get('/get')
@ -98,8 +102,7 @@ def test_bp_strict_slash_default_value():
request, response = app.test_client.post('/post')
assert response.status == 404
def test_bp_strict_slash_without_passing_default_value():
app = Sanic('test_route_strict_slash')
def test_bp_strict_slash_without_passing_default_value(app):
bp = Blueprint('test_text')
@bp.get('/get')
@ -118,8 +121,7 @@ def test_bp_strict_slash_without_passing_default_value():
request, response = app.test_client.post('/post')
assert response.text == 'OK'
def test_bp_strict_slash_default_value_can_be_overwritten():
app = Sanic('test_route_strict_slash')
def test_bp_strict_slash_default_value_can_be_overwritten(app):
bp = Blueprint('test_text', strict_slashes=True)
@bp.get('/get', strict_slashes=False)
@ -138,8 +140,7 @@ def test_bp_strict_slash_default_value_can_be_overwritten():
request, response = app.test_client.post('/post')
assert response.text == 'OK'
def test_bp_with_url_prefix():
app = Sanic('test_text')
def test_bp_with_url_prefix(app):
bp = Blueprint('test_text', url_prefix='/test1')
@bp.route('/')
@ -152,8 +153,7 @@ def test_bp_with_url_prefix():
assert response.text == 'Hello'
def test_several_bp_with_url_prefix():
app = Sanic('test_text')
def test_several_bp_with_url_prefix(app):
bp = Blueprint('test_text', url_prefix='/test1')
bp2 = Blueprint('test_text2', url_prefix='/test2')
@ -173,8 +173,7 @@ def test_several_bp_with_url_prefix():
request, response = app.test_client.get('/test2/')
assert response.text == 'Hello2'
def test_bp_with_host():
app = Sanic('test_bp_host')
def test_bp_with_host(app):
bp = Blueprint('test_bp_host', url_prefix='/test1', host="example.com")
@bp.route('/')
@ -200,8 +199,7 @@ def test_bp_with_host():
assert response.text == 'Hello subdomain!'
def test_several_bp_with_host():
app = Sanic('test_text')
def test_several_bp_with_host(app):
bp = Blueprint('test_text',
url_prefix='/test',
host="example.com")
@ -244,8 +242,7 @@ def test_several_bp_with_host():
headers=headers)
assert response.text == 'Hello3'
def test_bp_middleware():
app = Sanic('test_middleware')
def test_bp_middleware(app):
blueprint = Blueprint('test_middleware')
@blueprint.middleware('response')
@ -263,8 +260,7 @@ def test_bp_middleware():
assert response.status == 200
assert response.text == 'OK'
def test_bp_exception_handler():
app = Sanic('test_middleware')
def test_bp_exception_handler(app):
blueprint = Blueprint('test_middleware')
@blueprint.route('/1')
@ -296,8 +292,7 @@ def test_bp_exception_handler():
request, response = app.test_client.get('/3')
assert response.status == 200
def test_bp_listeners():
app = Sanic('test_middleware')
def test_bp_listeners(app):
blueprint = Blueprint('test_middleware')
order = []
@ -332,12 +327,11 @@ def test_bp_listeners():
assert order == [1,2,3,4,5,6]
def test_bp_static():
def test_bp_static(app):
current_file = inspect.getfile(inspect.currentframe())
with open(current_file, 'rb') as file:
current_file_contents = file.read()
app = Sanic('test_static')
blueprint = Blueprint('test_static')
blueprint.static('/testing.file', current_file)
@ -348,8 +342,28 @@ def test_bp_static():
assert response.status == 200
assert response.body == current_file_contents
def test_bp_shorthand():
app = Sanic('test_shorhand_routes')
@pytest.mark.parametrize('file_name', ['test.html'])
def test_bp_static_content_type(app, file_name):
# This is done here, since no other test loads a file here
current_file = inspect.getfile(inspect.currentframe())
current_directory = os.path.dirname(os.path.abspath(current_file))
static_directory = os.path.join(current_directory, 'static')
blueprint = Blueprint('test_static')
blueprint.static(
'/testing.file',
get_file_path(static_directory, file_name),
content_type='text/html; charset=utf-8'
)
app.blueprint(blueprint)
request, response = app.test_client.get('/testing.file')
assert response.status == 200
assert response.body == get_file_content(static_directory, file_name)
assert response.headers['Content-Type'] == 'text/html; charset=utf-8'
def test_bp_shorthand(app):
blueprint = Blueprint('test_shorhand_routes')
ev = asyncio.Event()
@ -447,9 +461,7 @@ def test_bp_shorthand():
assert response.status == 101
assert ev.is_set()
def test_bp_group():
app = Sanic('test_nested_bp_groups')
def test_bp_group(app):
deep_0 = Blueprint('deep_0', url_prefix='/deep')
deep_1 = Blueprint('deep_1', url_prefix = '/deep1')

View File

@ -5,8 +5,7 @@ from tempfile import NamedTemporaryFile
from sanic import Sanic
def test_load_from_object():
app = Sanic('test_load_from_object')
def test_load_from_object(app):
class Config:
not_for_config = 'should not be used'
CONFIG_VALUE = 'should be used'
@ -16,26 +15,29 @@ def test_load_from_object():
assert app.config.CONFIG_VALUE == 'should be used'
assert 'not_for_config' not in app.config
def test_auto_load_env():
environ["SANIC_TEST_ANSWER"] = "42"
app = Sanic()
assert app.config.TEST_ANSWER == 42
del environ["SANIC_TEST_ANSWER"]
def test_dont_load_env():
environ["SANIC_TEST_ANSWER"] = "42"
app = Sanic(load_env=False)
assert getattr(app.config, 'TEST_ANSWER', None) == None
assert getattr(app.config, 'TEST_ANSWER', None) is None
del environ["SANIC_TEST_ANSWER"]
def test_load_env_prefix():
environ["MYAPP_TEST_ANSWER"] = "42"
app = Sanic(load_env='MYAPP_')
assert app.config.TEST_ANSWER == 42
del environ["MYAPP_TEST_ANSWER"]
def test_load_from_file():
app = Sanic('test_load_from_file')
def test_load_from_file(app):
config = b"""
VALUE = 'some value'
condition = 1 == 1
@ -53,14 +55,12 @@ if condition:
assert 'condition' not in app.config
def test_load_from_missing_file():
app = Sanic('test_load_from_missing_file')
def test_load_from_missing_file(app):
with pytest.raises(IOError):
app.config.from_pyfile('non-existent file')
def test_load_from_envvar():
app = Sanic('test_load_from_envvar')
def test_load_from_envvar(app):
config = b"VALUE = 'some value'"
with NamedTemporaryFile() as config_file:
config_file.write(config)
@ -71,15 +71,17 @@ def test_load_from_envvar():
assert app.config.VALUE == 'some value'
def test_load_from_missing_envvar():
app = Sanic('test_load_from_missing_envvar')
with pytest.raises(RuntimeError):
def test_load_from_missing_envvar(app):
with pytest.raises(RuntimeError) as e:
app.config.from_envvar('non-existent variable')
assert str(e.value) == ("The environment variable 'non-existent "
"variable' is not set and thus configuration "
"could not be loaded.")
def test_overwrite_exisiting_config():
app = Sanic('test_overwrite_exisiting_config')
def test_overwrite_exisiting_config(app):
app.config.DEFAULT = 1
class Config:
DEFAULT = 2
@ -87,7 +89,7 @@ def test_overwrite_exisiting_config():
assert app.config.DEFAULT == 2
def test_missing_config():
app = Sanic('test_missing_config')
with pytest.raises(AttributeError):
def test_missing_config(app):
with pytest.raises(AttributeError) as e:
app.config.NON_EXISTENT
assert str(e.value) == ("Config has no 'NON_EXISTENT'")

View File

@ -9,8 +9,7 @@ import pytest
# GET
# ------------------------------------------------------------ #
def test_cookies():
app = Sanic('test_text')
def test_cookies(app):
@app.route('/')
def handler(request):
@ -25,12 +24,12 @@ def test_cookies():
assert response.text == 'Cookies are: working!'
assert response_cookies['right_back'].value == 'at you'
@pytest.mark.parametrize("httponly,expected", [
(False, False),
(True, True),
])
def test_false_cookies_encoded(httponly, expected):
app = Sanic('test_text')
def test_false_cookies_encoded(app, httponly, expected):
@app.route('/')
def handler(request):
@ -48,8 +47,7 @@ def test_false_cookies_encoded(httponly, expected):
(False, False),
(True, True),
])
def test_false_cookies(httponly, expected):
app = Sanic('test_text')
def test_false_cookies(app, httponly, expected):
@app.route('/')
def handler(request):
@ -64,8 +62,7 @@ def test_false_cookies(httponly, expected):
assert ('HttpOnly' in response_cookies['right_back'].output()) == expected
def test_http2_cookies():
app = Sanic('test_http2_cookies')
def test_http2_cookies(app):
@app.route('/')
async def handler(request):
@ -77,8 +74,7 @@ def test_http2_cookies():
assert response.text == 'Cookies are: working!'
def test_cookie_options():
app = Sanic('test_text')
def test_cookie_options(app):
@app.route('/')
def handler(request):
@ -95,8 +91,7 @@ def test_cookie_options():
assert response_cookies['test'].value == 'at you'
assert response_cookies['test']['httponly'] == True
def test_cookie_deletion():
app = Sanic('test_text')
def test_cookie_deletion(app):
@app.route('/')
def handler(request):

View File

@ -1,18 +1,16 @@
from sanic import Sanic
from sanic.response import text
from threading import Event
import asyncio
from queue import Queue
def test_create_task():
def test_create_task(app):
e = Event()
async def coro():
await asyncio.sleep(0.05)
e.set()
app = Sanic('test_create_task')
app.add_task(coro)
@app.route('/early')
@ -30,8 +28,7 @@ def test_create_task():
request, response = app.test_client.get('/late')
assert response.body == b'True'
def test_create_task_with_app_arg():
app = Sanic('test_add_task')
def test_create_task_with_app_arg(app):
q = Queue()
@app.route('/')
@ -44,4 +41,4 @@ def test_create_task_with_app_arg():
app.add_task(coro)
request, response = app.test_client.get('/')
assert q.get() == 'test_add_task'
assert q.get() == 'test_create_task_with_app_arg'

View File

@ -1,9 +1,6 @@
from sanic import Sanic
from sanic.server import HttpProtocol
from sanic.response import text
app = Sanic('test_custom_porotocol')
class CustomHttpProtocol(HttpProtocol):
@ -16,12 +13,12 @@ class CustomHttpProtocol(HttpProtocol):
self.transport.close()
def test_use_custom_protocol(app):
@app.route('/1')
async def handler_1(request):
return 'OK'
def test_use_custom_protocol():
server_kwargs = {
'protocol': CustomHttpProtocol
}

View File

@ -10,8 +10,7 @@ import pytest
("put", "text", "OK2 test"),
("delete", "status", 405),
])
def test_overload_dynamic_routes(method, attr, expected):
app = Sanic('test_dynamic_route')
def test_overload_dynamic_routes(app, method, attr, expected):
@app.route('/overload/<param>', methods=['GET'])
async def handler1(request, param):
@ -25,8 +24,7 @@ def test_overload_dynamic_routes(method, attr, expected):
assert getattr(response, attr) == expected
def test_overload_dynamic_routes_exist():
app = Sanic('test_dynamic_route')
def test_overload_dynamic_routes_exist(app):
@app.route('/overload/<param>', methods=['GET'])
async def handler1(request, param):

View File

@ -81,8 +81,7 @@ def exception_app():
return app
def test_catch_exception_list():
app = Sanic('exception_list')
def test_catch_exception_list(app):
@app.exception([SanicExceptionTestException, NotFound])
def exception_list(request, exception):

View File

@ -9,14 +9,39 @@ import aiohttp
from aiohttp import TCPConnector
from sanic.testing import SanicTestClient, HOST, PORT
try:
try:
import packaging # direct use
except ImportError:
# setuptools v39.0 and above.
try:
from setuptools.extern import packaging
except ImportError:
# Before setuptools v39.0
from pkg_resources.extern import packaging
version = packaging.version
except ImportError:
raise RuntimeError("The 'packaging' library is missing.")
aiohttp_version = version.parse(aiohttp.__version__)
class ReuseableTCPConnector(TCPConnector):
def __init__(self, *args, **kwargs):
super(ReuseableTCPConnector, self).__init__(*args, **kwargs)
self.old_proto = None
if aiohttp.__version__ >= '3.0':
if aiohttp_version >= version.parse('3.3.0'):
async def connect(self, req, traces, timeout):
new_conn = await super(ReuseableTCPConnector, self)\
.connect(req, traces, timeout)
if self.old_proto is not None:
if self.old_proto != new_conn._protocol:
raise RuntimeError(
"We got a new connection, wanted the same one!")
print(new_conn.__dict__)
self.old_proto = new_conn._protocol
return new_conn
elif aiohttp_version >= version.parse('3.0.0'):
async def connect(self, req, traces=None):
new_conn = await super(ReuseableTCPConnector, self)\
.connect(req, traces=traces)
@ -28,7 +53,6 @@ class ReuseableTCPConnector(TCPConnector):
self.old_proto = new_conn._protocol
return new_conn
else:
async def connect(self, req):
new_conn = await super(ReuseableTCPConnector, self)\
.connect(req)

View File

@ -23,7 +23,7 @@ def reset_logging():
reload(logging)
def test_log():
def test_log(app):
log_stream = StringIO()
for handler in logging.root.handlers[:]:
logging.root.removeHandler(handler)
@ -33,7 +33,6 @@ def test_log():
stream=log_stream
)
log = logging.getLogger()
app = Sanic('test_logging')
rand_string = str(uuid.uuid4())
@app.route('/')
@ -80,9 +79,8 @@ def test_logging_pass_customer_logconfig():
@pytest.mark.parametrize('debug', (True, False, ))
def test_log_connection_lost(debug, monkeypatch):
def test_log_connection_lost(app, debug, monkeypatch):
""" Should not log Connection lost exception on non debug """
app = Sanic('connection_lost')
stream = StringIO()
root = logging.getLogger('root')
root.addHandler(logging.StreamHandler(stream))

View File

@ -1,7 +1,5 @@
from json import loads as json_loads, dumps as json_dumps
from sanic import Sanic
from sanic.request import Request
from sanic.response import json, text, HTTPResponse
from sanic.response import text, HTTPResponse
from sanic.exceptions import NotFound
@ -9,9 +7,7 @@ from sanic.exceptions import NotFound
# GET
# ------------------------------------------------------------ #
def test_middleware_request():
app = Sanic('test_middleware_request')
def test_middleware_request(app):
results = []
@app.middleware
@ -28,9 +24,7 @@ def test_middleware_request():
assert type(results[0]) is Request
def test_middleware_response():
app = Sanic('test_middleware_response')
def test_middleware_response(app):
results = []
@app.middleware('request')
@ -54,8 +48,7 @@ def test_middleware_response():
assert isinstance(results[2], HTTPResponse)
def test_middleware_response_exception():
app = Sanic('test_middleware_response_exception')
def test_middleware_response_exception(app):
result = {'status_code': None}
@app.middleware('response')
@ -75,8 +68,7 @@ def test_middleware_response_exception():
assert response.text == 'OK'
assert result['status_code'] == 404
def test_middleware_override_request():
app = Sanic('test_middleware_override_request')
def test_middleware_override_request(app):
@app.middleware
async def halt_request(request):
@ -92,8 +84,7 @@ def test_middleware_override_request():
assert response.text == 'OK'
def test_middleware_override_response():
app = Sanic('test_middleware_override_response')
def test_middleware_override_response(app):
@app.middleware('response')
async def process_response(request, response):
@ -109,10 +100,7 @@ def test_middleware_override_response():
assert response.text == 'OK'
def test_middleware_order():
app = Sanic('test_middleware_order')
def test_middleware_order(app):
order = []
@app.middleware('request')

View File

@ -2,15 +2,13 @@ import multiprocessing
import random
import signal
from sanic import Sanic
from sanic.testing import HOST, PORT
def test_multiprocessing():
def test_multiprocessing(app):
"""Tests that the number of children we produce is correct"""
# Selects a number at random so we can spot check
num_workers = random.choice(range(2, multiprocessing.cpu_count() * 2 + 1))
app = Sanic('test_multiprocessing')
process_list = set()
def stop_on_alarm(*args):

View File

@ -4,7 +4,6 @@
import asyncio
import pytest
from sanic import Sanic
from sanic.blueprints import Blueprint
from sanic.response import text
from sanic.exceptions import URLBuildError
@ -16,8 +15,7 @@ from sanic.constants import HTTP_METHODS
# ------------------------------------------------------------ #
@pytest.mark.parametrize('method', HTTP_METHODS)
def test_versioned_named_routes_get(method):
app = Sanic('test_shorhand_routes_get')
def test_versioned_named_routes_get(app, method):
bp = Blueprint('test_bp', url_prefix='/bp')
method = method.lower()
@ -57,8 +55,7 @@ def test_versioned_named_routes_get(method):
app.url_for('handler')
def test_shorthand_default_routes_get():
app = Sanic('test_shorhand_routes_get')
def test_shorthand_default_routes_get(app):
@app.get('/get')
def handler(request):
@ -68,8 +65,7 @@ def test_shorthand_default_routes_get():
assert app.url_for('handler') == '/get'
def test_shorthand_named_routes_get():
app = Sanic('test_shorhand_routes_get')
def test_shorthand_named_routes_get(app):
bp = Blueprint('test_bp', url_prefix='/bp')
@app.get('/get', name='route_get')
@ -93,8 +89,7 @@ def test_shorthand_named_routes_get():
app.url_for('test_bp.handler2')
def test_shorthand_named_routes_post():
app = Sanic('test_shorhand_routes_post')
def test_shorthand_named_routes_post(app):
@app.post('/post', name='route_name')
def handler(request):
@ -106,8 +101,7 @@ def test_shorthand_named_routes_post():
app.url_for('handler')
def test_shorthand_named_routes_put():
app = Sanic('test_shorhand_routes_put')
def test_shorthand_named_routes_put(app):
@app.put('/put', name='route_put')
def handler(request):
@ -121,8 +115,7 @@ def test_shorthand_named_routes_put():
app.url_for('handler')
def test_shorthand_named_routes_delete():
app = Sanic('test_shorhand_routes_delete')
def test_shorthand_named_routes_delete(app):
@app.delete('/delete', name='route_delete')
def handler(request):
@ -136,8 +129,7 @@ def test_shorthand_named_routes_delete():
app.url_for('handler')
def test_shorthand_named_routes_patch():
app = Sanic('test_shorhand_routes_patch')
def test_shorthand_named_routes_patch(app):
@app.patch('/patch', name='route_patch')
def handler(request):
@ -151,8 +143,7 @@ def test_shorthand_named_routes_patch():
app.url_for('handler')
def test_shorthand_named_routes_head():
app = Sanic('test_shorhand_routes_head')
def test_shorthand_named_routes_head(app):
@app.head('/head', name='route_head')
def handler(request):
@ -166,8 +157,7 @@ def test_shorthand_named_routes_head():
app.url_for('handler')
def test_shorthand_named_routes_options():
app = Sanic('test_shorhand_routes_options')
def test_shorthand_named_routes_options(app):
@app.options('/options', name='route_options')
def handler(request):
@ -181,8 +171,7 @@ def test_shorthand_named_routes_options():
app.url_for('handler')
def test_named_static_routes():
app = Sanic('test_dynamic_route')
def test_named_static_routes(app):
@app.route('/test', name='route_test')
async def handler1(request):
@ -205,9 +194,7 @@ def test_named_static_routes():
app.url_for('handler2')
def test_named_dynamic_route():
app = Sanic('test_dynamic_route')
def test_named_dynamic_route(app):
results = []
@app.route('/folder/<name>', name='route_dynamic')
@ -221,8 +208,7 @@ def test_named_dynamic_route():
app.url_for('handler')
def test_dynamic_named_route_regex():
app = Sanic('test_dynamic_route_regex')
def test_dynamic_named_route_regex(app):
@app.route('/folder/<folder_id:[A-Za-z0-9]{0,4}>', name='route_re')
async def handler(request, folder_id):
@ -235,8 +221,7 @@ def test_dynamic_named_route_regex():
app.url_for('handler')
def test_dynamic_named_route_path():
app = Sanic('test_dynamic_route_path')
def test_dynamic_named_route_path(app):
@app.route('/<path:path>/info', name='route_dynamic_path')
async def handler(request, path):
@ -249,8 +234,7 @@ def test_dynamic_named_route_path():
app.url_for('handler')
def test_dynamic_named_route_unhashable():
app = Sanic('test_dynamic_route_unhashable')
def test_dynamic_named_route_unhashable(app):
@app.route('/folder/<unhashable:[A-Za-z0-9/]+>/end/',
name='route_unhashable')
@ -265,8 +249,7 @@ def test_dynamic_named_route_unhashable():
app.url_for('handler')
def test_websocket_named_route():
app = Sanic('test_websocket_route')
def test_websocket_named_route(app):
ev = asyncio.Event()
@app.websocket('/ws', name='route_ws')
@ -280,8 +263,7 @@ def test_websocket_named_route():
app.url_for('handler')
def test_websocket_named_route_with_subprotocols():
app = Sanic('test_websocket_route')
def test_websocket_named_route_with_subprotocols(app):
results = []
@app.websocket('/ws', subprotocols=['foo', 'bar'], name='route_ws')
@ -294,8 +276,7 @@ def test_websocket_named_route_with_subprotocols():
app.url_for('handler')
def test_static_add_named_route():
app = Sanic('test_static_add_route')
def test_static_add_named_route(app):
async def handler1(request):
return text('OK1')
@ -319,9 +300,7 @@ def test_static_add_named_route():
app.url_for('handler2')
def test_dynamic_add_named_route():
app = Sanic('test_dynamic_add_route')
def test_dynamic_add_named_route(app):
results = []
async def handler(request, name):
@ -335,8 +314,7 @@ def test_dynamic_add_named_route():
app.url_for('handler')
def test_dynamic_add_named_route_unhashable():
app = Sanic('test_dynamic_add_route_unhashable')
def test_dynamic_add_named_route_unhashable(app):
async def handler(request, unhashable):
return text('OK')
@ -351,8 +329,7 @@ def test_dynamic_add_named_route_unhashable():
app.url_for('handler')
def test_overload_routes():
app = Sanic('test_dynamic_route')
def test_overload_routes(app):
@app.route('/overload', methods=['GET'], name='route_first')
async def handler1(request):

View File

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

View File

@ -1,12 +1,10 @@
import pytest
from sanic import Sanic
from sanic.response import text, redirect
@pytest.fixture
def redirect_app():
app = Sanic('test_redirection')
def redirect_app(app):
@app.route('/redirect_init')
async def redirect_init(request):
@ -32,6 +30,10 @@ def redirect_app():
def handler(request):
return text('OK')
@app.route('/redirect_with_header_injection')
async def redirect_with_header_injection(request):
return redirect("/unsafe\ntest-header: test-value\n\ntest-body")
return app
@ -92,3 +94,16 @@ def test_chained_redirect(redirect_app):
assert response.url.endswith('/3')
except AttributeError:
assert response.url.path.endswith('/3')
def test_redirect_with_header_injection(redirect_app):
"""
Test redirection to a URL with header and body injections.
"""
request, response = redirect_app.test_client.get(
"/redirect_with_header_injection",
allow_redirects=False)
assert response.status == 302
assert "test-header" not in response.headers
assert not response.text.startswith('test-body')

View File

@ -1,6 +1,5 @@
import random
from sanic import Sanic
from sanic.response import json
try:
@ -9,8 +8,7 @@ except ImportError:
from json import loads
def test_storage():
app = Sanic('test_text')
def test_storage(app):
@app.middleware('request')
def store(request):
@ -29,8 +27,7 @@ def test_storage():
assert response_json.get('sidekick') is None
def test_app_injection():
app = Sanic('test_app_injection')
def test_app_injection(app):
expected = random.choice(range(0, 100))
@app.listener('after_server_start')

View File

@ -1,5 +1,4 @@
import asyncio
from sanic import Sanic
from sanic.blueprints import Blueprint
from sanic.views import CompositionView
from sanic.views import HTTPMethodView
@ -9,11 +8,9 @@ from sanic.response import stream, text
data = "abc" * 100000
def test_request_stream_method_view():
def test_request_stream_method_view(app):
'''for self.is_request_stream = True'''
app = Sanic('test_request_stream_method_view')
class SimpleView(HTTPMethodView):
def get(self, request):
@ -44,11 +41,9 @@ def test_request_stream_method_view():
assert response.text == data
def test_request_stream_app():
def test_request_stream_app(app):
'''for self.is_request_stream = True and decorators'''
app = Sanic('test_request_stream_app')
@app.get('/get')
async def get(request):
assert request.stream is None
@ -83,7 +78,7 @@ def test_request_stream_app():
body = await request.stream.get()
if body is None:
break
response.write(body.decode('utf-8'))
await response.write(body.decode('utf-8'))
return stream(streaming)
@app.put('/_put')
@ -100,7 +95,7 @@ def test_request_stream_app():
body = await request.stream.get()
if body is None:
break
response.write(body.decode('utf-8'))
await response.write(body.decode('utf-8'))
return stream(streaming)
@app.patch('/_patch')
@ -117,7 +112,7 @@ def test_request_stream_app():
body = await request.stream.get()
if body is None:
break
response.write(body.decode('utf-8'))
await response.write(body.decode('utf-8'))
return stream(streaming)
assert app.is_request_stream is True
@ -163,11 +158,9 @@ def test_request_stream_app():
assert response.text == data
def test_request_stream_handle_exception():
def test_request_stream_handle_exception(app):
'''for handling exceptions properly'''
app = Sanic('test_request_stream_exception')
@app.post('/post/<id>', stream=True)
async def post(request, id):
assert isinstance(request.stream, asyncio.Queue)
@ -177,7 +170,7 @@ def test_request_stream_handle_exception():
body = await request.stream.get()
if body is None:
break
response.write(body.decode('utf-8'))
await response.write(body.decode('utf-8'))
return stream(streaming)
# 404
@ -191,10 +184,8 @@ def test_request_stream_handle_exception():
assert response.text == 'Error: Method GET not allowed for URL /post/random_id'
def test_request_stream_blueprint():
def test_request_stream_blueprint(app):
'''for self.is_request_stream = True'''
app = Sanic('test_request_stream_blueprint')
bp = Blueprint('test_blueprint_request_stream_blueprint')
@app.get('/get')
@ -231,7 +222,7 @@ def test_request_stream_blueprint():
body = await request.stream.get()
if body is None:
break
response.write(body.decode('utf-8'))
await response.write(body.decode('utf-8'))
return stream(streaming)
@bp.put('/_put')
@ -248,7 +239,7 @@ def test_request_stream_blueprint():
body = await request.stream.get()
if body is None:
break
response.write(body.decode('utf-8'))
await response.write(body.decode('utf-8'))
return stream(streaming)
@bp.patch('/_patch')
@ -265,7 +256,7 @@ def test_request_stream_blueprint():
body = await request.stream.get()
if body is None:
break
response.write(body.decode('utf-8'))
await response.write(body.decode('utf-8'))
return stream(streaming)
app.blueprint(bp)
@ -313,11 +304,9 @@ def test_request_stream_blueprint():
assert response.text == data
def test_request_stream_composition_view():
def test_request_stream_composition_view(app):
'''for self.is_request_stream = True'''
app = Sanic('test_request_stream__composition_view')
def get_handler(request):
assert request.stream is None
return text('OK')
@ -348,11 +337,9 @@ def test_request_stream_composition_view():
assert response.text == data
def test_request_stream():
def test_request_stream(app):
'''test for complex application'''
bp = Blueprint('test_blueprint_request_stream')
app = Sanic('test_request_stream')
class SimpleView(HTTPMethodView):
@ -380,7 +367,7 @@ def test_request_stream():
body = await request.stream.get()
if body is None:
break
response.write(body.decode('utf-8'))
await response.write(body.decode('utf-8'))
return stream(streaming)
@app.get('/get')

View File

@ -5,9 +5,24 @@ import asyncio
from sanic.response import text
from sanic.config import Config
import aiohttp
from aiohttp import TCPConnector
from aiohttp import TCPConnector, ClientResponse
from sanic.testing import SanicTestClient, HOST, PORT
try:
try:
import packaging # direct use
except ImportError:
# setuptools v39.0 and above.
try:
from setuptools.extern import packaging
except ImportError:
# Before setuptools v39.0
from pkg_resources.extern import packaging
version = packaging.version
except ImportError:
raise RuntimeError("The 'packaging' library is missing.")
aiohttp_version = version.parse(aiohttp.__version__)
class DelayableTCPConnector(TCPConnector):
@ -38,6 +53,9 @@ class DelayableTCPConnector(TCPConnector):
self.orig_start = getattr(resp, 'start')
try:
if aiohttp_version >= version.parse("3.3.0"):
ret = await self.orig_start(connection)
else:
ret = await self.orig_start(connection,
read_until_eof)
except Exception as e:
@ -58,14 +76,30 @@ class DelayableTCPConnector(TCPConnector):
t = req.loop.time()
print("sending at {}".format(t), flush=True)
conn = next(iter(args)) # first arg is connection
if aiohttp.__version__ >= "3.1.0":
if aiohttp_version >= version.parse("3.1.0"):
try:
delayed_resp = await self.orig_send(*args, **kwargs)
except Exception as e:
if aiohttp_version >= version.parse("3.3.0"):
return aiohttp.ClientResponse(req.method, req.url,
writer=None, continue100=None, timer=None,
request_info=None, auto_decompress=None, traces=[],
loop=req.loop, session=None)
writer=None,
continue100=None,
timer=None,
request_info=None,
traces=[],
loop=req.loop,
session=None)
else:
return aiohttp.ClientResponse(req.method, req.url,
writer=None,
continue100=None,
timer=None,
request_info=None,
auto_decompress=None,
traces=[],
loop=req.loop,
session=None)
else:
try:
delayed_resp = self.orig_send(*args, **kwargs)
@ -73,7 +107,7 @@ class DelayableTCPConnector(TCPConnector):
return aiohttp.ClientResponse(req.method, req.url)
return delayed_resp
if aiohttp.__version__ >= "3.1.0":
if aiohttp_version >= version.parse("3.1.0"):
# aiohttp changed the request.send method to async
async def send(self, *args, **kwargs):
gen = self.delayed_send(*args, **kwargs)
@ -96,12 +130,25 @@ class DelayableTCPConnector(TCPConnector):
self._post_connect_delay = _post_connect_delay
self._pre_request_delay = _pre_request_delay
if aiohttp.__version__ >= '3.0':
if aiohttp_version >= version.parse("3.3.0"):
async def connect(self, req, traces, timeout):
d_req = DelayableTCPConnector.\
RequestContextManager(req, self._pre_request_delay)
conn = await super(DelayableTCPConnector, self).\
connect(req, traces, timeout)
if self._post_connect_delay and self._post_connect_delay > 0:
await asyncio.sleep(self._post_connect_delay,
loop=self._loop)
req.send = d_req.send
t = req.loop.time()
print("Connected at {}".format(t), flush=True)
return conn
elif aiohttp_version >= version.parse("3.0.0"):
async def connect(self, req, traces=None):
d_req = DelayableTCPConnector.\
RequestContextManager(req, self._pre_request_delay)
conn = await super(DelayableTCPConnector, self).connect(req, traces=traces)
conn = await super(DelayableTCPConnector, self).\
connect(req, traces=traces)
if self._post_connect_delay and self._post_connect_delay > 0:
await asyncio.sleep(self._post_connect_delay,
loop=self._loop)

View File

@ -5,7 +5,6 @@ import ssl
import pytest
from sanic import Sanic
from sanic.exceptions import ServerError
from sanic.response import json, text
from sanic.request import DEFAULT_HTTP_CONTENT_TYPE
@ -16,8 +15,7 @@ from sanic.testing import HOST, PORT
# GET
# ------------------------------------------------------------ #
def test_sync():
app = Sanic('test_text')
def test_sync(app):
@app.route('/')
def handler(request):
@ -27,8 +25,8 @@ def test_sync():
assert response.text == 'Hello'
def test_remote_address():
app = Sanic('test_text')
def test_remote_address(app):
@app.route('/')
def handler(request):
@ -38,8 +36,8 @@ def test_remote_address():
assert response.text == '127.0.0.1'
def test_text():
app = Sanic('test_text')
def test_text(app):
@app.route('/')
async def handler(request):
@ -50,8 +48,7 @@ def test_text():
assert response.text == 'Hello'
def test_headers():
app = Sanic('test_text')
def test_headers(app):
@app.route('/')
async def handler(request):
@ -63,8 +60,7 @@ def test_headers():
assert response.headers.get('spam') == 'great'
def test_non_str_headers():
app = Sanic('test_text')
def test_non_str_headers(app):
@app.route('/')
async def handler(request):
@ -75,8 +71,8 @@ def test_non_str_headers():
assert response.headers.get('answer') == '42'
def test_invalid_response():
app = Sanic('test_invalid_response')
def test_invalid_response(app):
@app.exception(ServerError)
def handler_exception(request, exception):
@ -91,8 +87,7 @@ def test_invalid_response():
assert response.text == "Internal Server Error."
def test_json():
app = Sanic('test_json')
def test_json(app):
@app.route('/')
async def handler(request):
@ -105,8 +100,7 @@ def test_json():
assert results.get('test') == True
def test_empty_json():
app = Sanic('test_json')
def test_empty_json(app):
@app.route('/')
async def handler(request):
@ -118,8 +112,7 @@ def test_empty_json():
assert response.text == 'null'
def test_invalid_json():
app = Sanic('test_json')
def test_invalid_json(app):
@app.route('/')
async def handler(request):
@ -131,8 +124,7 @@ def test_invalid_json():
assert response.status == 400
def test_query_string():
app = Sanic('test_query_string')
def test_query_string(app):
@app.route('/')
async def handler(request):
@ -145,8 +137,7 @@ def test_query_string():
assert request.args.get('test2') == 'false'
def test_uri_template():
app = Sanic('test_uri_template')
def test_uri_template(app):
@app.route('/foo/<id:int>/bar/<name:[A-z]+>')
async def handler(request):
@ -156,8 +147,7 @@ def test_uri_template():
assert request.uri_template == '/foo/<id:int>/bar/<name:[A-z]+>'
def test_token():
app = Sanic('test_post_token')
def test_token(app):
@app.route('/')
async def handler(request):
@ -204,8 +194,7 @@ def test_token():
assert request.token is None
def test_content_type():
app = Sanic('test_content_type')
def test_content_type(app):
@app.route('/')
async def handler(request):
@ -223,8 +212,7 @@ def test_content_type():
assert response.text == 'application/json'
def test_remote_addr():
app = Sanic('test_content_type')
def test_remote_addr(app):
@app.route('/')
async def handler(request):
@ -249,8 +237,7 @@ def test_remote_addr():
assert response.text == '127.0.0.1'
def test_match_info():
app = Sanic('test_match_info')
def test_match_info(app):
@app.route('/api/v1/user/<user_id>/')
async def handler(request, user_id):
@ -266,8 +253,7 @@ def test_match_info():
# POST
# ------------------------------------------------------------ #
def test_post_json():
app = Sanic('test_post_json')
def test_post_json(app):
@app.route('/', methods=['POST'])
async def handler(request):
@ -283,8 +269,7 @@ def test_post_json():
assert response.text == 'OK'
def test_post_form_urlencoded():
app = Sanic('test_post_form_urlencoded')
def test_post_form_urlencoded(app):
@app.route('/', methods=['POST'])
async def handler(request):
@ -311,8 +296,7 @@ def test_post_form_urlencoded():
'OK\r\n' \
'------sanic--\r\n',
])
def test_post_form_multipart_form_data(payload):
app = Sanic('test_post_form_multipart_form_data')
def test_post_form_multipart_form_data(app, payload):
@app.route('/', methods=['POST'])
async def handler(request):
@ -331,8 +315,7 @@ def test_post_form_multipart_form_data(payload):
('/bar/baz', '', 'http://{}:{}/bar/baz'),
('/moo/boo', 'arg1=val1', 'http://{}:{}/moo/boo?arg1=val1')
])
def test_url_attributes_no_ssl(path, query, expected_url):
app = Sanic('test_url_attrs_no_ssl')
def test_url_attributes_no_ssl(app, path, query, expected_url):
async def handler(request):
return text('OK')
@ -356,9 +339,7 @@ def test_url_attributes_no_ssl(path, query, expected_url):
('/bar/baz', '', 'https://{}:{}/bar/baz'),
('/moo/boo', 'arg1=val1', 'https://{}:{}/moo/boo?arg1=val1')
])
def test_url_attributes_with_ssl(path, query, expected_url):
app = Sanic('test_url_attrs_with_ssl')
def test_url_attributes_with_ssl(app, path, query, expected_url):
current_dir = os.path.dirname(os.path.realpath(__file__))
context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)
context.load_cert_chain(

View File

@ -8,17 +8,16 @@ from urllib.parse import unquote
import pytest
from random import choice
from sanic import Sanic
from sanic.response import HTTPResponse, stream, StreamingHTTPResponse, file, file_stream, json
from sanic.server import HttpProtocol
from sanic.testing import HOST, PORT
from unittest.mock import MagicMock
JSON_DATA = {'ok': True}
def test_response_body_not_a_string():
def test_response_body_not_a_string(app):
"""Test when a response body sent from the application is not a string"""
app = Sanic('response_body_not_a_string')
random_num = choice(range(1000))
@app.route('/hello')
@ -30,13 +29,12 @@ def test_response_body_not_a_string():
async def sample_streaming_fn(response):
response.write('foo,')
await response.write('foo,')
await asyncio.sleep(.001)
response.write('bar')
await response.write('bar')
def test_method_not_allowed():
app = Sanic('method_not_allowed')
def test_method_not_allowed(app):
@app.get('/')
async def test(request):
@ -64,9 +62,27 @@ def test_method_not_allowed():
assert response.headers['Content-Length'] == '0'
def test_response_header(app):
@app.get('/')
async def test(request):
return json({
"ok": True
}, headers={
'CONTENT-TYPE': 'application/json'
})
request, response = app.test_client.get('/')
assert dict(response.headers) == {
'Connection': 'keep-alive',
'Keep-Alive': '2',
'Content-Length': '11',
'Content-Type': 'application/json',
}
@pytest.fixture
def json_app():
app = Sanic('json')
def json_app(app):
@app.route("/")
async def test(request):
@ -124,8 +140,7 @@ def test_no_content(json_app):
@pytest.fixture
def streaming_app():
app = Sanic('streaming')
def streaming_app(app):
@app.route("/")
async def test(request):
@ -170,20 +185,30 @@ def test_stream_response_includes_chunked_header():
def test_stream_response_writes_correct_content_to_transport(streaming_app):
response = StreamingHTTPResponse(sample_streaming_fn)
response.transport = MagicMock(asyncio.Transport)
response.protocol = MagicMock(HttpProtocol)
response.protocol.transport = MagicMock(asyncio.Transport)
async def mock_drain():
pass
def mock_push_data(data):
response.protocol.transport.write(data)
response.protocol.push_data = mock_push_data
response.protocol.drain = mock_drain
@streaming_app.listener('after_server_start')
async def run_stream(app, loop):
await response.stream()
assert response.transport.write.call_args_list[1][0][0] == (
assert response.protocol.transport.write.call_args_list[1][0][0] == (
b'4\r\nfoo,\r\n'
)
assert response.transport.write.call_args_list[2][0][0] == (
assert response.protocol.transport.write.call_args_list[2][0][0] == (
b'3\r\nbar\r\n'
)
assert response.transport.write.call_args_list[3][0][0] == (
assert response.protocol.transport.write.call_args_list[3][0][0] == (
b'0\r\n\r\n'
)
@ -208,25 +233,25 @@ def get_file_content(static_file_directory, file_name):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt', 'python.png'])
def test_file_response(file_name, static_file_directory):
app = Sanic('test_file_helper')
@pytest.mark.parametrize('status', [200, 401])
def test_file_response(app, file_name, static_file_directory, status):
@app.route('/files/<filename>', methods=['GET'])
def file_route(request, filename):
file_path = os.path.join(static_file_directory, filename)
file_path = os.path.abspath(unquote(file_path))
return file(file_path, mime_type=guess_type(file_path)[0] or 'text/plain')
return file(file_path, status=status,
mime_type=guess_type(file_path)[0] or 'text/plain')
request, response = app.test_client.get('/files/{}'.format(file_name))
assert response.status == 200
assert response.status == status
assert response.body == get_file_content(static_file_directory, file_name)
assert 'Content-Disposition' not in response.headers
@pytest.mark.parametrize('source,dest', [
('test.file', 'my_file.txt'), ('decode me.txt', 'readme.md'), ('python.png', 'logo.png')])
def test_file_response_custom_filename(source, dest, static_file_directory):
app = Sanic('test_file_helper')
def test_file_response_custom_filename(app, source, dest, static_file_directory):
@app.route('/files/<filename>', methods=['GET'])
def file_route(request, filename):
@ -241,8 +266,7 @@ def test_file_response_custom_filename(source, dest, static_file_directory):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_file_head_response(file_name, static_file_directory):
app = Sanic('test_file_helper')
def test_file_head_response(app, file_name, static_file_directory):
@app.route('/files/<filename>', methods=['GET', 'HEAD'])
async def file_route(request, filename):
@ -270,8 +294,7 @@ def test_file_head_response(file_name, static_file_directory):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt', 'python.png'])
def test_file_stream_response(file_name, static_file_directory):
app = Sanic('test_file_helper')
def test_file_stream_response(app, file_name, static_file_directory):
@app.route('/files/<filename>', methods=['GET'])
def file_route(request, filename):
@ -288,8 +311,7 @@ def test_file_stream_response(file_name, static_file_directory):
@pytest.mark.parametrize('source,dest', [
('test.file', 'my_file.txt'), ('decode me.txt', 'readme.md'), ('python.png', 'logo.png')])
def test_file_stream_response_custom_filename(source, dest, static_file_directory):
app = Sanic('test_file_helper')
def test_file_stream_response_custom_filename(app, source, dest, static_file_directory):
@app.route('/files/<filename>', methods=['GET'])
def file_route(request, filename):
@ -304,8 +326,7 @@ def test_file_stream_response_custom_filename(source, dest, static_file_director
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_file_stream_head_response(file_name, static_file_directory):
app = Sanic('test_file_helper')
def test_file_stream_head_response(app, file_name, static_file_directory):
@app.route('/files/<filename>', methods=['GET', 'HEAD'])
async def file_route(request, filename):

View File

@ -7,6 +7,7 @@ from sanic.config import Config
Config.RESPONSE_TIMEOUT = 1
response_timeout_app = Sanic('test_response_timeout')
response_timeout_default_app = Sanic('test_response_timeout_default')
response_handler_cancelled_app = Sanic('test_response_handler_cancelled')
@response_timeout_app.route('/1')
@ -36,3 +37,29 @@ def test_default_server_error_response_timeout():
request, response = response_timeout_default_app.test_client.get('/1')
assert response.status == 503
assert response.text == 'Error: Response Timeout'
response_handler_cancelled_app.flag = False
@response_handler_cancelled_app.exception(asyncio.CancelledError)
def handler_cancelled(request, exception):
# If we get a CancelledError, it means sanic has already sent a response,
# we should not ever have to handle a CancelledError.
response_handler_cancelled_app.flag = True
return text("App received CancelledError!", 500)
# The client will never receive this response, because the socket
# is already closed when we get a CancelledError.
@response_handler_cancelled_app.route('/1')
async def handler_3(request):
await asyncio.sleep(2)
return text('OK')
def test_response_handler_cancelled():
request, response = response_handler_cancelled_app.test_client.get('/1')
assert response.status == 503
assert response.text == 'Error: Response Timeout'
assert response_handler_cancelled_app.flag is False

View File

@ -12,9 +12,7 @@ from sanic.constants import HTTP_METHODS
# ------------------------------------------------------------ #
@pytest.mark.parametrize('method', HTTP_METHODS)
def test_versioned_routes_get(method):
app = Sanic('test_shorhand_routes_get')
def test_versioned_routes_get(app, method):
method = method.lower()
func = getattr(app, method)
@ -32,8 +30,7 @@ def test_versioned_routes_get(method):
assert response.status == 200
def test_shorthand_routes_get():
app = Sanic('test_shorhand_routes_get')
def test_shorthand_routes_get(app):
@app.get('/get')
def handler(request):
@ -46,8 +43,7 @@ def test_shorthand_routes_get():
assert response.status == 405
def test_shorthand_routes_multiple():
app = Sanic('test_shorthand_routes_multiple')
def test_shorthand_routes_multiple(app):
@app.get('/get')
def get_handler(request):
@ -65,8 +61,7 @@ def test_shorthand_routes_multiple():
assert response.status == 200
def test_route_strict_slash():
app = Sanic('test_route_strict_slash')
def test_route_strict_slash(app):
@app.get('/get', strict_slashes=True)
def handler(request):
@ -93,9 +88,8 @@ def test_route_strict_slash():
assert response.status == 404
def test_route_invalid_parameter_syntax():
def test_route_invalid_parameter_syntax(app):
with pytest.raises(ValueError):
app = Sanic('test_route_invalid_param_syntax')
@app.get('/get/<:string>', strict_slashes=True)
def handler(request):
@ -115,8 +109,7 @@ def test_route_strict_slash_default_value():
assert response.status == 404
def test_route_strict_slash_without_passing_default_value():
app = Sanic('test_route_strict_slash')
def test_route_strict_slash_without_passing_default_value(app):
@app.get('/get')
def handler(request):
@ -137,8 +130,7 @@ def test_route_strict_slash_default_value_can_be_overwritten():
assert response.text == 'OK'
def test_route_slashes_overload():
app = Sanic('test_route_slashes_overload')
def test_route_slashes_overload(app):
@app.get('/hello/')
def handler(request):
@ -161,8 +153,7 @@ def test_route_slashes_overload():
assert response.text == 'OK'
def test_route_optional_slash():
app = Sanic('test_route_optional_slash')
def test_route_optional_slash(app):
@app.get('/get')
def handler(request):
@ -174,9 +165,8 @@ def test_route_optional_slash():
request, response = app.test_client.get('/get/')
assert response.text == 'OK'
def test_route_strict_slashes_set_to_false_and_host_is_a_list():
def test_route_strict_slashes_set_to_false_and_host_is_a_list(app):
#Part of regression test for issue #1120
app = Sanic('test_route_strict_slashes_set_to_false_and_host_is_a_list')
site1 = 'localhost:{}'.format(app.test_client.port)
@ -209,8 +199,7 @@ def test_route_strict_slashes_set_to_false_and_host_is_a_list():
request, response = app.test_client.delete('http://' + site1 +'/delete')
assert response.text == 'OK'
def test_shorthand_routes_post():
app = Sanic('test_shorhand_routes_post')
def test_shorthand_routes_post(app):
@app.post('/post')
def handler(request):
@ -223,8 +212,7 @@ def test_shorthand_routes_post():
assert response.status == 405
def test_shorthand_routes_put():
app = Sanic('test_shorhand_routes_put')
def test_shorthand_routes_put(app):
@app.put('/put')
def handler(request):
@ -240,8 +228,7 @@ def test_shorthand_routes_put():
assert response.status == 405
def test_shorthand_routes_delete():
app = Sanic('test_shorhand_routes_delete')
def test_shorthand_routes_delete(app):
@app.delete('/delete')
def handler(request):
@ -257,8 +244,7 @@ def test_shorthand_routes_delete():
assert response.status == 405
def test_shorthand_routes_patch():
app = Sanic('test_shorhand_routes_patch')
def test_shorthand_routes_patch(app):
@app.patch('/patch')
def handler(request):
@ -274,8 +260,7 @@ def test_shorthand_routes_patch():
assert response.status == 405
def test_shorthand_routes_head():
app = Sanic('test_shorhand_routes_head')
def test_shorthand_routes_head(app):
@app.head('/head')
def handler(request):
@ -291,8 +276,7 @@ def test_shorthand_routes_head():
assert response.status == 405
def test_shorthand_routes_options():
app = Sanic('test_shorhand_routes_options')
def test_shorthand_routes_options(app):
@app.options('/options')
def handler(request):
@ -308,8 +292,7 @@ def test_shorthand_routes_options():
assert response.status == 405
def test_static_routes():
app = Sanic('test_dynamic_route')
def test_static_routes(app):
@app.route('/test')
async def handler1(request):
@ -326,9 +309,7 @@ def test_static_routes():
assert response.text == 'OK2'
def test_dynamic_route():
app = Sanic('test_dynamic_route')
def test_dynamic_route(app):
results = []
@app.route('/folder/<name>')
@ -342,9 +323,7 @@ def test_dynamic_route():
assert results[0] == 'test123'
def test_dynamic_route_string():
app = Sanic('test_dynamic_route_string')
def test_dynamic_route_string(app):
results = []
@app.route('/folder/<name:string>')
@ -363,9 +342,7 @@ def test_dynamic_route_string():
assert results[1] == 'favicon.ico'
def test_dynamic_route_int():
app = Sanic('test_dynamic_route_int')
def test_dynamic_route_int(app):
results = []
@app.route('/folder/<folder_id:int>')
@ -381,9 +358,7 @@ def test_dynamic_route_int():
assert response.status == 404
def test_dynamic_route_number():
app = Sanic('test_dynamic_route_number')
def test_dynamic_route_number(app):
results = []
@app.route('/weight/<weight:number>')
@ -402,8 +377,7 @@ def test_dynamic_route_number():
assert response.status == 404
def test_dynamic_route_regex():
app = Sanic('test_dynamic_route_regex')
def test_dynamic_route_regex(app):
@app.route('/folder/<folder_id:[A-Za-z0-9]{0,4}>')
async def handler(request, folder_id):
@ -422,9 +396,8 @@ def test_dynamic_route_regex():
assert response.status == 200
def test_dynamic_route_uuid():
def test_dynamic_route_uuid(app):
import uuid
app = Sanic('test_dynamic_route_uuid')
results = []
@ -444,8 +417,7 @@ def test_dynamic_route_uuid():
assert response.status == 404
def test_dynamic_route_path():
app = Sanic('test_dynamic_route_path')
def test_dynamic_route_path(app):
@app.route('/<path:path>/info')
async def handler(request, path):
@ -468,8 +440,7 @@ def test_dynamic_route_path():
assert response.status == 200
def test_dynamic_route_unhashable():
app = Sanic('test_dynamic_route_unhashable')
def test_dynamic_route_unhashable(app):
@app.route('/folder/<unhashable:[A-Za-z0-9/]+>/end/')
async def handler(request, unhashable):
@ -488,8 +459,7 @@ def test_dynamic_route_unhashable():
assert response.status == 404
def test_websocket_route():
app = Sanic('test_websocket_route')
def test_websocket_route(app):
ev = asyncio.Event()
@app.websocket('/ws')
@ -506,8 +476,7 @@ def test_websocket_route():
assert ev.is_set()
def test_websocket_route_with_subprotocols():
app = Sanic('test_websocket_route')
def test_websocket_route_with_subprotocols(app):
results = []
@app.websocket('/ws', subprotocols=['foo', 'bar'])
@ -548,8 +517,7 @@ def test_websocket_route_with_subprotocols():
assert results == ['bar', 'bar', None, None]
def test_route_duplicate():
app = Sanic('test_route_duplicate')
def test_route_duplicate(app):
with pytest.raises(RouteExists):
@app.route('/test')
@ -570,8 +538,7 @@ def test_route_duplicate():
pass
def test_method_not_allowed():
app = Sanic('test_method_not_allowed')
def test_method_not_allowed(app):
@app.route('/test', methods=['GET'])
async def handler(request):
@ -584,8 +551,7 @@ def test_method_not_allowed():
assert response.status == 405
def test_static_add_route():
app = Sanic('test_static_add_route')
def test_static_add_route(app):
async def handler1(request):
return text('OK1')
@ -603,8 +569,7 @@ def test_static_add_route():
assert response.text == 'OK2'
def test_dynamic_add_route():
app = Sanic('test_dynamic_add_route')
def test_dynamic_add_route(app):
results = []
@ -619,8 +584,7 @@ def test_dynamic_add_route():
assert results[0] == 'test123'
def test_dynamic_add_route_string():
app = Sanic('test_dynamic_add_route_string')
def test_dynamic_add_route_string(app):
results = []
@ -640,9 +604,7 @@ def test_dynamic_add_route_string():
assert results[1] == 'favicon.ico'
def test_dynamic_add_route_int():
app = Sanic('test_dynamic_add_route_int')
def test_dynamic_add_route_int(app):
results = []
async def handler(request, folder_id):
@ -659,9 +621,7 @@ def test_dynamic_add_route_int():
assert response.status == 404
def test_dynamic_add_route_number():
app = Sanic('test_dynamic_add_route_number')
def test_dynamic_add_route_number(app):
results = []
async def handler(request, weight):
@ -681,8 +641,7 @@ def test_dynamic_add_route_number():
assert response.status == 404
def test_dynamic_add_route_regex():
app = Sanic('test_dynamic_route_int')
def test_dynamic_add_route_regex(app):
async def handler(request, folder_id):
return text('OK')
@ -702,8 +661,7 @@ def test_dynamic_add_route_regex():
assert response.status == 200
def test_dynamic_add_route_unhashable():
app = Sanic('test_dynamic_add_route_unhashable')
def test_dynamic_add_route_unhashable(app):
async def handler(request, unhashable):
return text('OK')
@ -723,8 +681,7 @@ def test_dynamic_add_route_unhashable():
assert response.status == 404
def test_add_route_duplicate():
app = Sanic('test_add_route_duplicate')
def test_add_route_duplicate(app):
with pytest.raises(RouteExists):
async def handler1(request):
@ -747,8 +704,7 @@ def test_add_route_duplicate():
app.add_route(handler2, '/test/<dynamic>/')
def test_add_route_method_not_allowed():
app = Sanic('test_add_route_method_not_allowed')
def test_add_route_method_not_allowed(app):
async def handler(request):
return text('OK')
@ -762,8 +718,7 @@ def test_add_route_method_not_allowed():
assert response.status == 405
def test_remove_static_route():
app = Sanic('test_remove_static_route')
def test_remove_static_route(app):
async def handler1(request):
return text('OK1')
@ -790,8 +745,7 @@ def test_remove_static_route():
assert response.status == 404
def test_remove_dynamic_route():
app = Sanic('test_remove_dynamic_route')
def test_remove_dynamic_route(app):
async def handler(request, name):
return text('OK')
@ -806,15 +760,13 @@ def test_remove_dynamic_route():
assert response.status == 404
def test_remove_inexistent_route():
app = Sanic('test_remove_inexistent_route')
def test_remove_inexistent_route(app):
with pytest.raises(RouteDoesNotExist):
app.remove_route('/test')
def test_removing_slash():
app = Sanic(__name__)
def test_removing_slash(app):
@app.get('/rest/<resource>')
def get(_):
@ -827,8 +779,7 @@ def test_removing_slash():
assert len(app.router.routes_all.keys()) == 2
def test_remove_unhashable_route():
app = Sanic('test_remove_unhashable_route')
def test_remove_unhashable_route(app):
async def handler(request, unhashable):
return text('OK')
@ -856,8 +807,7 @@ def test_remove_unhashable_route():
assert response.status == 404
def test_remove_route_without_clean_cache():
app = Sanic('test_remove_static_route')
def test_remove_route_without_clean_cache(app):
async def handler(request):
return text('OK')
@ -884,8 +834,7 @@ def test_remove_route_without_clean_cache():
assert response.status == 200
def test_overload_routes():
app = Sanic('test_dynamic_route')
def test_overload_routes(app):
@app.route('/overload', methods=['GET'])
async def handler1(request):
@ -913,8 +862,7 @@ def test_overload_routes():
return text('Duplicated')
def test_unmergeable_overload_routes():
app = Sanic('test_dynamic_route')
def test_unmergeable_overload_routes(app):
@app.route('/overload_whole', methods=None)
async def handler1(request):
@ -947,8 +895,7 @@ def test_unmergeable_overload_routes():
assert response.status == 405
def test_unicode_routes():
app = Sanic('test_unicode_routes')
def test_unicode_routes(app):
@app.get('/你好')
def handler1(request):
@ -965,8 +912,7 @@ def test_unicode_routes():
assert response.text == 'OK2 你好'
def test_uri_with_different_method_and_different_params():
app = Sanic('test_uri')
def test_uri_with_different_method_and_different_params(app):
@app.route('/ads/<ad_id>', methods=['GET'])
async def ad_get(request, ad_id):

View File

@ -1,11 +1,7 @@
from io import StringIO
from random import choice
from string import ascii_letters
import signal
import pytest
from sanic import Sanic
from sanic.testing import HOST, PORT
AVAILABLE_LISTENERS = [
@ -37,54 +33,46 @@ def start_stop_app(random_name_app, **run_kwargs):
@pytest.mark.parametrize('listener_name', AVAILABLE_LISTENERS)
def test_single_listener(listener_name):
def test_single_listener(app, 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()
output = []
# Register listener
random_name_app.listener(listener_name)(
app.listener(listener_name)(
create_listener(listener_name, output))
start_stop_app(random_name_app)
assert random_name_app.name + listener_name == output.pop()
start_stop_app(app)
assert app.name + listener_name == output.pop()
@pytest.mark.parametrize('listener_name', AVAILABLE_LISTENERS)
def test_register_listener(listener_name):
def test_register_listener(app, listener_name):
"""
Test that listeners on their own work with
app.register_listener method
"""
random_name_app = Sanic(''.join(
[choice(ascii_letters) for _ in range(choice(range(5, 10)))]))
output = list()
output = []
# Register listener
listener = create_listener(listener_name, output)
random_name_app.register_listener(listener,
app.register_listener(listener,
event=listener_name)
start_stop_app(random_name_app)
assert random_name_app.name + listener_name == output.pop()
start_stop_app(app)
assert 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()
def test_all_listeners(app):
output = []
for listener_name in AVAILABLE_LISTENERS:
listener = create_listener(listener_name, output)
random_name_app.listener(listener_name)(listener)
start_stop_app(random_name_app)
app.listener(listener_name)(listener)
start_stop_app(app)
for listener_name in AVAILABLE_LISTENERS:
assert random_name_app.name + listener_name == output.pop()
assert app.name + listener_name == output.pop()
async def test_trigger_before_events_create_server():
async def test_trigger_before_events_create_server(app):
class MySanicDb:
pass
app = Sanic("test_sanic_app")
@app.listener('before_server_start')
async def init_db(app, loop):
app.db = MySanicDb()

View File

@ -1,4 +1,3 @@
from sanic import Sanic
from sanic.response import HTTPResponse
from sanic.testing import HOST, PORT
from unittest.mock import MagicMock
@ -18,9 +17,8 @@ def set_loop(app, loop):
def after(app, loop):
calledq.put(loop.add_signal_handler.called)
def test_register_system_signals():
def test_register_system_signals(app):
"""Test if sanic register system signals"""
app = Sanic('test_register_system_signals')
@app.route('/hello')
async def hello_route(request):
@ -34,9 +32,8 @@ def test_register_system_signals():
assert calledq.get() == True
def test_dont_register_system_signals():
def test_dont_register_system_signals(app):
"""Test if sanic don't register system signals"""
app = Sanic('test_register_system_signals')
@app.route('/hello')
async def hello_route(request):

View File

@ -3,8 +3,6 @@ import os
import pytest
from sanic import Sanic
@pytest.fixture(scope='module')
def static_file_directory():
@ -26,8 +24,7 @@ def get_file_content(static_file_directory, file_name):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt', 'python.png'])
def test_static_file(static_file_directory, file_name):
app = Sanic('test_static')
def test_static_file(app, static_file_directory, file_name):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name))
@ -36,11 +33,23 @@ def test_static_file(static_file_directory, file_name):
assert response.body == get_file_content(static_file_directory, file_name)
@pytest.mark.parametrize('file_name', ['test.html'])
def test_static_file_content_type(app, static_file_directory, file_name):
app.static(
'/testing.file',
get_file_path(static_file_directory, file_name),
content_type='text/html; charset=utf-8'
)
request, response = app.test_client.get('/testing.file')
assert response.status == 200
assert response.body == get_file_content(static_file_directory, file_name)
assert response.headers['Content-Type'] == 'text/html; charset=utf-8'
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
@pytest.mark.parametrize('base_uri', ['/static', '', '/dir'])
def test_static_directory(file_name, base_uri, static_file_directory):
app = Sanic('test_static')
def test_static_directory(app, file_name, base_uri, static_file_directory):
app.static(base_uri, static_file_directory)
request, response = app.test_client.get(
@ -50,8 +59,7 @@ def test_static_directory(file_name, base_uri, static_file_directory):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_head_request(file_name, static_file_directory):
app = Sanic('test_static')
def test_static_head_request(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)
@ -66,8 +74,7 @@ def test_static_head_request(file_name, static_file_directory):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_content_range_correct(file_name, static_file_directory):
app = Sanic('test_static')
def test_static_content_range_correct(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)
@ -87,8 +94,7 @@ def test_static_content_range_correct(file_name, static_file_directory):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_content_range_front(file_name, static_file_directory):
app = Sanic('test_static')
def test_static_content_range_front(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)
@ -108,8 +114,7 @@ def test_static_content_range_front(file_name, static_file_directory):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_content_range_back(file_name, static_file_directory):
app = Sanic('test_static')
def test_static_content_range_back(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)
@ -129,8 +134,7 @@ def test_static_content_range_back(file_name, static_file_directory):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_content_range_empty(file_name, static_file_directory):
app = Sanic('test_static')
def test_static_content_range_empty(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)
@ -146,8 +150,7 @@ def test_static_content_range_empty(file_name, static_file_directory):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_content_range_error(file_name, static_file_directory):
app = Sanic('test_static')
def test_static_content_range_error(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)
@ -164,8 +167,7 @@ def test_static_content_range_error(file_name, static_file_directory):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt', 'python.png'])
def test_static_file_specified_host(static_file_directory, file_name):
app = Sanic('test_static')
def test_static_file_specified_host(app, static_file_directory, file_name):
app.static(
'/testing.file',
get_file_path(static_file_directory, file_name),

View File

@ -1,7 +1,6 @@
import pytest as pytest
from urllib.parse import urlsplit, parse_qsl
from sanic import Sanic
from sanic.response import text
from sanic.views import HTTPMethodView
from sanic.blueprints import Blueprint
@ -30,8 +29,7 @@ def _generate_handlers_from_names(app, l):
@pytest.fixture
def simple_app():
app = Sanic('simple_app')
def simple_app(app):
handler_names = list(string.ascii_letters)
_generate_handlers_from_names(app, handler_names)
@ -54,8 +52,7 @@ def test_simple_url_for_getting(simple_app):
(URL_FOR_ARGS2, URL_FOR_VALUE2),
(URL_FOR_ARGS3, URL_FOR_VALUE3),
(URL_FOR_ARGS4, URL_FOR_VALUE4)])
def test_simple_url_for_getting_with_more_params(args, url):
app = Sanic('more_url_build')
def test_simple_url_for_getting_with_more_params(app, args, url):
@app.route('/myurl')
def passes(request):
@ -67,8 +64,7 @@ def test_simple_url_for_getting_with_more_params(args, url):
assert response.text == 'this should pass'
def test_fails_if_endpoint_not_found():
app = Sanic('fail_url_build')
def test_fails_if_endpoint_not_found(app):
@app.route('/fail')
def fail(request):
@ -80,14 +76,12 @@ def test_fails_if_endpoint_not_found():
assert str(e.value) == 'Endpoint with name `passes` was not found'
def test_fails_url_build_if_param_not_passed():
def test_fails_url_build_if_param_not_passed(app):
url = '/'
for letter in string.ascii_letters:
url += '<{}>/'.format(letter)
app = Sanic('fail_url_build')
@app.route(url)
def fail(request):
return text('this should fail')
@ -103,8 +97,7 @@ def test_fails_url_build_if_param_not_passed():
assert 'Required parameter `Z` was not passed to url_for' in str(e.value)
def test_fails_url_build_if_params_not_passed():
app = Sanic('fail_url_build')
def test_fails_url_build_if_params_not_passed(app):
@app.route('/fail')
def fail(request):
@ -126,8 +119,7 @@ PASSING_KWARGS = {
EXPECTED_BUILT_URL = '/4/woof/ba/normal/1.001'
def test_fails_with_int_message():
app = Sanic('fail_url_build')
def test_fails_with_int_message(app):
@app.route(COMPLEX_PARAM_URL)
def fail(request):
@ -145,8 +137,7 @@ def test_fails_with_int_message():
assert str(e.value) == expected_error
def test_fails_with_two_letter_string_message():
app = Sanic('fail_url_build')
def test_fails_with_two_letter_string_message(app):
@app.route(COMPLEX_PARAM_URL)
def fail(request):
@ -165,8 +156,7 @@ def test_fails_with_two_letter_string_message():
assert str(e.value) == expected_error
def test_fails_with_number_message():
app = Sanic('fail_url_build')
def test_fails_with_number_message(app):
@app.route(COMPLEX_PARAM_URL)
def fail(request):
@ -185,8 +175,7 @@ def test_fails_with_number_message():
assert str(e.value) == expected_error
def test_adds_other_supplied_values_as_query_string():
app = Sanic('passes')
def test_adds_other_supplied_values_as_query_string(app):
@app.route(COMPLEX_PARAM_URL)
def passes(request):
@ -205,8 +194,7 @@ def test_adds_other_supplied_values_as_query_string():
@pytest.fixture
def blueprint_app():
app = Sanic('blueprints')
def blueprint_app(app):
first_print = Blueprint('first', url_prefix='/first')
second_print = Blueprint('second', url_prefix='/second')
@ -252,8 +240,7 @@ def test_blueprints_work_with_params(blueprint_app):
@pytest.fixture
def methodview_app():
app = Sanic('methodview')
def methodview_app(app):
class ViewOne(HTTPMethodView):
def get(self, request):

View File

@ -3,7 +3,6 @@ import os
import pytest
from sanic import Sanic
from sanic.blueprints import Blueprint
@ -27,8 +26,7 @@ def get_file_content(static_file_directory, file_name):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt', 'python.png'])
def test_static_file(static_file_directory, file_name):
app = Sanic('test_static')
def test_static_file(app, static_file_directory, file_name):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name))
app.static(
@ -102,9 +100,7 @@ def test_static_file(static_file_directory, file_name):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
@pytest.mark.parametrize('base_uri', ['/static', '', '/dir'])
def test_static_directory(file_name, base_uri, static_file_directory):
app = Sanic('test_static')
def test_static_directory(app, file_name, base_uri, static_file_directory):
app.static(base_uri, static_file_directory)
base_uri2 = base_uri + '/2'
app.static(base_uri2, static_file_directory, name='uploads')
@ -156,10 +152,8 @@ def test_static_directory(file_name, base_uri, static_file_directory):
assert response.body == get_file_content(static_file_directory, file_name)
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_head_request(file_name, static_file_directory):
app = Sanic('test_static')
def test_static_head_request(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)
@ -198,8 +192,7 @@ def test_static_head_request(file_name, static_file_directory):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_content_range_correct(file_name, static_file_directory):
app = Sanic('test_static')
def test_static_content_range_correct(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)
@ -250,8 +243,7 @@ def test_static_content_range_correct(file_name, static_file_directory):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_content_range_front(file_name, static_file_directory):
app = Sanic('test_static')
def test_static_content_range_front(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)
@ -302,8 +294,7 @@ def test_static_content_range_front(file_name, static_file_directory):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_content_range_back(file_name, static_file_directory):
app = Sanic('test_static')
def test_static_content_range_back(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)
@ -354,8 +345,7 @@ def test_static_content_range_back(file_name, static_file_directory):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_content_range_empty(file_name, static_file_directory):
app = Sanic('test_static')
def test_static_content_range_empty(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)
@ -401,8 +391,7 @@ def test_static_content_range_empty(file_name, static_file_directory):
@pytest.mark.parametrize('file_name', ['test.file', 'decode me.txt'])
def test_static_content_range_error(file_name, static_file_directory):
app = Sanic('test_static')
def test_static_content_range_error(app, file_name, static_file_directory):
app.static(
'/testing.file', get_file_path(static_file_directory, file_name),
use_content_range=True)

View File

@ -1,14 +1,12 @@
from json import loads as json_loads, dumps as json_dumps
from sanic import Sanic
from sanic.response import json, text
from json import dumps as json_dumps
from sanic.response import text
# ------------------------------------------------------------ #
# UTF-8
# ------------------------------------------------------------ #
def test_utf8_query_string():
app = Sanic('test_utf8_query_string')
def test_utf8_query_string(app):
@app.route('/')
async def handler(request):
@ -18,8 +16,7 @@ def test_utf8_query_string():
assert request.args.get('utf8') == ''
def test_utf8_response():
app = Sanic('test_utf8_response')
def test_utf8_response(app):
@app.route('/')
async def handler(request):
@ -29,8 +26,7 @@ def test_utf8_response():
assert response.text == ''
def skip_test_utf8_route():
app = Sanic('skip_test_utf8_route')
def skip_test_utf8_route(app):
@app.route('/')
async def handler(request):
@ -41,8 +37,7 @@ def skip_test_utf8_route():
assert response.text == 'OK'
def test_utf8_post_json():
app = Sanic('test_utf8_post_json')
def test_utf8_post_json(app):
@app.route('/')
async def handler(request):

View File

@ -1,9 +1,7 @@
from sanic import Sanic
from sanic.response import json, text
from sanic.response import text
def test_vhosts():
app = Sanic('test_vhosts')
def test_vhosts(app):
@app.route('/', host="example.com")
async def handler(request):
@ -22,8 +20,7 @@ def test_vhosts():
assert response.text == "You're at subdomain.example.com!"
def test_vhosts_with_list():
app = Sanic('test_vhosts')
def test_vhosts_with_list(app):
@app.route('/', host=["hello.com", "world.com"])
async def handler(request):
@ -37,8 +34,8 @@ def test_vhosts_with_list():
request, response = app.test_client.get('/', headers=headers)
assert response.text == "Hello, world!"
def test_vhosts_with_defaults():
app = Sanic('test_vhosts')
def test_vhosts_with_defaults(app):
@app.route('/', host="hello.com")
async def handler(request):

View File

@ -1,6 +1,5 @@
import pytest as pytest
from sanic import Sanic
from sanic.exceptions import InvalidUsage
from sanic.response import text, HTTPResponse
from sanic.views import HTTPMethodView, CompositionView
@ -10,8 +9,7 @@ from sanic.constants import HTTP_METHODS
@pytest.mark.parametrize('method', HTTP_METHODS)
def test_methods(method):
app = Sanic('test_methods')
def test_methods(app, method):
class DummyView(HTTPMethodView):
@ -44,8 +42,7 @@ def test_methods(method):
assert response.headers['method'] == method
def test_unexisting_methods():
app = Sanic('test_unexisting_methods')
def test_unexisting_methods(app):
class DummyView(HTTPMethodView):
@ -59,8 +56,7 @@ def test_unexisting_methods():
assert response.text == 'Error: Method POST not allowed for URL /'
def test_argument_methods():
app = Sanic('test_argument_methods')
def test_argument_methods(app):
class DummyView(HTTPMethodView):
@ -74,8 +70,7 @@ def test_argument_methods():
assert response.text == 'I am get method with test123'
def test_with_bp():
app = Sanic('test_with_bp')
def test_with_bp(app):
bp = Blueprint('test_text')
class DummyView(HTTPMethodView):
@ -93,8 +88,7 @@ def test_with_bp():
assert response.text == 'I am get method'
def test_with_bp_with_url_prefix():
app = Sanic('test_with_bp_with_url_prefix')
def test_with_bp_with_url_prefix(app):
bp = Blueprint('test_text', url_prefix='/test1')
class DummyView(HTTPMethodView):
@ -110,8 +104,7 @@ def test_with_bp_with_url_prefix():
assert response.text == 'I am get method'
def test_with_middleware():
app = Sanic('test_with_middleware')
def test_with_middleware(app):
class DummyView(HTTPMethodView):
@ -132,9 +125,7 @@ def test_with_middleware():
assert type(results[0]) is Request
def test_with_middleware_response():
app = Sanic('test_with_middleware_response')
def test_with_middleware_response(app):
results = []
@app.middleware('request')
@ -161,8 +152,7 @@ def test_with_middleware_response():
assert isinstance(results[2], HTTPResponse)
def test_with_custom_class_methods():
app = Sanic('test_with_custom_class_methods')
def test_with_custom_class_methods(app):
class DummyView(HTTPMethodView):
global_var = 0
@ -179,9 +169,7 @@ def test_with_custom_class_methods():
assert response.text == 'I am get method and global var is 10'
def test_with_decorator():
app = Sanic('test_with_decorator')
def test_with_decorator(app):
results = []
def stupid_decorator(view):
@ -227,9 +215,7 @@ def test_composition_view_rejects_duplicate_methods():
@pytest.mark.parametrize('method', HTTP_METHODS)
def test_composition_view_runs_methods_as_expected(method):
app = Sanic('test_composition_view')
def test_composition_view_runs_methods_as_expected(app, method):
view = CompositionView()
def first(request):
@ -251,9 +237,7 @@ def test_composition_view_runs_methods_as_expected(method):
@pytest.mark.parametrize('method', HTTP_METHODS)
def test_composition_view_rejects_invalid_methods(method):
app = Sanic('test_composition_view')
def test_composition_view_rejects_invalid_methods(app, method):
view = CompositionView()
view.add(['GET', 'POST', 'PUT'], lambda x: text('first method'))