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

Merge pull request #1562 from huge-success/testing-client

Browse Source 
Testing client
This commit is contained in:
7 2019-05-03 06:32:26 +08:00 committed by GitHub
commit ae2b8f0056
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
56 changed files with 649 additions and 578 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.appveyor.yml
View File

@ -2,11 +2,6 @@ version: "{branch}.{build}"
environment:
matrix:
- TOXENV: py35-no-ext
PYTHON: "C:\\Python35-x64"
PYTHON_VERSION: "3.5.x"
PYTHON_ARCH: "64"
- TOXENV: py36-no-ext
PYTHON: "C:\\Python36-x64"
PYTHON_VERSION: "3.6.x"

0
pyproject.toml → .black.toml
View File

4
.travis.yml
View File

@ -5,10 +5,6 @@ cache:
- $HOME/.cache/pip
matrix:
include:
- env: TOX_ENV=py35
python: 3.5
- env: TOX_ENV=py35-no-ext
python: 3.5
- env: TOX_ENV=py36
python: 3.6
- env: TOX_ENV=py36-no-ext

15
CHANGELOG.md
View File

@ -1,3 +1,18 @@
Version 19.6
------------
19.6.0
- Changes:
- Remove `aiohttp` dependencey and create new `SanicTestClient` based upon
[`requests-async`](https://github.com/encode/requests-async).
- Deprecation:
- Support for Python 3.5
Note: Sanic will not support Python 3.5 from version 19.6 and forward. However,
version 18.12LTS will have its support period extended thru December 2020, and
therefore passing Python's official support version 3.5, which is set to expire
in September 2020.
Version 19.3
-------------
19.3.1

2
LICENSE
View File

@ -1,6 +1,6 @@
MIT License
Copyright (c) 2016-present Channel Cat
Copyright (c) 2016-present Sanic Community
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal

6
Makefile
View File

@ -47,12 +47,12 @@ ifdef include_tests
isort -rc sanic tests
else
$(info Sorting Imports)
isort -rc sanic
isort -rc sanic tests
endif
endif
black:
black --config ./pyproject.toml sanic tests
black --config ./.black.toml sanic tests
fix-import: black
isort -rc sanic
isort -rc sanic tests

7
README.rst
View File

@ -17,6 +17,8 @@ Sanic | Build fast. Run fast.
- | |PyPI| |PyPI version| |Wheel| |Supported implementations| |Code style black|
* - Support
- | |Forums| |Join the chat at https://gitter.im/sanic-python/Lobby|
* - Stats
- | |Downloads|
.. |Forums| image:: https://img.shields.io/badge/forums-community-ff0068.svg
:target: https://community.sanicframework.org/
@ -42,10 +44,13 @@ Sanic | Build fast. Run fast.
.. |Supported implementations| image:: https://img.shields.io/pypi/implementation/sanic.svg
:alt: Supported implementations
:target: https://pypi.python.org/pypi/sanic
.. |Downloads| image:: https://pepy.tech/badge/sanic/month
:alt: Downloads
:target: https://pepy.tech/project/sanic
.. end-badges
Sanic is a Python web server and web framework that's written to go fast. It allows the usage of the ``async/await`` syntax added in Python 3.5, which makes your code non-blocking and speedy.
Sanic is a **Python 3.6+** web server and web framework that's written to go fast. It allows the usage of the ``async/await`` syntax added in Python 3.5, which makes your code non-blocking and speedy.
`Source code on GitHub <https://github.com/huge-success/sanic/>`_ | `Help and discussion board <https://community.sanicframework.org/>`_.

15
docs/sanic/changelog.md
View File

@ -1,3 +1,18 @@
Version 19.6
------------
19.6.0
- Changes:
- Remove `aiohttp` dependencey and create new `SanicTestClient` based upon
[`requests-async`](https://github.com/encode/requests-async).
- Deprecation:
- Support for Python 3.5
Note: Sanic will not support Python 3.5 from version 19.6 and forward. However,
version 18.12LTS will have its support period extended thru December 2020, and
therefore passing Python's official support version 3.5, which is set to expire
in September 2020.
Version 19.3
-------------
19.3.1

12
docs/sanic/testing.md
View File

@ -1,8 +1,8 @@
# Testing
Sanic endpoints can be tested locally using the `test_client` object, which
depends on the additional [aiohttp](https://aiohttp.readthedocs.io/en/stable/)
library.
depends on the additional [`requests-async`](https://github.com/encode/requests-async)
library, which implements an API that mirrors the `requests` library.
The `test_client` exposes `get`, `post`, `put`, `delete`, `patch`, `head` and `options` methods
for you to run against your application. A simple example (using pytest) is like follows:
@ -21,7 +21,7 @@ def test_index_put_not_allowed():
```
Internally, each time you call one of the `test_client` methods, the Sanic app is run at `127.0.0.1:42101` and
your test request is executed against your application, using `aiohttp`.
your test request is executed against your application, using `requests-async`.
The `test_client` methods accept the following arguments and keyword arguments:
@ -33,7 +33,7 @@ The `test_client` methods accept the following arguments and keyword arguments:
- `server_kwargs` *(default `{}`) a dict of additional arguments to pass into `app.run` before the test request is run.
- `debug` *(default `False`)* A boolean which determines whether to run the server in debug mode.
The function further takes the `*request_args` and `**request_kwargs`, which are passed directly to the aiohttp ClientSession request.
The function further takes the `*request_args` and `**request_kwargs`, which are passed directly to the request.
For example, to supply data to a GET request, you would do the following:
@ -55,8 +55,8 @@ def test_post_json_request_includes_data():
More information about
the available arguments to aiohttp can be found
[in the documentation for ClientSession](https://aiohttp.readthedocs.io/en/stable/client_reference.html#client-session).
the available arguments to `requests-async` can be found
[in the documentation for `requests`](https://2.python-requests.org/en/master/).
## Using a random port

63
sanic/testing.py
View File

@ -1,6 +1,9 @@
from json import JSONDecodeError
from socket import socket
import requests_async as requests
import websockets
from sanic.exceptions import MethodNotSupported
from sanic.log import logger
from sanic.response import text
@ -16,32 +19,41 @@ class SanicTestClient:
self.app = app
self.port = port
async def _local_request(self, method, url, cookies=None, *args, **kwargs):
import aiohttp
def get_new_session(self):
return requests.Session()
async def _local_request(self, method, url, *args, **kwargs):
logger.info(url)
conn = aiohttp.TCPConnector(ssl=False)
async with aiohttp.ClientSession(
cookies=cookies, connector=conn
) as session:
async with getattr(session, method.lower())(
url, *args, **kwargs
) as response:
try:
response.text = await response.text()
except UnicodeDecodeError:
response.text = None
raw_cookies = kwargs.pop("raw_cookies", None)
if method == "websocket":
async with websockets.connect(url, *args, **kwargs) as websocket:
websocket.opened = websocket.open
return websocket
else:
async with self.get_new_session() as session:
try:
response.json = await response.json()
except (
JSONDecodeError,
UnicodeDecodeError,
aiohttp.ClientResponseError,
):
response = await getattr(session, method.lower())(
url, verify=False, *args, **kwargs
)
except NameError:
raise Exception(response.status_code)
try:
response.json = response.json()
except (JSONDecodeError, UnicodeDecodeError):
response.json = None
response.body = await response.read()
response.status = response.status_code
response.content_type = response.headers.get("content-type")
if raw_cookies:
response.raw_cookies = {}
for cookie in response.cookies:
response.raw_cookies[cookie.name] = cookie
return response
def _sanic_endpoint_test(
@ -83,11 +95,15 @@ class SanicTestClient:
server_kwargs = dict(sock=sock, **server_kwargs)
host, port = sock.getsockname()
if uri.startswith(("http:", "https:", "ftp:", "ftps://", "//")):
if uri.startswith(
("http:", "https:", "ftp:", "ftps://", "//", "ws:", "wss:")
):
url = uri
else:
url = "http://{host}:{port}{uri}".format(
host=host, port=port, uri=uri
uri = uri if uri.startswith("/") else "/{uri}".format(uri=uri)
scheme = "ws" if method == "websocket" else "http"
url = "{scheme}://{host}:{port}{uri}".format(
scheme=scheme, host=host, port=port, uri=uri
)
@self.app.listener("after_server_start")
@ -146,3 +162,6 @@ class SanicTestClient:
def head(self, *args, **kwargs):
return self._sanic_endpoint_test("head", *args, **kwargs)
def websocket(self, *args, **kwargs):
return self._sanic_endpoint_test("websocket", *args, **kwargs)

2
setup.cfg
View File

@ -14,7 +14,7 @@ multi_line_output = 3
not_skip = __init__.py
[version]
current_version = 0.8.3
current_version = 19.3.1
file = sanic/__init__.py
current_version_pattern = __version__ = "{current_version}"
new_version_pattern = __version__ = "{new_version}"

14
setup.py
View File

@ -50,12 +50,12 @@ with open_local(["README.rst"]) as rm:
setup_kwargs = {
"name": "sanic",
"version": version,
"url": "http://github.com/channelcat/sanic/",
"url": "http://github.com/huge-success/sanic/",
"license": "MIT",
"author": "Channel Cat",
"author_email": "channelcat@gmail.com",
"author": "Sanic Community",
"author_email": "admhpkns@gmail.com",
"description": (
"A microframework based on uvloop, httptools, and learnings of flask"
"A web server and web framework that's written to go fast. Build fast. Run fast."
),
"long_description": long_description,
"packages": ["sanic"],
@ -64,7 +64,6 @@ setup_kwargs = {
"Development Status :: 4 - Beta",
"Environment :: Web Environment",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3.5",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
],
@ -90,7 +89,8 @@ tests_require = [
"multidict>=4.0,<5.0",
"gunicorn",
"pytest-cov",
"aiohttp>=2.3.0,<=3.2.1",
"httpcore==0.1.1",
"requests-async==0.4.0",
"beautifulsoup4",
uvloop,
ujson,
@ -119,7 +119,7 @@ extras_require = {
"recommonmark",
"sphinxcontrib-asyncio",
"docutils",
"pygments"
"pygments",
],
}

2
tests/benchmark/test_route_resolution_benchmark.py
View File

@ -1,8 +1,10 @@
from random import choice, seed
from pytest import mark
import sanic.router
seed("Pack my box with five dozen liquor jugs.")
# Disable Caching for testing purpose

1
tests/conftest.py
View File

@ -9,6 +9,7 @@ import pytest
from sanic import Sanic
from sanic.router import RouteExists, Router
random.seed("Pack my box with five dozen liquor jugs.")
if sys.platform in ["win32", "cygwin"]:

7
tests/performance/aiohttp/simple_server.py
View File

@ -1,10 +1,13 @@
# Run with python3 simple_server.py PORT
from aiohttp import web
import asyncio
import sys
import uvloop
import ujson as json
import uvloop
from aiohttp import web
loop = uvloop.new_event_loop()
asyncio.set_event_loop(loop)

3
tests/performance/bottle/simple_server.py
View File

@ -1,8 +1,9 @@
# Run with: gunicorn --workers=1 --worker-class=meinheld.gmeinheld.MeinheldWorker -b :8000 simple_server:app
import bottle
from bottle import route, run
import ujson
from bottle import route, run
@route("/")
def index():

5
tests/performance/kyoukai/simple_server.py
View File

@ -1,10 +1,13 @@
# Run with: python3 -O simple_server.py
import asyncio
from kyoukai import Kyoukai, HTTPRequestContext
import logging
import ujson
import uvloop
from kyoukai import HTTPRequestContext, Kyoukai
loop = uvloop.new_event_loop()
asyncio.set_event_loop(loop)

14
tests/performance/sanic/http_response.py
View File

@ -1,16 +1,18 @@
import asyncpg
import sys
import os
import inspect
import os
import sys
import timeit
import asyncpg
from sanic.response import json
currentdir = os.path.dirname(
os.path.abspath(inspect.getfile(inspect.currentframe()))
)
sys.path.insert(0, currentdir + "/../../../")
import timeit
from sanic.response import json
print(json({"test": True}).output())

10
tests/performance/sanic/simple_server.py
View File

@ -1,14 +1,16 @@
import sys
import os
import inspect
import os
import sys
from sanic import Sanic
from sanic.response import json
currentdir = os.path.dirname(
os.path.abspath(inspect.getfile(inspect.currentframe()))
)
sys.path.insert(0, currentdir + "/../../../")
from sanic import Sanic
from sanic.response import json
app = Sanic("test")

14
tests/performance/sanic/varied_server.py
View File

@ -1,15 +1,17 @@
import sys
import os
import inspect
import os
import sys
from sanic import Sanic
from sanic.exceptions import ServerError
from sanic.response import json, text
currentdir = os.path.dirname(
os.path.abspath(inspect.getfile(inspect.currentframe()))
)
sys.path.insert(0, currentdir + "/../../../")
from sanic import Sanic
from sanic.response import json, text
from sanic.exceptions import ServerError
app = Sanic("test")
@ -56,8 +58,6 @@ def query_string(request):
)
import sys
app.run(host="0.0.0.0", port=sys.argv[1])

1
tests/performance/tornado/simple_server.py
View File

@ -1,5 +1,6 @@
# Run with: python simple_server.py
import ujson
from tornado import ioloop, web

13
tests/performance/wheezy/simple_server.py
View File

@ -2,15 +2,16 @@
""" Minimal helloworld application.
"""
from wheezy.http import HTTPResponse
from wheezy.http import WSGIApplication
import ujson
from wheezy.http import HTTPResponse, WSGIApplication
from wheezy.http.response import json_response
from wheezy.routing import url
from wheezy.web.handlers import BaseHandler
from wheezy.web.middleware import bootstrap_defaults
from wheezy.web.middleware import path_routing_middleware_factory
import ujson
from wheezy.web.middleware import (
bootstrap_defaults,
path_routing_middleware_factory,
)
class WelcomeHandler(BaseHandler):

3
tests/test_app.py
View File

@ -3,6 +3,7 @@ import logging
import sys
from inspect import isawaitable
import pytest
from sanic.exceptions import SanicException
@ -11,7 +12,7 @@ from sanic.response import text
def uvloop_installed():
try:
import uvloop
import uvloop # noqa
return True
except ImportError:

3
tests/test_blueprint_group.py
View File

@ -3,7 +3,8 @@ from pytest import raises
from sanic.app import Sanic
from sanic.blueprints import Blueprint
from sanic.request import Request
from sanic.response import text, HTTPResponse
from sanic.response import HTTPResponse, text
MIDDLEWARE_INVOKE_COUNTER = {"request": 0, "response": 0}

31
tests/test_blueprints.py
View File

@ -7,9 +7,9 @@ import pytest
from sanic.app import Sanic
from sanic.blueprints import Blueprint
from sanic.constants import HTTP_METHODS
from sanic.exceptions import NotFound, ServerError, InvalidUsage
from sanic.exceptions import InvalidUsage, NotFound, ServerError
from sanic.request import Request
from sanic.response import text, json
from sanic.response import json, text
from sanic.views import CompositionView
@ -467,16 +467,8 @@ def test_bp_shorthand(app):
request, response = app.test_client.get("/delete")
assert response.status == 405
request, response = app.test_client.get(
"/ws/",
headers={
"Upgrade": "websocket",
"Connection": "upgrade",
"Sec-WebSocket-Key": "dGhlIHNhbXBsZSBub25jZQ==",
"Sec-WebSocket-Version": "13",
},
)
assert response.status == 101
request, response = app.test_client.websocket("/ws/")
assert response.opened is True
assert ev.is_set()
@ -595,14 +587,13 @@ def test_blueprint_middleware_with_args(app: Sanic):
"/wa", headers={"content-type": "plain/text"}
)
assert response.json.get("test") == "value"
d = {}
@pytest.mark.parametrize("file_name", ["test.file"])
def test_static_blueprint_name(app: Sanic, static_file_directory, file_name):
current_file = inspect.getfile(inspect.currentframe())
with open(current_file, "rb") as file:
current_file_contents = file.read()
file.read()
bp = Blueprint(name="static", url_prefix="/static", strict_slashes=False)
@ -662,16 +653,8 @@ def test_websocket_route(app: Sanic):
app.blueprint(bp)
_, response = app.test_client.get(
"/ws/test",
headers={
"Upgrade": "websocket",
"Connection": "upgrade",
"Sec-WebSocket-Key": "dGhlIHNhbXBsZSBub25jZQ==",
"Sec-WebSocket-Version": "13",
},
)
assert response.status == 101
_, response = app.test_client.websocket("/ws/test")
assert response.opened is True
assert event.is_set()

5
tests/test_config.py
View File

@ -1,12 +1,13 @@
from contextlib import contextmanager
from os import environ
from pathlib import Path
from contextlib import contextmanager
from tempfile import TemporaryDirectory
from textwrap import dedent
import pytest
from sanic import Sanic
from sanic.config import Config, DEFAULT_CONFIG
from sanic.config import DEFAULT_CONFIG, Config
from sanic.exceptions import PyFileError

63
tests/test_cookies.py
View File

@ -1,8 +1,11 @@
from datetime import datetime, timedelta
from http.cookies import SimpleCookie
from sanic.response import text
import pytest
from sanic.cookies import Cookie, DEFAULT_MAX_AGE
from sanic.cookies import Cookie
from sanic.response import text
# ------------------------------------------------------------ #
# GET
@ -100,7 +103,7 @@ def test_cookie_deletion(app):
assert int(response_cookies["i_want_to_die"]["max-age"]) == 0
with pytest.raises(KeyError):
_ = response.cookies["i_never_existed"]
response.cookies["i_never_existed"]
def test_cookie_reserved_cookie():
@ -135,7 +138,7 @@ def test_cookie_set_same_key(app):
request, response = app.test_client.get("/", cookies=cookies)
assert response.status == 200
assert response.cookies["test"].value == "pass"
assert response.cookies["test"] == "pass"
@pytest.mark.parametrize("max_age", ["0", 30, 30.0, 30.1, "30", "test"])
@ -149,19 +152,42 @@ def test_cookie_max_age(app, max_age):
response.cookies["test"]["max-age"] = max_age
return response
request, response = app.test_client.get("/", cookies=cookies)
request, response = app.test_client.get(
"/", cookies=cookies, raw_cookies=True
)
assert response.status == 200
assert response.cookies["test"].value == "pass"
cookie = response.cookies.get("test")
if (
str(max_age).isdigit()
and int(max_age) == float(max_age)
and int(max_age) != 0
):
cookie_expires = datetime.utcfromtimestamp(
response.raw_cookies["test"].expires
).replace(microsecond=0)
if str(max_age).isdigit() and int(max_age) == float(max_age):
assert response.cookies["test"]["max-age"] == str(max_age)
# Grabbing utcnow after the response may lead to it being off slightly.
# Therefore, we 0 out the microseconds, and accept the test if there
# is a 1 second difference.
expires = datetime.utcnow().replace(microsecond=0) + timedelta(
seconds=int(max_age)
)
assert cookie == "pass"
assert (
cookie_expires == expires
or cookie_expires == expires + timedelta(seconds=-1)
)
else:
assert response.cookies["test"]["max-age"] == str(DEFAULT_MAX_AGE)
assert cookie is None
@pytest.mark.parametrize("expires", [datetime.now() + timedelta(seconds=60)])
@pytest.mark.parametrize(
"expires", [datetime.utcnow() + timedelta(seconds=60)]
)
def test_cookie_expires(app, expires):
expires = expires.replace(microsecond=0)
cookies = {"test": "wait"}
@app.get("/")
@ -171,15 +197,16 @@ def test_cookie_expires(app, expires):
response.cookies["test"]["expires"] = expires
return response
request, response = app.test_client.get("/", cookies=cookies)
request, response = app.test_client.get(
"/", cookies=cookies, raw_cookies=True
)
cookie_expires = datetime.utcfromtimestamp(
response.raw_cookies["test"].expires
).replace(microsecond=0)
assert response.status == 200
assert response.cookies["test"].value == "pass"
if isinstance(expires, datetime):
expires = expires.strftime("%a, %d-%b-%Y %T GMT")
assert response.cookies["test"]["expires"] == expires
assert response.cookies["test"] == "pass"
assert cookie_expires == expires
@pytest.mark.parametrize("expires", ["Fri, 21-Dec-2018 15:30:00 GMT"])

6
tests/test_create_task.py
View File

@ -1,7 +1,9 @@
from sanic.response import text
from threading import Event
import asyncio
from queue import Queue
from threading import Event
from sanic.response import text
def test_create_task(app):

2
tests/test_custom_protocol.py
View File

@ -1,5 +1,5 @@
from sanic.server import HttpProtocol
from sanic.response import text
from sanic.server import HttpProtocol
class CustomHttpProtocol(HttpProtocol):

3
tests/test_dynamic_routes.py
View File

@ -1,6 +1,7 @@
import pytest
from sanic.response import text
from sanic.router import RouteExists
import pytest
@pytest.mark.parametrize(

11
tests/test_exceptions.py
View File

@ -1,10 +1,17 @@
import pytest
from bs4 import BeautifulSoup
from sanic import Sanic
from sanic.exceptions import (
Forbidden,
InvalidUsage,
NotFound,
ServerError,
Unauthorized,
abort,
)
from sanic.response import text
from sanic.exceptions import InvalidUsage, ServerError, NotFound, Unauthorized
from sanic.exceptions import Forbidden, abort
class SanicExceptionTestException(Exception):

10
tests/test_exceptions_handler.py
View File

@ -1,9 +1,11 @@
from sanic import Sanic
from sanic.response import text
from sanic.exceptions import InvalidUsage, ServerError, NotFound
from sanic.handlers import ErrorHandler
from bs4 import BeautifulSoup
from sanic import Sanic
from sanic.exceptions import InvalidUsage, NotFound, ServerError
from sanic.handlers import ErrorHandler
from sanic.response import text
exception_handler_app = Sanic("test_exception_handler")

287
tests/test_keep_alive_timeout.py
View File

@ -1,39 +1,143 @@
from json import JSONDecodeError
from sanic import Sanic
import asyncio
import functools
import socket
from asyncio import sleep as aio_sleep
from http.client import _encode
from json import JSONDecodeError
import httpcore
import requests_async as requests
from httpcore import PoolTimeout
from sanic import Sanic, server
from sanic.response import text
from sanic import server
import aiohttp
from aiohttp import TCPConnector
from sanic.testing import SanicTestClient, HOST, PORT
from sanic.testing import HOST, PORT, SanicTestClient
# import traceback
CONFIG_FOR_TESTS = {"KEEP_ALIVE_TIMEOUT": 2, "KEEP_ALIVE": True}
old_conn = None
class ReuseableTCPConnector(TCPConnector):
def __init__(self, *args, **kwargs):
super(ReuseableTCPConnector, self).__init__(*args, **kwargs)
self.old_proto = None
async def connect(self, req, *args, **kwargs):
new_conn = await super(ReuseableTCPConnector, self).connect(
req, *args, **kwargs
class ReusableSanicConnectionPool(httpcore.ConnectionPool):
async def acquire_connection(self, url, ssl, timeout):
global old_conn
if timeout.connect_timeout and not timeout.pool_timeout:
timeout.pool_timeout = timeout.connect_timeout
key = (url.scheme, url.hostname, url.port, ssl, timeout)
try:
connection = self._keepalive_connections[key].pop()
if not self._keepalive_connections[key]:
del self._keepalive_connections[key]
self.num_keepalive_connections -= 1
self.num_active_connections += 1
except (KeyError, IndexError):
ssl_context = await self.get_ssl_context(url, ssl)
try:
await asyncio.wait_for(
self._max_connections.acquire(), timeout.pool_timeout
)
if self.old_proto is not None:
if self.old_proto != new_conn._protocol:
except asyncio.TimeoutError:
raise PoolTimeout()
release = functools.partial(self.release_connection, key=key)
connection = httpcore.connections.Connection(
timeout=timeout, on_release=release
)
self.num_active_connections += 1
await connection.open(url.hostname, url.port, ssl=ssl_context)
if old_conn is not None:
if old_conn != connection:
raise RuntimeError(
"We got a new connection, wanted the same one!"
)
print(new_conn.__dict__)
self.old_proto = new_conn._protocol
return new_conn
old_conn = connection
return connection
class ReusableSanicAdapter(requests.adapters.HTTPAdapter):
def __init__(self):
self.pool = ReusableSanicConnectionPool()
async def send(
self,
request,
stream=False,
timeout=None,
verify=True,
cert=None,
proxies=None,
):
method = request.method
url = request.url
headers = [
(_encode(k), _encode(v)) for k, v in request.headers.items()
]
if not request.body:
body = b""
elif isinstance(request.body, str):
body = _encode(request.body)
else:
body = request.body
if isinstance(timeout, tuple):
timeout_kwargs = {
"connect_timeout": timeout[0],
"read_timeout": timeout[1],
}
else:
timeout_kwargs = {
"connect_timeout": timeout,
"read_timeout": timeout,
}
ssl = httpcore.SSLConfig(cert=cert, verify=verify)
timeout = httpcore.TimeoutConfig(**timeout_kwargs)
try:
response = await self.pool.request(
method,
url,
headers=headers,
body=body,
stream=stream,
ssl=ssl,
timeout=timeout,
)
except (httpcore.BadResponse, socket.error) as err:
raise ConnectionError(err, request=request)
except httpcore.ConnectTimeout as err:
raise requests.exceptions.ConnectTimeout(err, request=request)
except httpcore.ReadTimeout as err:
raise requests.exceptions.ReadTimeout(err, request=request)
return self.build_response(request, response)
class ResusableSanicSession(requests.Session):
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
adapter = ReusableSanicAdapter()
self.mount("http://", adapter)
self.mount("https://", adapter)
class ReuseableSanicTestClient(SanicTestClient):
def __init__(self, app, loop=None):
super(ReuseableSanicTestClient, self).__init__(app)
super().__init__(app)
if loop is None:
loop = asyncio.get_event_loop()
self._loop = loop
@ -51,12 +155,11 @@ class ReuseableSanicTestClient(SanicTestClient):
debug=False,
server_kwargs={"return_asyncio_server": True},
*request_args,
**request_kwargs
**request_kwargs,
):
loop = self._loop
results = [None, None]
exceptions = []
do_kill_server = request_kwargs.pop("end_server", False)
if gather_request:
def _collect_request(request):
@ -65,21 +168,27 @@ class ReuseableSanicTestClient(SanicTestClient):
self.app.request_middleware.appendleft(_collect_request)
if uri.startswith(
("http:", "https:", "ftp:", "ftps://", "//", "ws:", "wss:")
):
url = uri
else:
uri = uri if uri.startswith("/") else "/{uri}".format(uri=uri)
scheme = "http"
url = "{scheme}://{host}:{port}{uri}".format(
scheme=scheme, host=HOST, port=PORT, uri=uri
)
@self.app.listener("after_server_start")
async def _collect_response(loop):
try:
if do_kill_server:
request_kwargs["end_session"] = True
response = await self._local_request(
method, uri, *request_args, **request_kwargs
method, url, *request_args, **request_kwargs
)
results[-1] = response
except Exception as e2:
import traceback
traceback.print_tb(e2.__traceback__)
# traceback.print_tb(e2.__traceback__)
exceptions.append(e2)
# Don't stop here! self.app.stop()
if self._server is not None:
_server = self._server
@ -94,27 +203,14 @@ class ReuseableSanicTestClient(SanicTestClient):
try:
loop._stopping = False
http_server = loop.run_until_complete(_server_co)
_server = loop.run_until_complete(_server_co)
except Exception as e1:
import traceback
traceback.print_tb(e1.__traceback__)
# traceback.print_tb(e1.__traceback__)
raise e1
self._server = _server = http_server
self._server = _server
server.trigger_events(self.app.listeners["after_server_start"], loop)
self.app.listeners["after_server_start"].pop()
if do_kill_server:
try:
_server.close()
self._server = None
loop.run_until_complete(_server.wait_closed())
self.app.stop()
except Exception as e3:
import traceback
traceback.print_tb(e3.__traceback__)
exceptions.append(e3)
if exceptions:
raise ValueError("Exception during request: {}".format(exceptions))
@ -137,59 +233,61 @@ class ReuseableSanicTestClient(SanicTestClient):
"Request object expected, got ({})".format(results)
)
def kill_server(self):
try:
if self._server:
self._server.close()
self._loop.run_until_complete(self._server.wait_closed())
self._server = None
self.app.stop()
if self._session:
self._loop.run_until_complete(self._session.close())
self._session = None
except Exception as e3:
raise e3
# Copied from SanicTestClient, but with some changes to reuse the
# same TCPConnection and the sane ClientSession more than once.
# Note, you cannot use the same session if you are in a _different_
# loop, so the changes above are required too.
async def _local_request(self, method, uri, cookies=None, *args, **kwargs):
async def _local_request(self, method, url, *args, **kwargs):
raw_cookies = kwargs.pop("raw_cookies", None)
request_keepalive = kwargs.pop(
"request_keepalive", CONFIG_FOR_TESTS["KEEP_ALIVE_TIMEOUT"]
)
if uri.startswith(("http:", "https:", "ftp:", "ftps://" "//")):
url = uri
else:
url = "http://{host}:{port}{uri}".format(
host=HOST, port=self.port, uri=uri
)
do_kill_session = kwargs.pop("end_session", False)
if self._session:
session = self._session
_session = self._session
else:
if self._tcp_connector:
conn = self._tcp_connector
else:
conn = ReuseableTCPConnector(
ssl=False,
loop=self._loop,
keepalive_timeout=request_keepalive,
)
self._tcp_connector = conn
session = aiohttp.ClientSession(
cookies=cookies, connector=conn, loop=self._loop
)
self._session = session
async with getattr(session, method.lower())(
url, *args, **kwargs
) as response:
_session = ResusableSanicSession()
self._session = _session
async with _session as session:
try:
response.text = await response.text()
except UnicodeDecodeError:
response.text = None
response = await getattr(session, method.lower())(
url,
verify=False,
timeout=request_keepalive,
*args,
**kwargs,
)
except NameError:
raise Exception(response.status_code)
try:
response.json = await response.json()
except (
JSONDecodeError,
UnicodeDecodeError,
aiohttp.ClientResponseError,
):
response.json = response.json()
except (JSONDecodeError, UnicodeDecodeError):
response.json = None
response.body = await response.read()
if do_kill_session:
await session.close()
self._session = None
response.status = response.status_code
response.content_type = response.headers.get("content-type")
if raw_cookies:
response.raw_cookies = {}
for cookie in response.cookies:
response.raw_cookies[cookie.name] = cookie
return response
@ -229,9 +327,10 @@ def test_keep_alive_timeout_reuse():
assert response.status == 200
assert response.text == "OK"
loop.run_until_complete(aio_sleep(1))
request, response = client.get("/1", end_server=True)
request, response = client.get("/1")
assert response.status == 200
assert response.text == "OK"
client.kill_server()
def test_keep_alive_client_timeout():
@ -241,20 +340,21 @@ def test_keep_alive_client_timeout():
asyncio.set_event_loop(loop)
client = ReuseableSanicTestClient(keep_alive_app_client_timeout, loop)
headers = {"Connection": "keep-alive"}
request, response = client.get("/1", headers=headers, request_keepalive=1)
try:
request, response = client.get(
"/1", headers=headers, request_keepalive=1
)
assert response.status == 200
assert response.text == "OK"
loop.run_until_complete(aio_sleep(2))
exception = None
try:
request, response = client.get(
"/1", end_server=True, request_keepalive=1
)
request, response = client.get("/1", request_keepalive=1)
except ValueError as e:
exception = e
assert exception is not None
assert isinstance(exception, ValueError)
assert "got a new connection" in exception.args[0]
client.kill_server()
def test_keep_alive_server_timeout():
@ -266,15 +366,15 @@ def test_keep_alive_server_timeout():
asyncio.set_event_loop(loop)
client = ReuseableSanicTestClient(keep_alive_app_server_timeout, loop)
headers = {"Connection": "keep-alive"}
request, response = client.get("/1", headers=headers, request_keepalive=60)
try:
request, response = client.get(
"/1", headers=headers, request_keepalive=60
)
assert response.status == 200
assert response.text == "OK"
loop.run_until_complete(aio_sleep(3))
exception = None
try:
request, response = client.get(
"/1", request_keepalive=60, end_server=True
)
request, response = client.get("/1", request_keepalive=60)
except ValueError as e:
exception = e
assert exception is not None
@ -283,3 +383,4 @@ def test_keep_alive_server_timeout():
"Connection reset" in exception.args[0]
or "got a new connection" in exception.args[0]
)
client.kill_server()

14
tests/test_logging.py
View File

@ -1,17 +1,17 @@
import uuid
import logging
import uuid
from io import StringIO
from importlib import reload
import pytest
from io import StringIO
from unittest.mock import Mock
import pytest
import sanic
from sanic.response import text
from sanic.log import LOGGING_CONFIG_DEFAULTS
from sanic import Sanic
from sanic.log import logger
from sanic.log import LOGGING_CONFIG_DEFAULTS, logger
from sanic.response import text
logging_format = """module: %(module)s; \

3
tests/test_logo.py
View File

@ -1,8 +1,9 @@
import logging
import asyncio
import logging
from sanic.config import BASE_LOGO
try:
import uvloop # noqa

2
tests/test_middleware.py
View File

@ -1,10 +1,12 @@
import logging
from asyncio import CancelledError
from sanic.exceptions import NotFound
from sanic.request import Request
from sanic.response import HTTPResponse, text
# ------------------------------------------------------------ #
# GET
# ------------------------------------------------------------ #

5
tests/test_multiprocessing.py
View File

@ -1,11 +1,12 @@
import multiprocessing
import pickle
import random
import signal
import pickle
import pytest
from sanic.testing import HOST, PORT
from sanic.response import text
from sanic.testing import HOST, PORT
@pytest.mark.skipif(

5
tests/test_named_routes.py
View File

@ -2,12 +2,13 @@
# -*- coding: utf-8 -*-
import asyncio
import pytest
from sanic.blueprints import Blueprint
from sanic.response import text
from sanic.exceptions import URLBuildError
from sanic.constants import HTTP_METHODS
from sanic.exceptions import URLBuildError
from sanic.response import text
# ------------------------------------------------------------ #

5
tests/test_redirect.py
View File

@ -1,7 +1,8 @@
import pytest
from urllib.parse import quote
from sanic.response import text, redirect
import pytest
from sanic.response import redirect, text
@pytest.fixture

2
tests/test_request_cancel.py
View File

@ -1,7 +1,7 @@
import asyncio
import contextlib
from sanic.response import text, stream
from sanic.response import stream, text
async def test_request_cancel_when_connection_lost(loop, app, test_client):

1
tests/test_request_data.py
View File

@ -2,6 +2,7 @@ import random
from sanic.response import json
try:
from ujson import loads
except ImportError:

8
tests/test_request_stream.py
View File

@ -1,10 +1,8 @@
import asyncio
from sanic.blueprints import Blueprint
from sanic.views import CompositionView
from sanic.views import HTTPMethodView
from sanic.views import stream as stream_decorator
from sanic.response import stream, text
from sanic.request import StreamBuffer
from sanic.response import stream, text
from sanic.views import CompositionView, HTTPMethodView
from sanic.views import stream as stream_decorator
data = "abc" * 10000000

226
tests/test_request_timeout.py
View File

@ -1,183 +1,73 @@
from json import JSONDecodeError
import asyncio
import httpcore
import requests_async as requests
from sanic import Sanic
import asyncio
from sanic.response import text
import aiohttp
from aiohttp import TCPConnector
from sanic.testing import SanicTestClient, HOST
try:
try:
# direct use
import packaging
version = packaging.version
except (ImportError, AttributeError):
# 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.")
from sanic.testing import SanicTestClient
aiohttp_version = version.parse(aiohttp.__version__)
class DelayableSanicConnectionPool(httpcore.ConnectionPool):
def __init__(self, request_delay=None, *args, **kwargs):
self._request_delay = request_delay
super().__init__(*args, **kwargs)
async def request(
self,
method,
url,
headers=(),
body=b"",
stream=False,
ssl=None,
timeout=None,
):
if ssl is None:
ssl = self.ssl_config
if timeout is None:
timeout = self.timeout
class DelayableTCPConnector(TCPConnector):
class RequestContextManager(object):
def __new__(cls, req, delay):
cls = super(
DelayableTCPConnector.RequestContextManager, cls
).__new__(cls)
cls.req = req
cls.send_task = None
cls.resp = None
cls.orig_send = getattr(req, "send")
cls.orig_start = None
cls.delay = delay
cls._acting_as = req
return cls
def __getattr__(self, item):
acting_as = self._acting_as
return getattr(acting_as, item)
async def start(self, connection, read_until_eof=False):
if self.send_task is None:
raise RuntimeError("do a send() before you do a start()")
resp = await self.send_task
self.send_task = None
self.resp = resp
self._acting_as = self.resp
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:
raise e
return ret
def close(self):
if self.resp is not None:
self.resp.close()
if self.send_task is not None:
self.send_task.cancel()
async def delayed_send(self, *args, **kwargs):
req = self.req
if self.delay and self.delay > 0:
# sync_sleep(self.delay)
await asyncio.sleep(self.delay)
t = req.loop.time()
print("sending at {}".format(t), flush=True)
next(iter(args)) # first arg is connection
try:
return await self.orig_send(*args, **kwargs)
except Exception as e:
if aiohttp_version < version.parse("3.1.0"):
return aiohttp.ClientResponse(req.method, req.url)
kw = dict(
writer=None,
continue100=None,
timer=None,
request_info=None,
traces=[],
loop=req.loop,
session=None,
parsed_url = httpcore.URL(url)
request = httpcore.Request(
method, parsed_url, headers=headers, body=body
)
if aiohttp_version < version.parse("3.3.0"):
kw["auto_decompress"] = None
return aiohttp.ClientResponse(req.method, req.url, **kw)
def _send(self, *args, **kwargs):
gen = self.delayed_send(*args, **kwargs)
task = self.req.loop.create_task(gen)
self.send_task = task
self._acting_as = task
return self
if aiohttp_version >= version.parse("3.1.0"):
# aiohttp changed the request.send method to async
async def send(self, *args, **kwargs):
return self._send(*args, **kwargs)
else:
send = _send
def __init__(self, *args, **kwargs):
_post_connect_delay = kwargs.pop("post_connect_delay", 0)
_pre_request_delay = kwargs.pop("pre_request_delay", 0)
super(DelayableTCPConnector, self).__init__(*args, **kwargs)
self._post_connect_delay = _post_connect_delay
self._pre_request_delay = _pre_request_delay
async def connect(self, req, *args, **kwargs):
d_req = DelayableTCPConnector.RequestContextManager(
req, self._pre_request_delay
connection = await self.acquire_connection(
parsed_url, ssl=ssl, timeout=timeout
)
conn = await super(DelayableTCPConnector, self).connect(
req, *args, **kwargs
)
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
if self._request_delay:
print(f"\t>> Sleeping ({self._request_delay})")
await asyncio.sleep(self._request_delay)
response = await connection.send(request)
if not stream:
try:
await response.read()
finally:
await response.close()
return response
class DelayableSanicAdapter(requests.adapters.HTTPAdapter):
def __init__(self, request_delay=None):
self.pool = DelayableSanicConnectionPool(request_delay=request_delay)
class DelayableSanicSession(requests.Session):
def __init__(self, request_delay=None, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
adapter = DelayableSanicAdapter(request_delay=request_delay)
self.mount("http://", adapter)
self.mount("https://", adapter)
class DelayableSanicTestClient(SanicTestClient):
def __init__(self, app, loop, request_delay=1):
super(DelayableSanicTestClient, self).__init__(app)
def __init__(self, app, request_delay=None):
super().__init__(app)
self._request_delay = request_delay
self._loop = None
async def _local_request(self, method, uri, cookies=None, *args, **kwargs):
if self._loop is None:
self._loop = asyncio.get_event_loop()
if uri.startswith(("http:", "https:", "ftp:", "ftps://" "//")):
url = uri
else:
url = "http://{host}:{port}{uri}".format(
host=HOST, port=self.port, uri=uri
)
conn = DelayableTCPConnector(
pre_request_delay=self._request_delay, ssl=False, loop=self._loop
)
async with aiohttp.ClientSession(
cookies=cookies, connector=conn, loop=self._loop
) as session:
# Insert a delay after creating the connection
# But before sending the request.
async with getattr(session, method.lower())(
url, *args, **kwargs
) as response:
try:
response.text = await response.text()
except UnicodeDecodeError:
response.text = None
try:
response.json = await response.json()
except (
JSONDecodeError,
UnicodeDecodeError,
aiohttp.ClientResponseError,
):
response.json = None
response.body = await response.read()
return response
def get_new_session(self):
return DelayableSanicSession(request_delay=self._request_delay)
request_timeout_default_app = Sanic("test_request_timeout_default")
@ -202,14 +92,14 @@ async def ws_handler1(request, ws):
def test_default_server_error_request_timeout():
client = DelayableSanicTestClient(request_timeout_default_app, None, 2)
client = DelayableSanicTestClient(request_timeout_default_app, 2)
request, response = client.get("/1")
assert response.status == 408
assert response.text == "Error: Request Timeout"
def test_default_server_error_request_dont_timeout():
client = DelayableSanicTestClient(request_no_timeout_app, None, 0.2)
client = DelayableSanicTestClient(request_no_timeout_app, 0.2)
request, response = client.get("/1")
assert response.status == 200
assert response.text == "OK"
@ -224,7 +114,7 @@ def test_default_server_error_websocket_request_timeout():
"Sec-WebSocket-Version": "13",
}
client = DelayableSanicTestClient(request_timeout_default_app, None, 2)
client = DelayableSanicTestClient(request_timeout_default_app, 2)
request, response = client.get("/ws1", headers=headers)
assert response.status == 408

87
tests/test_requests.py
View File

@ -1,19 +1,20 @@
import logging
import os
import ssl
from json import dumps as json_dumps
from json import loads as json_loads
from urllib.parse import urlparse
import pytest
from sanic import Sanic
from sanic import Blueprint
from sanic import Blueprint, Sanic
from sanic.exceptions import ServerError
from sanic.request import DEFAULT_HTTP_CONTENT_TYPE, RequestParameters
from sanic.response import json, text
from sanic.testing import HOST, PORT
# ------------------------------------------------------------ #
# GET
# ------------------------------------------------------------ #
@ -529,36 +530,54 @@ def test_request_string_representation(app):
@pytest.mark.parametrize(
"payload,filename",
[
("------sanic\r\n"
(
"------sanic\r\n"
'Content-Disposition: form-data; filename="filename"; name="test"\r\n'
"\r\n"
"OK\r\n"
"------sanic--\r\n", "filename"),
("------sanic\r\n"
"------sanic--\r\n",
"filename",
),
(
"------sanic\r\n"
'content-disposition: form-data; filename="filename"; name="test"\r\n'
"\r\n"
'content-type: application/json; {"field": "value"}\r\n'
"------sanic--\r\n", "filename"),
("------sanic\r\n"
"------sanic--\r\n",
"filename",
),
(
"------sanic\r\n"
'Content-Disposition: form-data; filename=""; name="test"\r\n'
"\r\n"
"OK\r\n"
"------sanic--\r\n", ""),
("------sanic\r\n"
"------sanic--\r\n",
"",
),
(
"------sanic\r\n"
'content-disposition: form-data; filename=""; name="test"\r\n'
"\r\n"
'content-type: application/json; {"field": "value"}\r\n'
"------sanic--\r\n", ""),
("------sanic\r\n"
"------sanic--\r\n",
"",
),
(
"------sanic\r\n"
'Content-Disposition: form-data; filename*="utf-8\'\'filename_%C2%A0_test"; name="test"\r\n'
"\r\n"
"OK\r\n"
"------sanic--\r\n", "filename_\u00A0_test"),
("------sanic\r\n"
"------sanic--\r\n",
"filename_\u00A0_test",
),
(
"------sanic\r\n"
'content-disposition: form-data; filename*="utf-8\'\'filename_%C2%A0_test"; name="test"\r\n'
"\r\n"
'content-type: application/json; {"field": "value"}\r\n'
"------sanic--\r\n", "filename_\u00A0_test"),
"------sanic--\r\n",
"filename_\u00A0_test",
),
],
)
def test_request_multipart_files(app, payload, filename):
@ -743,7 +762,7 @@ def test_request_raw_args(app):
def test_request_query_args(app):
# test multiple params with the same key
params = [('test', 'value1'), ('test', 'value2')]
params = [("test", "value1"), ("test", "value2")]
@app.get("/")
def handler(request):
@ -754,7 +773,10 @@ def test_request_query_args(app):
assert request.query_args == params
# test cached value
assert request.parsed_not_grouped_args[(False, False, "utf-8", "replace")] == request.query_args
assert (
request.parsed_not_grouped_args[(False, False, "utf-8", "replace")]
== request.query_args
)
# test params directly in the url
request, response = app.test_client.get("/?test=value1&test=value2")
@ -762,7 +784,7 @@ def test_request_query_args(app):
assert request.query_args == params
# test unique params
params = [('test1', 'value1'), ('test2', 'value2')]
params = [("test1", "value1"), ("test2", "value2")]
request, response = app.test_client.get("/", params=params)
@ -779,25 +801,22 @@ def test_request_query_args_custom_parsing(app):
def handler(request):
return text("pass")
request, response = app.test_client.get("/?test1=value1&test2=&test3=value3")
request, response = app.test_client.get(
"/?test1=value1&test2=&test3=value3"
)
assert request.get_query_args(
keep_blank_values=True
) == [
('test1', 'value1'), ('test2', ''), ('test3', 'value3')
assert request.get_query_args(keep_blank_values=True) == [
("test1", "value1"),
("test2", ""),
("test3", "value3"),
]
assert request.query_args == [
('test1', 'value1'), ('test3', 'value3')
]
assert request.get_query_args(
keep_blank_values=False
) == [
('test1', 'value1'), ('test3', 'value3')
assert request.query_args == [("test1", "value1"), ("test3", "value3")]
assert request.get_query_args(keep_blank_values=False) == [
("test1", "value1"),
("test3", "value3"),
]
assert request.get_args(
keep_blank_values=True
) == RequestParameters(
assert request.get_args(keep_blank_values=True) == RequestParameters(
{"test1": ["value1"], "test2": [""], "test3": ["value3"]}
)
@ -805,9 +824,7 @@ def test_request_query_args_custom_parsing(app):
{"test1": ["value1"], "test3": ["value3"]}
)
assert request.get_args(
keep_blank_values=False
) == RequestParameters(
assert request.get_args(keep_blank_values=False) == RequestParameters(
{"test1": ["value1"], "test3": ["value3"]}
)

14
tests/test_response.py
View File

@ -1,6 +1,7 @@
import asyncio
import inspect
import os
from collections import namedtuple
from mimetypes import guess_type
from random import choice
@ -8,6 +9,7 @@ from unittest.mock import MagicMock
from urllib.parse import unquote
import pytest
from aiofiles import os as async_os
from sanic.response import (
@ -18,11 +20,11 @@ from sanic.response import (
json,
raw,
stream,
text,
)
from sanic.server import HttpProtocol
from sanic.testing import HOST, PORT
JSON_DATA = {"ok": True}
@ -77,10 +79,10 @@ def test_response_header(app):
request, response = app.test_client.get("/")
assert dict(response.headers) == {
"Connection": "keep-alive",
"Keep-Alive": str(app.config.KEEP_ALIVE_TIMEOUT),
"Content-Length": "11",
"Content-Type": "application/json",
"connection": "keep-alive",
"keep-alive": str(app.config.KEEP_ALIVE_TIMEOUT),
"content-length": "11",
"content-type": "application/json",
}
@ -363,7 +365,7 @@ def test_stream_response_with_cookies(app):
return response
request, response = app.test_client.get("/")
assert response.cookies["test"].value == "pass"
assert response.cookies["test"] == "pass"
def test_stream_response_without_cookies(app):

6
tests/test_response_timeout.py
View File

@ -1,7 +1,9 @@
from sanic import Sanic
import asyncio
from sanic.response import text
from sanic import Sanic
from sanic.exceptions import ServiceUnavailable
from sanic.response import text
response_timeout_app = Sanic("test_response_timeout")
response_timeout_default_app = Sanic("test_response_timeout_default")

83
tests/test_routes.py
View File

@ -7,6 +7,7 @@ from sanic.constants import HTTP_METHODS
from sanic.response import json, text
from sanic.router import ParameterNameConflicts, RouteDoesNotExist, RouteExists
# ------------------------------------------------------------ #
# UTF-8
# ------------------------------------------------------------ #
@ -468,16 +469,8 @@ def test_websocket_route(app, url):
assert ws.subprotocol is None
ev.set()
request, response = app.test_client.get(
"/ws",
headers={
"Upgrade": "websocket",
"Connection": "upgrade",
"Sec-WebSocket-Key": "dGhlIHNhbXBsZSBub25jZQ==",
"Sec-WebSocket-Version": "13",
},
)
assert response.status == 101
request, response = app.test_client.websocket(url)
assert response.opened is True
assert ev.is_set()
@ -487,54 +480,24 @@ def test_websocket_route_with_subprotocols(app):
@app.websocket("/ws", subprotocols=["foo", "bar"])
async def handler(request, ws):
results.append(ws.subprotocol)
assert ws.subprotocol is not None
request, response = app.test_client.get(
"/ws",
headers={
"Upgrade": "websocket",
"Connection": "upgrade",
"Sec-WebSocket-Key": "dGhlIHNhbXBsZSBub25jZQ==",
"Sec-WebSocket-Version": "13",
"Sec-WebSocket-Protocol": "bar",
},
request, response = app.test_client.websocket("/ws", subprotocols=["bar"])
assert response.opened is True
assert results == ["bar"]
request, response = app.test_client.websocket(
"/ws", subprotocols=["bar", "foo"]
)
assert response.status == 101
assert response.opened is True
assert results == ["bar", "bar"]
request, response = app.test_client.get(
"/ws",
headers={
"Upgrade": "websocket",
"Connection": "upgrade",
"Sec-WebSocket-Key": "dGhlIHNhbXBsZSBub25jZQ==",
"Sec-WebSocket-Version": "13",
"Sec-WebSocket-Protocol": "bar, foo",
},
)
assert response.status == 101
request, response = app.test_client.get(
"/ws",
headers={
"Upgrade": "websocket",
"Connection": "upgrade",
"Sec-WebSocket-Key": "dGhlIHNhbXBsZSBub25jZQ==",
"Sec-WebSocket-Version": "13",
"Sec-WebSocket-Protocol": "baz",
},
)
assert response.status == 101
request, response = app.test_client.get(
"/ws",
headers={
"Upgrade": "websocket",
"Connection": "upgrade",
"Sec-WebSocket-Key": "dGhlIHNhbXBsZSBub25jZQ==",
"Sec-WebSocket-Version": "13",
},
)
assert response.status == 101
request, response = app.test_client.websocket("/ws", subprotocols=["baz"])
assert response.opened is True
assert results == ["bar", "bar", None]
request, response = app.test_client.websocket("/ws")
assert response.opened is True
assert results == ["bar", "bar", None, None]
@ -547,16 +510,8 @@ def test_add_webscoket_route(app, strict_slashes):
ev.set()
app.add_websocket_route(handler, "/ws", strict_slashes=strict_slashes)
request, response = app.test_client.get(
"/ws",
headers={
"Upgrade": "websocket",
"Connection": "upgrade",
"Sec-WebSocket-Key": "dGhlIHNhbXBsZSBub25jZQ==",
"Sec-WebSocket-Version": "13",
},
)
assert response.status == 101
request, response = app.test_client.websocket("/ws")
assert response.opened is True
assert ev.is_set()

1
tests/test_server_events.py
View File

@ -4,6 +4,7 @@ import pytest
from sanic.testing import HOST, PORT
AVAILABLE_LISTENERS = [
"before_server_start",
"after_server_start",

8
tests/test_signal_handlers.py
View File

@ -1,8 +1,10 @@
import asyncio
from queue import Queue
from unittest.mock import MagicMock
from sanic.response import HTTPResponse
from sanic.testing import HOST, PORT
from unittest.mock import MagicMock
import asyncio
from queue import Queue
async def stop(app, loop):

1
tests/test_static.py
View File

@ -1,5 +1,6 @@
import inspect
import os
from time import gmtime, strftime
import pytest

19
tests/test_test_client_port.py
View File

@ -1,7 +1,8 @@
import socket
from sanic.testing import PORT, SanicTestClient
from sanic.response import json, text
from sanic.testing import PORT, SanicTestClient
# ------------------------------------------------------------ #
# UTF-8
@ -9,26 +10,26 @@ from sanic.response import json, text
def test_test_client_port_none(app):
@app.get('/get')
@app.get("/get")
def handler(request):
return text('OK')
return text("OK")
test_client = SanicTestClient(app, port=None)
request, response = test_client.get('/get')
assert response.text == 'OK'
request, response = test_client.get("/get")
assert response.text == "OK"
request, response = test_client.post('/get')
request, response = test_client.post("/get")
assert response.status == 405
def test_test_client_port_default(app):
@app.get('/get')
@app.get("/get")
def handler(request):
return json(request.transport.get_extra_info('sockname')[1])
return json(request.transport.get_extra_info("sockname")[1])
test_client = SanicTestClient(app)
assert test_client.port == PORT
request, response = test_client.get('/get')
request, response = test_client.get("/get")
assert response.json == PORT

23
tests/test_url_building.py
View File

@ -1,14 +1,17 @@
import pytest as pytest
from urllib.parse import urlsplit, parse_qsl
from sanic.response import text
from sanic.views import HTTPMethodView
from sanic.blueprints import Blueprint
from sanic.testing import PORT as test_port, HOST as test_host
from sanic.exceptions import URLBuildError
import string
from urllib.parse import parse_qsl, urlsplit
import pytest as pytest
from sanic.blueprints import Blueprint
from sanic.exceptions import URLBuildError
from sanic.response import text
from sanic.testing import HOST as test_host
from sanic.testing import PORT as test_port
from sanic.views import HTTPMethodView
URL_FOR_ARGS1 = dict(arg1=["v1", "v2"])
URL_FOR_VALUE1 = "/myurl?arg1=v1&arg1=v2"
URL_FOR_ARGS2 = dict(arg1=["v1", "v2"], _anchor="anchor")
@ -170,7 +173,7 @@ def test_fails_with_int_message(app):
expected_error = (
r'Value "not_int" for parameter `foo` '
r'does not match pattern for type `int`: -?\d+'
r"does not match pattern for type `int`: -?\d+"
)
assert str(e.value) == expected_error

1
tests/test_utf8.py
View File

@ -1,4 +1,5 @@
from json import dumps as json_dumps
from sanic.response import text

8
tests/test_views.py
View File

@ -1,11 +1,11 @@
import pytest as pytest
from sanic.exceptions import InvalidUsage
from sanic.response import text, HTTPResponse
from sanic.views import HTTPMethodView, CompositionView
from sanic.blueprints import Blueprint
from sanic.request import Request
from sanic.constants import HTTP_METHODS
from sanic.exceptions import InvalidUsage
from sanic.request import Request
from sanic.response import HTTPResponse, text
from sanic.views import CompositionView, HTTPMethodView
@pytest.mark.parametrize("method", HTTP_METHODS)

11
tests/test_worker.py
View File

@ -1,14 +1,17 @@
import time
import asyncio
import json
import shlex
import subprocess
import time
import urllib.request
from unittest import mock
from sanic.worker import GunicornWorker
from sanic.app import Sanic
import asyncio
import pytest
from sanic.app import Sanic
from sanic.worker import GunicornWorker
@pytest.fixture(scope="module")
def gunicorn_worker():

13
tox.ini
View File

@ -1,24 +1,25 @@
[tox]
envlist = py35, py36, py37, {py35,py36,py37}-no-ext, lint, check
envlist = py36, py37, {py36,py37}-no-ext, lint, check
[testenv]
usedevelop = True
setenv =
{py35,py36,py37}-no-ext: SANIC_NO_UJSON=1
{py35,py36,py37}-no-ext: SANIC_NO_UVLOOP=1
{py36,py37}-no-ext: SANIC_NO_UJSON=1
{py36,py37}-no-ext: SANIC_NO_UVLOOP=1
deps =
coverage
pytest==4.1.0
pytest-cov
pytest-sanic
pytest-sugar
aiohttp>=2.3,<=3.2.1
httpcore==0.1.1
requests-async==0.4.0
chardet<=2.3.0
beautifulsoup4
gunicorn
pytest-benchmark
commands =
pytest tests --cov sanic --cov-report= {posargs}
pytest {posargs:tests --cov sanic}
- coverage combine --append
coverage report -m
coverage html -i
@ -31,7 +32,7 @@ deps =
commands =
flake8 sanic
black --check --verbose sanic
black --config ./.black.toml --check --verbose sanic
isort --check-only --recursive sanic
[testenv:check]