Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Allow sanic test client to bind to a random port #1376

Merged
merged 1 commit into from
Mar 4, 2019
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions docs/sanic/testing.md
Original file line number Diff line number Diff line change
Expand Up @@ -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).


## 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](https://github.com/yunstanford/pytest-sanic) is a pytest plugin, it helps you to test your code asynchronously.
Expand Down
31 changes: 21 additions & 10 deletions sanic/testing.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from sanic.exceptions import MethodNotSupported
from sanic.log import logger
from sanic.response import text
from socket import socket


HOST = "127.0.0.1"
Expand All @@ -11,19 +12,13 @@

class SanicTestClient:
def __init__(self, app, port=PORT):
"""Use port=None to bind to a random port"""
self.app = app
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

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)
conn = aiohttp.TCPConnector(ssl=False)
async with aiohttp.ClientSession(
Expand Down Expand Up @@ -79,19 +74,35 @@ async def error_handler(request, exception):
else:
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")
async def _collect_response(sanic, loop):
try:
response = await self._local_request(
method, uri, *request_args, **request_kwargs
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(host=HOST, debug=debug, port=self.port, **server_kwargs)
self.app.run(debug=debug, **server_kwargs)
self.app.listeners["after_server_start"].pop()

if exceptions:
Expand Down
34 changes: 34 additions & 0 deletions tests/test_test_client_port.py
Original file line number Diff line number Diff line change
@@ -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