fixed all tests

Author: seidnerj

tests/assets/config_ssl.py

@@ -2,7 +2,7 @@
 
 access_log_format = "bob"
 bind = ["127.0.0.1:5555"]
-certfile = "assets/cert.pem"
+certfile = "tests/assets/cert.pem"
 ciphers = "ECDHE+AESGCM"
 h11_max_incomplete_size = 4
-keyfile = "assets/key.pem"
+keyfile = "tests/assets/key.pem"

tests/asyncio/test_keep_alive.py

@@ -45,6 +45,7 @@ async def slow_framework(
 @pytest_asyncio.fixture(name="server", scope="function")  # type: ignore[misc]
 async def _server() -> AsyncGenerator[TCPServer, None]:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     config = Config()
     config.keep_alive_timeout = KEEP_ALIVE_TIMEOUT
     server = TCPServer(

tests/asyncio/test_lifespan.py

@@ -22,6 +22,7 @@ async def no_lifespan_app(scope: Scope, receive: Callable, send: Callable) -> No
 @pytest.mark.asyncio
 async def test_ensure_no_race_condition() -> None:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     config = Config()
     config.startup_timeout = 0.2
     lifespan = Lifespan(ASGIWrapper(no_lifespan_app), config, event_loop)
@@ -33,6 +34,7 @@ async def test_ensure_no_race_condition() -> None:
 @pytest.mark.asyncio
 async def test_startup_timeout_error() -> None:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     config = Config()
     config.startup_timeout = 0.01
     lifespan = Lifespan(ASGIWrapper(SlowLifespanFramework(0.02, asyncio.sleep)), config, event_loop)
@@ -46,6 +48,7 @@ async def test_startup_timeout_error() -> None:
 @pytest.mark.asyncio
 async def test_startup_failure() -> None:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     lifespan = Lifespan(ASGIWrapper(lifespan_failure), Config(), event_loop)
     lifespan_task = event_loop.create_task(lifespan.handle_lifespan())
     await lifespan.wait_for_startup()
@@ -62,6 +65,7 @@ async def return_app(scope: Scope, receive: Callable, send: Callable) -> None:
 @pytest.mark.asyncio
 async def test_lifespan_return() -> None:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     lifespan = Lifespan(ASGIWrapper(return_app), Config(), event_loop)
     lifespan_task = event_loop.create_task(lifespan.handle_lifespan())
     await lifespan.wait_for_startup()

tests/asyncio/test_sanity.py

@@ -18,6 +18,7 @@
 @pytest.mark.asyncio
 async def test_http1_request() -> None:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     server = TCPServer(
         ASGIWrapper(sanity_framework),
         event_loop,
@@ -76,6 +77,7 @@ async def test_http1_request() -> None:
 @pytest.mark.asyncio
 async def test_http1_websocket() -> None:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     server = TCPServer(
         ASGIWrapper(sanity_framework),
         event_loop,
@@ -114,6 +116,7 @@ async def test_http1_websocket() -> None:
 @pytest.mark.asyncio
 async def test_http2_request() -> None:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     server = TCPServer(
         ASGIWrapper(sanity_framework),
         event_loop,
@@ -178,6 +181,7 @@ async def test_http2_request() -> None:
 @pytest.mark.asyncio
 async def test_http2_websocket() -> None:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     server = TCPServer(
         ASGIWrapper(sanity_framework),
         event_loop,

tests/asyncio/test_task_group.py

@@ -14,6 +14,7 @@
 @pytest.mark.asyncio
 async def test_spawn_app(http_scope: HTTPScope) -> None:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     async def _echo_app(scope: Scope, receive: Callable, send: Callable) -> None:
         while True:
             message = await receive()
@@ -32,10 +33,9 @@ async def _echo_app(scope: Scope, receive: Callable, send: Callable) -> None:
 
 
 @pytest.mark.asyncio
-async def test_spawn_app_error(
-    http_scope: HTTPScope
-) -> None:
+async def test_spawn_app_error(http_scope: HTTPScope) -> None:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     async def _error_app(scope: Scope, receive: Callable, send: Callable) -> None:
         raise Exception()
 

tests/asyncio/test_tcp_server.py

@@ -15,6 +15,7 @@
 @pytest.mark.asyncio
 async def test_completes_on_closed() -> None:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     server = TCPServer(
         ASGIWrapper(echo_framework),
         event_loop,
@@ -32,6 +33,7 @@ async def test_completes_on_closed() -> None:
 @pytest.mark.asyncio
 async def test_complets_on_half_close() -> None:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     server = TCPServer(
         ASGIWrapper(echo_framework),
         event_loop,

tests/protocol/test_h11.py

@@ -136,10 +136,9 @@ async def test_protocol_send_stream_closed(
 
 
 @pytest.mark.asyncio
-async def test_protocol_instant_recycle(
-    protocol: H11Protocol
-) -> None:
+async def test_protocol_instant_recycle(protocol: H11Protocol) -> None:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     # This test task acts as the asgi app, spawned tasks act as the
     # server.
     data = b"GET / HTTP/1.1\r\nHost: hypercorn\r\n\r\n"

tests/protocol/test_h2.py

@@ -22,6 +22,7 @@
 @pytest.mark.asyncio
 async def test_stream_buffer_push_and_pop() -> None:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     stream_buffer = StreamBuffer(EventWrapper)
 
     async def _push_over_limit() -> bool:
@@ -39,6 +40,7 @@ async def _push_over_limit() -> bool:
 @pytest.mark.asyncio
 async def test_stream_buffer_drain() -> None:
     event_loop: asyncio.AbstractEventLoop = asyncio.get_running_loop()
+
     stream_buffer = StreamBuffer(EventWrapper)
     await stream_buffer.push(b"a" * 10)
 

tests/test_config.py

@@ -53,12 +53,23 @@ def test_create_ssl_context() -> None:
     path = os.path.join(os.path.dirname(__file__), "assets/config_ssl.py")
     config = Config.from_pyfile(path)
     context = config.create_ssl_context()
-    assert context.options & (
-        ssl.OP_NO_SSLv2
-        | ssl.OP_NO_SSLv3
-        | ssl.OP_NO_TLSv1
-        | ssl.OP_NO_TLSv1_1
-        | ssl.OP_NO_COMPRESSION
+
+    # NOTE: In earlier versions of python context.options is equal to <Options.OP_NO_COMPRESSION: 0>
+    #       hence the ANDing context.options with the specified ssl options results in
+    #       "<Options.OP_NO_COMPRESSION: 0>", which as a Boolean value, is False.
+    #
+    #       To overcome this, instead of checking that the result in True, we will check that it is
+    #        equal to "context.options".
+    assert (
+        context.options
+        & (
+            ssl.OP_NO_SSLv2
+            | ssl.OP_NO_SSLv3
+            | ssl.OP_NO_TLSv1
+            | ssl.OP_NO_TLSv1_1
+            | ssl.OP_NO_COMPRESSION
+        )
+        == context.options
     )
 
 

tests/trio/test_lifespan.py

@@ -1,4 +1,5 @@
 from __future__ import annotations
+
 import sys
 
 if sys.version_info < (3, 11):