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

Remove test client (#2009)

Browse Source 
* Initial

* remove testmanager

* Resolve tests
This commit is contained in:
Adam Hopkins 2021-01-28 09:22:22 +02:00 committed by GitHub
parent 976a4c764d
commit 5545264cea
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
20 changed files with 72 additions and 340 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
sanic/app.py
View File

@ -34,7 +34,6 @@ from sanic.server import (
serve_multiple,
)
from sanic.static import register as static_register
from sanic.testing import SanicASGITestClient, SanicTestClient
from sanic.views import CompositionView
from sanic.websocket import ConnectionClosed, WebSocketProtocol
@ -87,6 +86,8 @@ class Sanic:
self.websocket_tasks: Set[Future] = set()
self.named_request_middleware: Dict[str, MiddlewareType] = {}
self.named_response_middleware: Dict[str, MiddlewareType] = {}
self._test_client = None
self._asgi_client = None
# Register alternative method names
self.go_fast = self.run
@ -1032,11 +1033,21 @@ class Sanic:
@property
def test_client(self):
return SanicTestClient(self)
if self._test_client:
return self._test_client
from sanic_testing.testing import SanicTestClient # type: ignore
self._test_client = SanicTestClient(self)
return self._test_client
@property
def asgi_client(self):
return SanicASGITestClient(self)
if self._asgi_client:
return self._asgi_client
from sanic_testing.testing import SanicASGITestClient # type: ignore
self._asgi_client = SanicASGITestClient(self)
return self._asgi_client
# -------------------------------------------------------------------- #
# Execution

284
sanic/testing.py
View File

@ -1,284 +0,0 @@
from json import JSONDecodeError
from socket import socket
import httpx
import websockets
from sanic.asgi import ASGIApp
from sanic.exceptions import MethodNotSupported
from sanic.log import logger
from sanic.response import text
ASGI_HOST = "mockserver"
ASGI_PORT = 1234
ASGI_BASE_URL = f"http://{ASGI_HOST}:{ASGI_PORT}"
HOST = "127.0.0.1"
PORT = None
class SanicTestClient:
def __init__(self, app, port=PORT, host=HOST):
"""Use port=None to bind to a random port"""
self.app = app
self.port = port
self.host = host
@app.listener("after_server_start")
def _start_test_mode(sanic, *args, **kwargs):
sanic.test_mode = True
@app.listener("before_server_end")
def _end_test_mode(sanic, *args, **kwargs):
sanic.test_mode = False
def get_new_session(self):
return httpx.AsyncClient(verify=False)
async def _local_request(self, method, url, *args, **kwargs):
logger.info(url)
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:
if method == "request":
args = [url] + list(args)
url = kwargs.pop("http_method", "GET").upper()
response = await getattr(session, method.lower())(
url, *args, **kwargs
)
except httpx.HTTPError as e:
if hasattr(e, "response"):
response = e.response
else:
logger.error(
f"{method.upper()} {url} received no response!",
exc_info=True,
)
return None
response.body = await response.aread()
response.status = response.status_code
response.content_type = response.headers.get("content-type")
# response can be decoded as json after response._content
# is set by response.aread()
try:
response.json = response.json()
except (JSONDecodeError, UnicodeDecodeError):
response.json = None
if raw_cookies:
response.raw_cookies = {}
for cookie in response.cookies.jar:
response.raw_cookies[cookie.name] = cookie
return response
def _sanic_endpoint_test(
self,
method="get",
uri="/",
gather_request=True,
debug=False,
server_kwargs={"auto_reload": False},
host=None,
*request_args,
**request_kwargs,
):
results = [None, None]
exceptions = []
if gather_request:
def _collect_request(request):
if results[0] is None:
results[0] = request
self.app.request_middleware.appendleft(_collect_request)
@self.app.exception(MethodNotSupported)
async def error_handler(request, exception):
if request.method in ["HEAD", "PATCH", "PUT", "DELETE"]:
return text(
"", exception.status_code, headers=exception.headers
)
else:
return self.app.error_handler.default(request, exception)
if self.port:
server_kwargs = dict(
host=host or self.host,
port=self.port,
**server_kwargs,
)
host, port = host or self.host, self.port
else:
sock = socket()
sock.bind((host or self.host, 0))
server_kwargs = dict(sock=sock, **server_kwargs)
host, port = sock.getsockname()
self.port = port
if uri.startswith(
("http:", "https:", "ftp:", "ftps://", "//", "ws:", "wss:")
):
url = uri
else:
uri = uri if uri.startswith("/") else f"/{uri}"
scheme = "ws" if method == "websocket" else "http"
url = f"{scheme}://{host}:{port}{uri}"
# Tests construct URLs using PORT = None, which means random port not
# known until this function is called, so fix that here
url = url.replace(":None/", f":{port}/")
@self.app.listener("after_server_start")
async def _collect_response(sanic, loop):
try:
response = await self._local_request(
method, url, *request_args, **request_kwargs
)
results[-1] = response
except Exception as e:
logger.exception("Exception")
exceptions.append(e)
self.app.stop()
self.app.run(debug=debug, **server_kwargs)
self.app.listeners["after_server_start"].pop()
if exceptions:
raise ValueError(f"Exception during request: {exceptions}")
if gather_request:
try:
request, response = results
return request, response
except BaseException: # noqa
raise ValueError(
f"Request and response object expected, got ({results})"
)
else:
try:
return results[-1]
except BaseException: # noqa
raise ValueError(f"Request object expected, got ({results})")
def request(self, *args, **kwargs):
return self._sanic_endpoint_test("request", *args, **kwargs)
def get(self, *args, **kwargs):
return self._sanic_endpoint_test("get", *args, **kwargs)
def post(self, *args, **kwargs):
return self._sanic_endpoint_test("post", *args, **kwargs)
def put(self, *args, **kwargs):
return self._sanic_endpoint_test("put", *args, **kwargs)
def delete(self, *args, **kwargs):
return self._sanic_endpoint_test("delete", *args, **kwargs)
def patch(self, *args, **kwargs):
return self._sanic_endpoint_test("patch", *args, **kwargs)
def options(self, *args, **kwargs):
return self._sanic_endpoint_test("options", *args, **kwargs)
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)
class TestASGIApp(ASGIApp):
async def __call__(self):
await super().__call__()
return self.request
async def app_call_with_return(self, scope, receive, send):
asgi_app = await TestASGIApp.create(self, scope, receive, send)
return await asgi_app()
class SanicASGITestClient(httpx.AsyncClient):
def __init__(
self,
app,
base_url: str = ASGI_BASE_URL,
suppress_exceptions: bool = False,
) -> None:
app.__class__.__call__ = app_call_with_return
app.asgi = True
self.app = app
transport = httpx.ASGITransport(app=app, client=(ASGI_HOST, ASGI_PORT))
super().__init__(transport=transport, base_url=base_url)
self.last_request = None
def _collect_request(request):
self.last_request = request
@app.listener("after_server_start")
def _start_test_mode(sanic, *args, **kwargs):
sanic.test_mode = True
@app.listener("before_server_end")
def _end_test_mode(sanic, *args, **kwargs):
sanic.test_mode = False
app.request_middleware.appendleft(_collect_request)
async def request(self, method, url, gather_request=True, *args, **kwargs):
self.gather_request = gather_request
response = await super().request(method, url, *args, **kwargs)
response.status = response.status_code
response.body = response.content
response.content_type = response.headers.get("content-type")
return self.last_request, response
async def websocket(self, uri, subprotocols=None, *args, **kwargs):
scheme = "ws"
path = uri
root_path = f"{scheme}://{ASGI_HOST}"
headers = kwargs.get("headers", {})
headers.setdefault("connection", "upgrade")
headers.setdefault("sec-websocket-key", "testserver==")
headers.setdefault("sec-websocket-version", "13")
if subprotocols is not None:
headers.setdefault(
"sec-websocket-protocol", ", ".join(subprotocols)
)
scope = {
"type": "websocket",
"asgi": {"version": "3.0"},
"http_version": "1.1",
"headers": [map(lambda y: y.encode(), x) for x in headers.items()],
"scheme": scheme,
"root_path": root_path,
"path": path,
"query_string": b"",
}
async def receive():
return {}
async def send(message):
pass
await self.app(scope, receive, send)
return None, {}

3
setup.py
View File

@ -89,15 +89,14 @@ requirements = [
"aiofiles>=0.6.0",
"websockets>=8.1,<9.0",
"multidict>=5.0,<6.0",
"httpx==0.15.4",
]
tests_require = [
"sanic-testing",
"pytest==5.2.1",
"multidict>=5.0,<6.0",
"gunicorn==20.0.4",
"pytest-cov",
"httpcore==0.11.*",
"beautifulsoup4",
uvloop,
ujson,

13
tests/conftest.py
View File

@ -6,6 +6,8 @@ import uuid
import pytest
from sanic_testing import TestManager
from sanic import Sanic
from sanic.router import RouteExists, Router
@ -17,6 +19,11 @@ if sys.platform in ["win32", "cygwin"]:
collect_ignore = ["test_worker.py"]
@pytest.fixture
def caplog(caplog):
yield caplog
async def _handler(request):
"""
Dummy placeholder method used for route resolver when creating a new
@ -127,6 +134,8 @@ def url_param_generator():
return TYPE_TO_GENERATOR_MAP
@pytest.fixture
@pytest.fixture(scope="function")
def app(request):
return Sanic(request.node.name)
app = Sanic(request.node.name)
# TestManager(app)
return app

3
tests/test_asgi.py
View File

@ -41,8 +41,7 @@ def transport(message_stack, receive, send):
@pytest.fixture
# @pytest.mark.asyncio
def protocol(transport, loop):
def protocol(transport):
return transport.get_protocol()

5
tests/test_asgi_client.py
View File

@ -1,5 +0,0 @@
from sanic.testing import SanicASGITestClient
def test_asgi_client_instantiation(app):
assert isinstance(app.asgi_client, SanicASGITestClient)

3
tests/test_keep_alive_timeout.py
View File

@ -8,10 +8,11 @@ import httpcore
import httpx
import pytest
from sanic_testing.testing import HOST, SanicTestClient
from sanic import Sanic, server
from sanic.compat import OS_IS_WINDOWS
from sanic.response import text
from sanic.testing import HOST, SanicTestClient
CONFIG_FOR_TESTS = {"KEEP_ALIVE_TIMEOUT": 2, "KEEP_ALIVE": True}

4
tests/test_logging.py
View File

@ -8,13 +8,14 @@ from io import StringIO
import pytest
from sanic_testing.testing import SanicTestClient
import sanic
from sanic import Sanic
from sanic.compat import OS_IS_WINDOWS
from sanic.log import LOGGING_CONFIG_DEFAULTS, logger
from sanic.response import text
from sanic.testing import SanicTestClient
logging_format = """module: %(module)s; \
@ -34,6 +35,7 @@ def test_log(app):
logging.basicConfig(
format=logging_format, level=logging.DEBUG, stream=log_stream
)
logging.getLogger("asyncio").setLevel(logging.WARNING)
log = logging.getLogger()
rand_string = str(uuid.uuid4())

27
tests/test_logo.py
View File

@ -1,16 +1,9 @@
import asyncio
import logging
from sanic_testing.testing import PORT
from sanic.config import BASE_LOGO
from sanic.testing import PORT
try:
import uvloop # noqa
ROW = 0
except BaseException:
ROW = 1
def test_logo_base(app, caplog):
@ -28,8 +21,8 @@ def test_logo_base(app, caplog):
loop.run_until_complete(_server.wait_closed())
app.stop()
assert caplog.record_tuples[ROW][1] == logging.DEBUG
assert caplog.record_tuples[ROW][2] == BASE_LOGO
assert caplog.record_tuples[0][1] == logging.DEBUG
assert caplog.record_tuples[0][2] == BASE_LOGO
def test_logo_false(app, caplog):
@ -49,8 +42,8 @@ def test_logo_false(app, caplog):
loop.run_until_complete(_server.wait_closed())
app.stop()
banner, port = caplog.record_tuples[ROW][2].rsplit(":", 1)
assert caplog.record_tuples[ROW][1] == logging.INFO
banner, port = caplog.record_tuples[0][2].rsplit(":", 1)
assert caplog.record_tuples[0][1] == logging.INFO
assert banner == "Goin' Fast @ http://127.0.0.1"
assert int(port) > 0
@ -72,8 +65,8 @@ def test_logo_true(app, caplog):
loop.run_until_complete(_server.wait_closed())
app.stop()
assert caplog.record_tuples[ROW][1] == logging.DEBUG
assert caplog.record_tuples[ROW][2] == BASE_LOGO
assert caplog.record_tuples[0][1] == logging.DEBUG
assert caplog.record_tuples[0][2] == BASE_LOGO
def test_logo_custom(app, caplog):
@ -93,5 +86,5 @@ def test_logo_custom(app, caplog):
loop.run_until_complete(_server.wait_closed())
app.stop()
assert caplog.record_tuples[ROW][1] == logging.DEBUG
assert caplog.record_tuples[ROW][2] == "My Custom Logo"
assert caplog.record_tuples[0][1] == logging.DEBUG
assert caplog.record_tuples[0][2] == "My Custom Logo"

3
tests/test_multiprocessing.py
View File

@ -5,9 +5,10 @@ import signal
import pytest
from sanic_testing.testing import HOST, PORT
from sanic import Blueprint
from sanic.response import text
from sanic.testing import HOST, PORT
@pytest.mark.skipif(

2
tests/test_request_timeout.py
View File

@ -16,10 +16,10 @@ from httpcore._async.connection_pool import ResponseByteStream
from httpcore._exceptions import LocalProtocolError, UnsupportedProtocol
from httpcore._types import TimeoutDict
from httpcore._utils import url_to_origin
from sanic_testing.testing import SanicTestClient
from sanic import Sanic
from sanic.response import text
from sanic.testing import SanicTestClient
class DelayableHTTPConnection(httpcore._async.connection.AsyncHTTPConnection):

11
tests/test_requests.py
View File

@ -8,11 +8,7 @@ from urllib.parse import urlparse
import pytest
from sanic import Blueprint, Sanic
from sanic.exceptions import ServerError
from sanic.request import DEFAULT_HTTP_CONTENT_TYPE, Request, RequestParameters
from sanic.response import html, json, text
from sanic.testing import (
from sanic_testing.testing import (
ASGI_BASE_URL,
ASGI_HOST,
ASGI_PORT,
@ -21,6 +17,11 @@ from sanic.testing import (
SanicTestClient,
)
from sanic import Blueprint, Sanic
from sanic.exceptions import ServerError
from sanic.request import DEFAULT_HTTP_CONTENT_TYPE, Request, RequestParameters
from sanic.response import html, json, text
# ------------------------------------------------------------ #
# GET

2
tests/test_response.py
View File

@ -12,6 +12,7 @@ from urllib.parse import unquote
import pytest
from aiofiles import os as async_os
from sanic_testing.testing import HOST, PORT
from sanic.response import (
HTTPResponse,
@ -25,7 +26,6 @@ from sanic.response import (
text,
)
from sanic.server import HttpProtocol
from sanic.testing import HOST, PORT
JSON_DATA = {"ok": True}

11
tests/test_routes.py
View File

@ -2,11 +2,12 @@ import asyncio
import pytest
from sanic_testing.testing import SanicTestClient
from sanic import Sanic
from sanic.constants import HTTP_METHODS
from sanic.response import json, text
from sanic.router import ParameterNameConflicts, RouteDoesNotExist, RouteExists
from sanic.testing import SanicTestClient
# ------------------------------------------------------------ #
@ -479,21 +480,21 @@ def test_websocket_route_with_subprotocols(app):
results.append(ws.subprotocol)
assert ws.subprotocol is not None
request, response = app.test_client.websocket("/ws", subprotocols=["bar"])
_, response = SanicTestClient(app).websocket("/ws", subprotocols=["bar"])
assert response.opened is True
assert results == ["bar"]
request, response = app.test_client.websocket(
_, response = SanicTestClient(app).websocket(
"/ws", subprotocols=["bar", "foo"]
)
assert response.opened is True
assert results == ["bar", "bar"]
request, response = app.test_client.websocket("/ws", subprotocols=["baz"])
_, response = SanicTestClient(app).websocket("/ws", subprotocols=["baz"])
assert response.opened is True
assert results == ["bar", "bar", None]
request, response = app.test_client.websocket("/ws")
_, response = SanicTestClient(app).websocket("/ws")
assert response.opened is True
assert results == ["bar", "bar", None, None]

2
tests/test_server_events.py
View File

@ -6,7 +6,7 @@ from socket import socket
import pytest
from sanic.testing import HOST, PORT
from sanic_testing.testing import HOST, PORT
AVAILABLE_LISTENERS = [

3
tests/test_signal_handlers.py
View File

@ -7,9 +7,10 @@ from unittest.mock import MagicMock
import pytest
from sanic_testing.testing import HOST, PORT
from sanic.compat import ctrlc_workaround_for_windows
from sanic.response import HTTPResponse
from sanic.testing import HOST, PORT
async def stop(app, loop):

3
tests/test_test_client_port.py
View File

@ -1,5 +1,6 @@
from sanic_testing.testing import PORT, SanicTestClient
from sanic.response import json, text
from sanic.testing import PORT, SanicTestClient
# ------------------------------------------------------------ #

5
tests/test_url_building.py
View File

@ -4,11 +4,12 @@ from urllib.parse import parse_qsl, urlsplit
import pytest as pytest
from sanic_testing.testing import HOST as test_host
from sanic_testing.testing import PORT as test_port
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

6
tests/test_url_for.py
View File

@ -1,5 +1,7 @@
import asyncio
from sanic_testing.testing import SanicTestClient
from sanic.blueprints import Blueprint
@ -48,14 +50,14 @@ def test_websocket_bp_route_name(app):
uri = app.url_for("test_bp.test_route")
assert uri == "/bp/route"
request, response = app.test_client.websocket(uri)
request, response = SanicTestClient(app).websocket(uri)
assert response.opened is True
assert event.is_set()
event.clear()
uri = app.url_for("test_bp.test_route2")
assert uri == "/bp/route2"
request, response = app.test_client.websocket(uri)
request, response = SanicTestClient(app).websocket(uri)
assert response.opened is True
assert event.is_set()

3
tox.ini
View File

@ -7,14 +7,13 @@ setenv =
{py36,py37,py38,py39,pyNightly}-no-ext: SANIC_NO_UJSON=1
{py36,py37,py38,py39,pyNightly}-no-ext: SANIC_NO_UVLOOP=1
deps =
sanic-testing==0.1.2
coverage==5.3
pytest==5.2.1
pytest-cov
pytest-sanic
pytest-sugar
pytest-benchmark
httpcore==0.11.*
httpx==0.15.4
chardet==3.*
beautifulsoup4
gunicorn==20.0.4