Adding windows to CI

This commit is contained in:
Almar Klein 2018-10-16 16:13:02 +02:00
parent e5062880f3
commit 5ccb70df1f
7 changed files with 149 additions and 72 deletions

View File

@ -1,14 +1,38 @@
language: python
cache: pip
python:
- "3.5"
- "3.6"
- "3.7-dev"
language: python
matrix:
include:
# The Windows job is important (obviously) to test on a non-Unix platform.
# It also runs the test with a quite minimal setup (no uvloop or httptools),
# but it does have websockets (needed to run the websockets tests).
- os: windows
language: bash
# The pypy3 job runs a fully pure Python setup, which triggers a part of the
# auto-detect tests that the other tests don't. We cannot test websockets though.
- os: linux
python: "pypy3.5"
# The main tests run on Linux for a all relevant Python versions.
- os: linux
python: "3.5"
- os: linux
python: "3.6"
- os: linux
python: "3.7-dev"
install:
- pip install -U -r requirements.txt
- if [ "$TRAVIS_OS_NAME" = "windows" ]; then
choco install python3;
export PATH=/c/Python37:/c/Python37/Scripts:$PATH;
python -m pip install click h11 wsproto websockets;
python -m pip install autoflake black codecov isort pytest pytest-cov requests;
elif [ "$TRAVIS_PYTHON_VERSION" = "pypy3.5" ]; then
python -m pip install click h11 wsproto;
python -m pip install pytest pytest-cov requests codecov;
else
pip install -U -r requirements.txt;
fi;
script:
- scripts/test

0
tests/__init__.py Normal file
View File

View File

@ -2,14 +2,21 @@ import asyncio
import h11
import pytest
from tests.response import Response
from uvicorn.config import Config
from uvicorn.main import ServerState
from uvicorn.protocols.http.h11_impl import H11Protocol
from uvicorn.protocols.http.httptools_impl import HttpToolsProtocol
from uvicorn.protocols.websockets.wsproto_impl import WSProtocol
from tests.response import Response
try:
from uvicorn.protocols.http.httptools_impl import HttpToolsProtocol
except ImportError:
HttpToolsProtocol = None
HTTP_PROTOCOLS = [p for p in [H11Protocol, HttpToolsProtocol] if p is not None]
SIMPLE_GET_REQUEST = b"\r\n".join([b"GET / HTTP/1.1", b"Host: example.org", b"", b""])
SIMPLE_HEAD_REQUEST = b"\r\n".join([b"HEAD / HTTP/1.1", b"Host: example.org", b"", b""])
@ -144,7 +151,7 @@ def get_connected_protocol(app, protocol_cls, **kwargs):
return protocol
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_get_request(protocol_cls):
def app(scope):
return Response("Hello, world", media_type="text/plain")
@ -156,7 +163,7 @@ def test_get_request(protocol_cls):
assert b"Hello, world" in protocol.transport.buffer
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_head_request(protocol_cls):
def app(scope):
return Response("Hello, world", media_type="text/plain")
@ -168,7 +175,7 @@ def test_head_request(protocol_cls):
assert b"Hello, world" not in protocol.transport.buffer
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_post_request(protocol_cls):
class App:
def __init__(self, scope):
@ -191,7 +198,7 @@ def test_post_request(protocol_cls):
assert b'Body: {"hello": "world"}' in protocol.transport.buffer
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_keepalive(protocol_cls):
def app(scope):
return Response(b"", status_code=204)
@ -204,7 +211,7 @@ def test_keepalive(protocol_cls):
assert not protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_keepalive_timeout(protocol_cls):
def app(scope):
return Response(b"", status_code=204)
@ -223,7 +230,7 @@ def test_keepalive_timeout(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_close(protocol_cls):
def app(scope):
return Response(b"", status_code=204, headers={"connection": "close"})
@ -235,7 +242,7 @@ def test_close(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_chunked_encoding(protocol_cls):
def app(scope):
return Response(
@ -250,7 +257,7 @@ def test_chunked_encoding(protocol_cls):
assert not protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_chunked_encoding_empty_body(protocol_cls):
def app(scope):
return Response(
@ -265,7 +272,7 @@ def test_chunked_encoding_empty_body(protocol_cls):
assert not protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_chunked_encoding_head_request(protocol_cls):
def app(scope):
return Response(
@ -279,7 +286,7 @@ def test_chunked_encoding_head_request(protocol_cls):
assert not protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_pipelined_requests(protocol_cls):
def app(scope):
return Response("Hello, world", media_type="text/plain")
@ -305,7 +312,7 @@ def test_pipelined_requests(protocol_cls):
protocol.transport.clear_buffer()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_undersized_request(protocol_cls):
def app(scope):
return Response(b"xxx", headers={"content-length": "10"})
@ -316,7 +323,7 @@ def test_undersized_request(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_oversized_request(protocol_cls):
def app(scope):
return Response(b"xxx" * 20, headers={"content-length": "10"})
@ -327,7 +334,7 @@ def test_oversized_request(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_large_post_request(protocol_cls):
def app(scope):
return Response("Hello, world", media_type="text/plain")
@ -339,7 +346,7 @@ def test_large_post_request(protocol_cls):
assert not protocol.transport.read_paused
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_invalid_http(protocol_cls):
app = lambda scope: None
protocol = get_connected_protocol(app, protocol_cls)
@ -347,7 +354,7 @@ def test_invalid_http(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_app_exception(protocol_cls):
class App:
def __init__(self, scope):
@ -363,7 +370,7 @@ def test_app_exception(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_app_init_exception(protocol_cls):
def app(scope):
raise Exception()
@ -375,7 +382,7 @@ def test_app_init_exception(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_exception_during_response(protocol_cls):
class App:
def __init__(self, scope):
@ -393,7 +400,7 @@ def test_exception_during_response(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_no_response_returned(protocol_cls):
class App:
def __init__(self, scope):
@ -409,7 +416,7 @@ def test_no_response_returned(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_partial_response_returned(protocol_cls):
class App:
def __init__(self, scope):
@ -425,7 +432,7 @@ def test_partial_response_returned(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_duplicate_start_message(protocol_cls):
class App:
def __init__(self, scope):
@ -442,7 +449,7 @@ def test_duplicate_start_message(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_missing_start_message(protocol_cls):
class App:
def __init__(self, scope):
@ -458,7 +465,7 @@ def test_missing_start_message(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_message_after_body_complete(protocol_cls):
class App:
def __init__(self, scope):
@ -476,7 +483,7 @@ def test_message_after_body_complete(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_value_returned(protocol_cls):
class App:
def __init__(self, scope):
@ -494,7 +501,7 @@ def test_value_returned(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_early_disconnect(protocol_cls):
got_disconnect_event = False
@ -520,7 +527,7 @@ def test_early_disconnect(protocol_cls):
assert got_disconnect_event
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_early_response(protocol_cls):
def app(scope):
return Response("Hello, world", media_type="text/plain")
@ -533,7 +540,7 @@ def test_early_response(protocol_cls):
assert not protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_read_after_response(protocol_cls):
message_after_response = None
@ -555,7 +562,7 @@ def test_read_after_response(protocol_cls):
assert message_after_response == {"type": "http.disconnect"}
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_http10_request(protocol_cls):
def app(scope):
content = "Version: %s" % scope["http_version"]
@ -568,7 +575,7 @@ def test_http10_request(protocol_cls):
assert b"Version: 1.0" in protocol.transport.buffer
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_root_path(protocol_cls):
def app(scope):
path = scope.get("root_path", "") + scope["path"]
@ -581,7 +588,7 @@ def test_root_path(protocol_cls):
assert b"Path: /app/" in protocol.transport.buffer
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_max_concurrency(protocol_cls):
app = lambda scope: None
protocol = get_connected_protocol(app, protocol_cls, limit_concurrency=1)
@ -590,7 +597,7 @@ def test_max_concurrency(protocol_cls):
assert b"HTTP/1.1 503 Service Unavailable" in protocol.transport.buffer
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_shutdown_during_request(protocol_cls):
def app(scope):
return Response(b"", status_code=204)
@ -603,7 +610,7 @@ def test_shutdown_during_request(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_shutdown_during_idle(protocol_cls):
app = lambda scope: None
protocol = get_connected_protocol(app, protocol_cls)
@ -612,7 +619,7 @@ def test_shutdown_during_idle(protocol_cls):
assert protocol.transport.is_closing()
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_100_continue_sent_when_body_consumed(protocol_cls):
class App:
def __init__(self, scope):
@ -647,7 +654,7 @@ def test_100_continue_sent_when_body_consumed(protocol_cls):
assert b'Body: {"hello": "world"}' in protocol.transport.buffer
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_100_continue_not_sent_when_body_not_consumed(protocol_cls):
def app(scope):
return Response(b"", status_code=204)
@ -670,7 +677,7 @@ def test_100_continue_not_sent_when_body_not_consumed(protocol_cls):
assert b"HTTP/1.1 204 No Content" in protocol.transport.buffer
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_unsupported_upgrade_request(protocol_cls):
def app(scope):
pass # pragma: no cover
@ -681,7 +688,7 @@ def test_unsupported_upgrade_request(protocol_cls):
assert b"Unsupported upgrade request." in protocol.transport.buffer
@pytest.mark.parametrize("protocol_cls", [HttpToolsProtocol, H11Protocol])
@pytest.mark.parametrize("protocol_cls", HTTP_PROTOCOLS)
def test_supported_upgrade_request(protocol_cls):
def app(scope):
pass # pragma: no cover

View File

@ -7,13 +7,24 @@ from contextlib import contextmanager
import pytest
import requests
import websockets
from uvicorn.config import Config
from uvicorn.main import ServerState
from uvicorn.protocols.http.h11_impl import H11Protocol
from uvicorn.protocols.websockets.websockets_impl import WebSocketProtocol
from uvicorn.protocols.websockets.wsproto_impl import WSProtocol
try:
import websockets
from uvicorn.protocols.websockets.websockets_impl import WebSocketProtocol
except ImportError:
websockets = None
WebSocketProtocol = None
WS_PROTOCOLS = [p for p in [WSProtocol, WebSocketProtocol] if p is not None]
pytestmark = pytest.mark.skipif(
websockets is None, reason="This test needs the websockets module"
)
class WebSocketResponse:
def __init__(self, scope):
@ -60,7 +71,7 @@ def run_server(app, protocol_cls):
thread.join()
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_invalid_upgrade(protocol_cls):
app = lambda scope: None
@ -78,7 +89,7 @@ def test_invalid_upgrade(protocol_cls):
]
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_accept_connection(protocol_cls):
class App(WebSocketResponse):
async def websocket_connect(self, message):
@ -95,7 +106,7 @@ def test_accept_connection(protocol_cls):
loop.close()
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_close_connection(protocol_cls):
class App(WebSocketResponse):
async def websocket_connect(self, message):
@ -115,7 +126,7 @@ def test_close_connection(protocol_cls):
loop.close()
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_send_text_data_to_client(protocol_cls):
class App(WebSocketResponse):
async def websocket_connect(self, message):
@ -133,7 +144,7 @@ def test_send_text_data_to_client(protocol_cls):
loop.close()
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_send_binary_data_to_client(protocol_cls):
class App(WebSocketResponse):
async def websocket_connect(self, message):
@ -151,7 +162,7 @@ def test_send_binary_data_to_client(protocol_cls):
loop.close()
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_send_and_close_connection(protocol_cls):
class App(WebSocketResponse):
async def websocket_connect(self, message):
@ -177,7 +188,7 @@ def test_send_and_close_connection(protocol_cls):
loop.close()
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_send_text_data_to_server(protocol_cls):
class App(WebSocketResponse):
async def websocket_connect(self, message):
@ -199,7 +210,7 @@ def test_send_text_data_to_server(protocol_cls):
loop.close()
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_send_binary_data_to_server(protocol_cls):
class App(WebSocketResponse):
async def websocket_connect(self, message):
@ -221,7 +232,7 @@ def test_send_binary_data_to_server(protocol_cls):
loop.close()
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_send_after_protocol_close(protocol_cls):
class App(WebSocketResponse):
async def websocket_connect(self, message):
@ -249,7 +260,7 @@ def test_send_after_protocol_close(protocol_cls):
loop.close()
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_missing_handshake(protocol_cls):
class App:
def __init__(self, scope):
@ -269,7 +280,7 @@ def test_missing_handshake(protocol_cls):
loop.close()
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_send_before_handshake(protocol_cls):
class App:
def __init__(self, scope):
@ -289,7 +300,7 @@ def test_send_before_handshake(protocol_cls):
loop.close()
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_duplicate_handshake(protocol_cls):
class App:
def __init__(self, scope):
@ -311,7 +322,7 @@ def test_duplicate_handshake(protocol_cls):
loop.close()
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_asgi_return_value(protocol_cls):
"""
The ASGI callable should return 'None'. If it doesn't make sure that
@ -338,7 +349,7 @@ def test_asgi_return_value(protocol_cls):
loop.close()
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_app_close(protocol_cls):
class App:
def __init__(self, scope):
@ -368,7 +379,7 @@ def test_app_close(protocol_cls):
loop.close()
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
def test_client_close(protocol_cls):
class App:
def __init__(self, scope):
@ -395,7 +406,7 @@ def test_client_close(protocol_cls):
loop.close()
@pytest.mark.parametrize("protocol_cls", [WebSocketProtocol, WSProtocol])
@pytest.mark.parametrize("protocol_cls", WS_PROTOCOLS)
@pytest.mark.parametrize("subprotocol", ["proto1", "proto2"])
def test_subprotocols(protocol_cls, subprotocol):
class App(WebSocketResponse):

View File

@ -1,24 +1,57 @@
import asyncio
import uvloop
from uvicorn.config import Config
from uvicorn.loops.auto import auto_loop_setup
from uvicorn.main import ServerState
from uvicorn.protocols.http.auto import AutoHTTPProtocol
from uvicorn.protocols.http.httptools_impl import HttpToolsProtocol
from uvicorn.protocols.websockets.auto import AutoWebSocketsProtocol
try:
import uvloop
except ImportError:
uvloop = None
try:
import httptools
except ImportError:
httptools = None
try:
import websockets
except ImportError: # Note that when this happens, the websocket tests are skipped
websockets = None
# TODO: Add pypy to our testing matrix, and assert we get the correct classes
# dependent on the platform we're running the tests under.
def test_loop_auto():
loop = auto_loop_setup()
policy = asyncio.get_event_loop_policy()
assert isinstance(policy, asyncio.events.BaseDefaultEventLoopPolicy)
if uvloop is None:
assert type(policy).__module__.startswith("asyncio")
else:
assert isinstance(policy, uvloop.EventLoopPolicy)
assert type(policy).__module__.startswith("uvloop")
def test_http_auto():
config = Config(app=None)
server_state = ServerState()
protocol = AutoHTTPProtocol(config=config, server_state=server_state)
assert isinstance(protocol, HttpToolsProtocol)
if httptools is None:
assert type(protocol).__name__ == "H11Protocol"
else:
assert type(protocol).__name__ == "HttpToolsProtocol"
def test_loop_auto():
loop = auto_loop_setup()
assert isinstance(asyncio.get_event_loop_policy(), uvloop.EventLoopPolicy)
def test_websocket_auto():
config = Config(app=None)
server_state = ServerState()
protocol = AutoWebSocketsProtocol(config=config, server_state=server_state)
if websockets is None:
assert type(protocol).__name__ == "WSProtocol"
else:
assert type(protocol).__name__ == "WebSocketProtocol"

View File

@ -2,7 +2,6 @@ import threading
import time
import requests
from uvicorn.config import Config
from uvicorn.main import Server

View File

@ -1,4 +1,5 @@
import contextlib
import sys
import tempfile
import threading
import time
@ -8,7 +9,6 @@ from functools import partialmethod
import pytest
import requests
from urllib3.exceptions import InsecureRequestWarning
from uvicorn.config import Config
from uvicorn.main import Server
@ -105,6 +105,9 @@ def create_certfile_and_keyfile(request):
def test_run(create_certfile_and_keyfile):
if sys.platform.startswith("win"):
pytest.skip("Skipping SSL test on Windows for now :(")
certfile, keyfile = create_certfile_and_keyfile
class App: