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

Cleanup ports on tests (#2338)

Browse Source
This commit is contained in:
Adam Hopkins 2021-12-13 19:48:30 +02:00 committed by GitHub
parent 377c2ada38
commit f282865362
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 19 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
tests/test_keep_alive_timeout.py
View File

@ -2,6 +2,7 @@ import asyncio
import platform import platform
from asyncio import sleep as aio_sleep from asyncio import sleep as aio_sleep
from itertools import count
from os import environ from os import environ
import pytest import pytest
@ -15,7 +16,12 @@ from sanic.response import text
CONFIG_FOR_TESTS = {"KEEP_ALIVE_TIMEOUT": 2, "KEEP_ALIVE": True} CONFIG_FOR_TESTS = {"KEEP_ALIVE_TIMEOUT": 2, "KEEP_ALIVE": True}
PORT = 42101 # test_keep_alive_timeout_reuse doesn't work with random port PORT = 42001 # test_keep_alive_timeout_reuse doesn't work with random port
port_counter = count()
def get_port():
return next(port_counter) + PORT
keep_alive_timeout_app_reuse = Sanic("test_ka_timeout_reuse") keep_alive_timeout_app_reuse = Sanic("test_ka_timeout_reuse")
@ -63,9 +69,10 @@ def test_keep_alive_timeout_reuse():
"""If the server keep-alive timeout and client keep-alive timeout are """If the server keep-alive timeout and client keep-alive timeout are
both longer than the delay, the client _and_ server will successfully both longer than the delay, the client _and_ server will successfully
reuse the existing connection.""" reuse the existing connection."""
port = get_port()
loop = asyncio.new_event_loop() loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)
client = ReusableClient(keep_alive_timeout_app_reuse, loop=loop, port=PORT) client = ReusableClient(keep_alive_timeout_app_reuse, loop=loop, port=port)
with client: with client:
headers = {"Connection": "keep-alive"} headers = {"Connection": "keep-alive"}
request, response = client.get("/1", headers=headers) request, response = client.get("/1", headers=headers)
@ -90,10 +97,11 @@ def test_keep_alive_timeout_reuse():
def test_keep_alive_client_timeout(): def test_keep_alive_client_timeout():
"""If the server keep-alive timeout is longer than the client """If the server keep-alive timeout is longer than the client
keep-alive timeout, client will try to create a new connection here.""" keep-alive timeout, client will try to create a new connection here."""
port = get_port()
loop = asyncio.new_event_loop() loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)
client = ReusableClient( client = ReusableClient(
keep_alive_app_client_timeout, loop=loop, port=PORT keep_alive_app_client_timeout, loop=loop, port=port
) )
with client: with client:
headers = {"Connection": "keep-alive"} headers = {"Connection": "keep-alive"}
@ -117,10 +125,11 @@ def test_keep_alive_server_timeout():
keep-alive timeout, the client will either a 'Connection reset' error keep-alive timeout, the client will either a 'Connection reset' error
_or_ a new connection. Depending on how the event-loop handles the _or_ a new connection. Depending on how the event-loop handles the
broken server connection.""" broken server connection."""
port = get_port()
loop = asyncio.new_event_loop() loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)
client = ReusableClient( client = ReusableClient(
keep_alive_app_server_timeout, loop=loop, port=PORT keep_alive_app_server_timeout, loop=loop, port=port
) )
with client: with client:
headers = {"Connection": "keep-alive"} headers = {"Connection": "keep-alive"}
@ -141,9 +150,10 @@ def test_keep_alive_server_timeout():
reason="Not testable with current client", reason="Not testable with current client",
) )
def test_keep_alive_connection_context(): def test_keep_alive_connection_context():
port = get_port()
loop = asyncio.new_event_loop() loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)
client = ReusableClient(keep_alive_app_context, loop=loop, port=PORT) client = ReusableClient(keep_alive_app_context, loop=loop, port=port)
with client: with client:
headers = {"Connection": "keep-alive"} headers = {"Connection": "keep-alive"}
request1, _ = client.post("/ctx", headers=headers) request1, _ = client.post("/ctx", headers=headers)

8
tests/test_reloader.py
View File

@ -117,8 +117,8 @@ argv = dict(
@pytest.mark.parametrize( @pytest.mark.parametrize(
"runargs, mode", "runargs, mode",
[ [
(dict(port=42102, auto_reload=True), "script"), (dict(port=42202, auto_reload=True), "script"),
(dict(port=42103, debug=True), "module"), (dict(port=42203, debug=True), "module"),
({}, "sanic"), ({}, "sanic"),
], ],
) )
@ -149,8 +149,8 @@ async def test_reloader_live(runargs, mode):
@pytest.mark.parametrize( @pytest.mark.parametrize(
"runargs, mode", "runargs, mode",
[ [
(dict(port=42102, auto_reload=True), "script"), (dict(port=42302, auto_reload=True), "script"),
(dict(port=42103, debug=True), "module"), (dict(port=42303, debug=True), "module"),
({}, "sanic"), ({}, "sanic"),
], ],
) )