refactor uvloop detection in its own method
This commit is contained in:
parent
1af16836d4
commit
f8f0241c27
|
@ -9,6 +9,14 @@ from sanic.exceptions import SanicException
|
|||
from sanic.response import text
|
||||
|
||||
|
||||
def uvloop_installed():
|
||||
try:
|
||||
import uvloop
|
||||
return True
|
||||
except ImportError:
|
||||
return False
|
||||
|
||||
|
||||
def test_app_loop_running(app):
|
||||
@app.get("/test")
|
||||
async def handler(request):
|
||||
|
@ -19,37 +27,28 @@ def test_app_loop_running(app):
|
|||
assert response.text == "pass"
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.version_info < (3, 7),
|
||||
@pytest.mark.skipif(sys.version_info < (3, 7) and uvloop_installed(),
|
||||
reason="requires python3.7 or higher")
|
||||
def test_create_asyncio_server(app):
|
||||
loop = asyncio.get_event_loop()
|
||||
asyncio_srv_coro = app.create_server(
|
||||
return_asyncio_server=True)
|
||||
assert isawaitable(asyncio_srv_coro)
|
||||
|
||||
try:
|
||||
import uvloop
|
||||
except ImportError:
|
||||
# this test is valid only for sanic + asyncio setup
|
||||
srv = loop.run_until_complete(asyncio_srv_coro)
|
||||
assert srv.is_serving() is True
|
||||
srv = loop.run_until_complete(asyncio_srv_coro)
|
||||
assert srv.is_serving() is True
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.version_info < (3, 7),
|
||||
@pytest.mark.skipif(sys.version_info < (3, 7) and uvloop_installed(),
|
||||
reason="requires python3.7 or higher")
|
||||
def test_asyncio_server_start_serving(app):
|
||||
try:
|
||||
import uvloop
|
||||
except ImportError:
|
||||
# this test is valid only for sanic + asyncio setup
|
||||
loop = asyncio.get_event_loop()
|
||||
asyncio_srv_coro = app.create_server(
|
||||
return_asyncio_server=True,
|
||||
asyncio_server_kwargs=dict(
|
||||
start_serving=False
|
||||
))
|
||||
srv = loop.run_until_complete(asyncio_srv_coro)
|
||||
assert srv.is_serving() is False
|
||||
loop = asyncio.get_event_loop()
|
||||
asyncio_srv_coro = app.create_server(
|
||||
return_asyncio_server=True,
|
||||
asyncio_server_kwargs=dict(
|
||||
start_serving=False
|
||||
))
|
||||
srv = loop.run_until_complete(asyncio_srv_coro)
|
||||
assert srv.is_serving() is False
|
||||
|
||||
|
||||
def test_app_loop_not_running(app):
|
||||
|
|
Loading…
Reference in New Issue
Block a user