Skip to content

Commit

Permalink
Introduce dynamic port fixture for test_app and helper function for t…
Browse files Browse the repository at this point in the history
…ests which require port within a loop.
  • Loading branch information
iAndriy committed Dec 20, 2023
1 parent d0bbcf5 commit 6e5b818
Show file tree
Hide file tree
Showing 2 changed files with 31 additions and 18 deletions.
9 changes: 9 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import asyncio
import socket
import inspect
import logging
import os
Expand Down Expand Up @@ -31,6 +32,14 @@
if sys.platform in ["win32", "cygwin"]:
collect_ignore = ["test_worker.py"]

def get_port():
sock = socket.socket()
sock.bind(('', 0)) # Bind to 0 port, so os will pick available port for us.

Check warning

Code scanning / CodeQL

Binding a socket to all network interfaces Medium test

'' binds a socket to all interfaces.
return sock.getsockname()[1]
@pytest.fixture(scope="function")
def port():
yield get_port()


async def _handler(request):
"""
Expand Down
40 changes: 22 additions & 18 deletions tests/test_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@
from sanic.response import text
from sanic.router import Route

from .conftest import get_port


@pytest.fixture(autouse=True)
def clear_app_registry():
Expand All @@ -37,19 +39,19 @@ async def handler(request):


@pytest.mark.asyncio
def test_create_asyncio_server(app: Sanic):
def test_create_asyncio_server(app: Sanic, port: int):
loop = asyncio.get_event_loop()
asyncio_srv_coro = app.create_server(return_asyncio_server=True)
asyncio_srv_coro = app.create_server(return_asyncio_server=True, port=port)
assert isawaitable(asyncio_srv_coro)
srv = loop.run_until_complete(asyncio_srv_coro)
assert srv.is_serving() is True


@pytest.mark.asyncio
def test_asyncio_server_no_start_serving(app: Sanic):
def test_asyncio_server_no_start_serving(app: Sanic, port):
loop = asyncio.get_event_loop()
asyncio_srv_coro = app.create_server(
port=43123,
port=port,
return_asyncio_server=True,
asyncio_server_kwargs=dict(start_serving=False),
)
Expand All @@ -76,11 +78,11 @@ def test_asyncio_server_start_serving(app: Sanic):


@pytest.mark.asyncio
def test_create_server_main(app: Sanic, caplog):
def test_create_server_main(app: Sanic, caplog, port):
app.listener("main_process_start")(lambda *_: ...)
loop = asyncio.get_event_loop()
with caplog.at_level(logging.INFO):
asyncio_srv_coro = app.create_server(return_asyncio_server=True)
asyncio_srv_coro = app.create_server(return_asyncio_server=True, port=port)
loop.run_until_complete(asyncio_srv_coro)
assert (
"sanic.root",
Expand All @@ -91,10 +93,10 @@ def test_create_server_main(app: Sanic, caplog):


@pytest.mark.asyncio
def test_create_server_no_startup(app: Sanic):
def test_create_server_no_startup(app: Sanic, port):
loop = asyncio.get_event_loop()
asyncio_srv_coro = app.create_server(
port=43124,
port=port,
return_asyncio_server=True,
asyncio_server_kwargs=dict(start_serving=False),
)
Expand All @@ -107,11 +109,11 @@ def test_create_server_no_startup(app: Sanic):


@pytest.mark.asyncio
def test_create_server_main_convenience(app: Sanic, caplog):
def test_create_server_main_convenience(app: Sanic, caplog, port):
app.main_process_start(lambda *_: ...)
loop = asyncio.get_event_loop()
with caplog.at_level(logging.INFO):
asyncio_srv_coro = app.create_server(return_asyncio_server=True)
asyncio_srv_coro = app.create_server(return_asyncio_server=True, port=port)
loop.run_until_complete(asyncio_srv_coro)
assert (
"sanic.root",
Expand All @@ -131,9 +133,9 @@ def test_app_loop_not_running(app: Sanic):
)


def test_app_run_raise_type_error(app: Sanic):
def test_app_run_raise_type_error(app: Sanic, port):
with pytest.raises(TypeError) as excinfo:
app.run(loop="loop")
app.run(loop="loop", port=port)

assert str(excinfo.value) == (
"loop is not a valid argument. To use an existing loop, "
Expand Down Expand Up @@ -489,6 +491,7 @@ def test_uvloop_cannot_never_called_with_create_server(caplog, monkeypatch):
srv_coro = app.create_server(
return_asyncio_server=True,
asyncio_server_kwargs=dict(start_serving=False),
port=get_port()
)
loop.run_until_complete(srv_coro)

Expand Down Expand Up @@ -527,6 +530,7 @@ def test_multiple_uvloop_configs_display_warning(caplog):
srv_coro = app.create_server(
return_asyncio_server=True,
asyncio_server_kwargs=dict(start_serving=False),
port=get_port()
)
srv = loop.run_until_complete(srv_coro)
loop.run_until_complete(srv.startup())
Expand All @@ -542,15 +546,15 @@ def test_multiple_uvloop_configs_display_warning(caplog):
assert counter[(logging.WARNING, message)] == 3


def test_cannot_run_fast_and_workers(app: Sanic):
def test_cannot_run_fast_and_workers(app: Sanic, port):
message = "You cannot use both fast=True and workers=X"
with pytest.raises(RuntimeError, match=message):
app.run(fast=True, workers=4)
app.run(fast=True, workers=4, port=port)


def test_no_workers(app: Sanic):
def test_no_workers(app: Sanic, port):
with pytest.raises(RuntimeError, match="Cannot serve with no workers"):
app.run(workers=0)
app.run(workers=0, port=port)


@pytest.mark.parametrize(
Expand All @@ -562,13 +566,13 @@ def test_no_workers(app: Sanic):
),
)
def test_cannot_run_single_process_and_workers_or_auto_reload(
app: Sanic, extra
app: Sanic, extra, port
):
message = (
"Single process cannot be run with multiple workers or auto-reload"
)
with pytest.raises(RuntimeError, match=message):
app.run(single_process=True, **extra)
app.run(single_process=True, port=port, **extra)


def test_default_configure_logging():
Expand Down

0 comments on commit 6e5b818

Please sign in to comment.