Allow sanic test client to bind to a random port (#1376)
This commit is contained in:
parent
348964fe12
commit
d5813152ab
|
@ -59,6 +59,23 @@ the available arguments to aiohttp can be found
|
||||||
[in the documentation for ClientSession](https://aiohttp.readthedocs.io/en/stable/client_reference.html#client-session).
|
[in the documentation for ClientSession](https://aiohttp.readthedocs.io/en/stable/client_reference.html#client-session).
|
||||||
|
|
||||||
|
|
||||||
|
## Using a random port
|
||||||
|
|
||||||
|
If you need to test using a free unpriveleged port chosen by the kernel
|
||||||
|
instead of the default with `SanicTestClient`, you can do so by specifying
|
||||||
|
`port=None`. On most systems the port will be in the range 1024 to 65535.
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Import the Sanic app, usually created with Sanic(__name__)
|
||||||
|
from external_server import app
|
||||||
|
from sanic.testing import SanicTestClient
|
||||||
|
|
||||||
|
def test_index_returns_200():
|
||||||
|
request, response = SanicTestClient(app, port=None).get('/')
|
||||||
|
assert response.status == 200
|
||||||
|
```
|
||||||
|
|
||||||
|
|
||||||
## pytest-sanic
|
## pytest-sanic
|
||||||
|
|
||||||
[pytest-sanic](https://github.com/yunstanford/pytest-sanic) is a pytest plugin, it helps you to test your code asynchronously.
|
[pytest-sanic](https://github.com/yunstanford/pytest-sanic) is a pytest plugin, it helps you to test your code asynchronously.
|
||||||
|
|
|
@ -3,6 +3,7 @@ from json import JSONDecodeError
|
||||||
from sanic.exceptions import MethodNotSupported
|
from sanic.exceptions import MethodNotSupported
|
||||||
from sanic.log import logger
|
from sanic.log import logger
|
||||||
from sanic.response import text
|
from sanic.response import text
|
||||||
|
from socket import socket
|
||||||
|
|
||||||
|
|
||||||
HOST = "127.0.0.1"
|
HOST = "127.0.0.1"
|
||||||
|
@ -11,19 +12,13 @@ PORT = 42101
|
||||||
|
|
||||||
class SanicTestClient:
|
class SanicTestClient:
|
||||||
def __init__(self, app, port=PORT):
|
def __init__(self, app, port=PORT):
|
||||||
|
"""Use port=None to bind to a random port"""
|
||||||
self.app = app
|
self.app = app
|
||||||
self.port = port
|
self.port = port
|
||||||
|
|
||||||
async def _local_request(self, method, uri, cookies=None, *args, **kwargs):
|
async def _local_request(self, method, url, cookies=None, *args, **kwargs):
|
||||||
import aiohttp
|
import aiohttp
|
||||||
|
|
||||||
if uri.startswith(("http:", "https:", "ftp:", "ftps://" "//")):
|
|
||||||
url = uri
|
|
||||||
else:
|
|
||||||
url = "http://{host}:{port}{uri}".format(
|
|
||||||
host=HOST, port=self.port, uri=uri
|
|
||||||
)
|
|
||||||
|
|
||||||
logger.info(url)
|
logger.info(url)
|
||||||
conn = aiohttp.TCPConnector(ssl=False)
|
conn = aiohttp.TCPConnector(ssl=False)
|
||||||
async with aiohttp.ClientSession(
|
async with aiohttp.ClientSession(
|
||||||
|
@ -79,11 +74,27 @@ class SanicTestClient:
|
||||||
else:
|
else:
|
||||||
return self.app.error_handler.default(request, exception)
|
return self.app.error_handler.default(request, exception)
|
||||||
|
|
||||||
|
if self.port:
|
||||||
|
server_kwargs = dict(host=HOST, port=self.port, **server_kwargs)
|
||||||
|
host, port = HOST, self.port
|
||||||
|
else:
|
||||||
|
sock = socket()
|
||||||
|
sock.bind((HOST, 0))
|
||||||
|
server_kwargs = dict(sock=sock, **server_kwargs)
|
||||||
|
host, port = sock.getsockname()
|
||||||
|
|
||||||
|
if uri.startswith(("http:", "https:", "ftp:", "ftps://", "//")):
|
||||||
|
url = uri
|
||||||
|
else:
|
||||||
|
url = "http://{host}:{port}{uri}".format(
|
||||||
|
host=host, port=port, uri=uri
|
||||||
|
)
|
||||||
|
|
||||||
@self.app.listener("after_server_start")
|
@self.app.listener("after_server_start")
|
||||||
async def _collect_response(sanic, loop):
|
async def _collect_response(sanic, loop):
|
||||||
try:
|
try:
|
||||||
response = await self._local_request(
|
response = await self._local_request(
|
||||||
method, uri, *request_args, **request_kwargs
|
method, url, *request_args, **request_kwargs
|
||||||
)
|
)
|
||||||
results[-1] = response
|
results[-1] = response
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
|
@ -91,7 +102,7 @@ class SanicTestClient:
|
||||||
exceptions.append(e)
|
exceptions.append(e)
|
||||||
self.app.stop()
|
self.app.stop()
|
||||||
|
|
||||||
self.app.run(host=HOST, debug=debug, port=self.port, **server_kwargs)
|
self.app.run(debug=debug, **server_kwargs)
|
||||||
self.app.listeners["after_server_start"].pop()
|
self.app.listeners["after_server_start"].pop()
|
||||||
|
|
||||||
if exceptions:
|
if exceptions:
|
||||||
|
|
34
tests/test_test_client_port.py
Normal file
34
tests/test_test_client_port.py
Normal file
|
@ -0,0 +1,34 @@
|
||||||
|
import socket
|
||||||
|
|
||||||
|
from sanic.testing import PORT, SanicTestClient
|
||||||
|
from sanic.response import json, text
|
||||||
|
|
||||||
|
# ------------------------------------------------------------ #
|
||||||
|
# UTF-8
|
||||||
|
# ------------------------------------------------------------ #
|
||||||
|
|
||||||
|
|
||||||
|
def test_test_client_port_none(app):
|
||||||
|
@app.get('/get')
|
||||||
|
def handler(request):
|
||||||
|
return text('OK')
|
||||||
|
|
||||||
|
test_client = SanicTestClient(app, port=None)
|
||||||
|
|
||||||
|
request, response = test_client.get('/get')
|
||||||
|
assert response.text == 'OK'
|
||||||
|
|
||||||
|
request, response = test_client.post('/get')
|
||||||
|
assert response.status == 405
|
||||||
|
|
||||||
|
|
||||||
|
def test_test_client_port_default(app):
|
||||||
|
@app.get('/get')
|
||||||
|
def handler(request):
|
||||||
|
return json(request.transport.get_extra_info('sockname')[1])
|
||||||
|
|
||||||
|
test_client = SanicTestClient(app)
|
||||||
|
assert test_client.port == PORT
|
||||||
|
|
||||||
|
request, response = test_client.get('/get')
|
||||||
|
assert response.json == PORT
|
Loading…
Reference in New Issue
Block a user