Fix Test Cases: test_http for Py3.9+, test_json_response_json for ujson 5.4.0+, and test_zero_downtime; Test Case Type Annotations (sanic-org#2504)

Author: ChihweiLHBird

tests/test_http.py

@@ -2,6 +2,7 @@
 
 from collections import namedtuple
 from pathlib import Path
+from sys import version_info
 
 import pytest
 
@@ -35,15 +36,15 @@ async def upload_handler(request):
 
 
 @pytest.fixture
-def runner(test_app):
+def runner(test_app: Sanic):
     client = ReusableClient(test_app, port=PORT)
     client.run()
     yield client
     client.stop()
 
 
 @pytest.fixture
-def client(runner):
+def client(runner: ReusableClient):
     client = namedtuple("Client", ("raw", "send", "recv"))
 
     raw = RawClient(runner.host, runner.port)
@@ -74,7 +75,10 @@ def test_full_message(client):
         """
     )
     response = client.recv()
-    assert len(response) == 151
+
+    # AltSvcCheck touchup removes the Alt-Svc header from the 
+    # response in the Python 3.9+ in this case
+    assert len(response) == (151 if version_info < (3, 9) else 140)
     assert b"200 OK" in response
 
 

tests/test_json_encoding.py

@@ -3,18 +3,27 @@
 from dataclasses import asdict, dataclass
 from functools import partial
 from json import dumps as sdumps
+from string import ascii_lowercase
+from typing import Dict
 
 import pytest
 
 
 try:
+    import ujson
+
     from ujson import dumps as udumps
 
+    ujson_version = tuple(
+        map(int, ujson.__version__.strip(ascii_lowercase).split("."))
+    )
+
     NO_UJSON = False
     DEFAULT_DUMPS = udumps
 except ModuleNotFoundError:
     NO_UJSON = True
     DEFAULT_DUMPS = partial(sdumps, separators=(",", ":"))
+    ujson_version = None
 
 from sanic import Sanic
 from sanic.response import BaseHTTPResponse, json
@@ -34,7 +43,7 @@ def foo():
 
 
 @pytest.fixture
-def payload(foo):
+def payload(foo: Foo):
     return {"foo": foo}
 
 
@@ -58,7 +67,7 @@ def my_custom_encoder():
 
 
 @pytest.mark.skipif(NO_UJSON is True, reason="ujson not installed")
-def test_json_response_ujson(payload):
+def test_json_response_ujson(payload: Dict[str, Foo]):
     """ujson will look at __json__"""
     response = json(payload)
     assert response.body == b'{"foo":{"bar":"bar"}}'
@@ -75,7 +84,13 @@ def test_json_response_ujson(payload):
         json(payload)
 
 
-@pytest.mark.skipif(NO_UJSON is True, reason="ujson not installed")
+@pytest.mark.skipif(
+    NO_UJSON is True or ujson_version >= (5, 4, 0),
+    reason=(
+        "ujson not installed or version is 5.4.0 or newer, "
+        "which can handle arbitrary size integers"
+    ),
+)
 def test_json_response_json():
     """One of the easiest ways to tell the difference is that ujson cannot
     serialize over 64 bits"""

tests/test_unix_socket.py

@@ -5,13 +5,17 @@
 import subprocess
 import sys
 
+from asyncio import AbstractEventLoop
 from string import ascii_lowercase
 
 import httpcore
 import httpx
 import pytest
 
+from pytest import LogCaptureFixture
+
 from sanic import Sanic
+from sanic.request import Request
 from sanic.response import text
 
 
@@ -45,7 +49,7 @@ def socket_cleanup():
         pass
 
 
-def test_unix_socket_creation(caplog):
+def test_unix_socket_creation(caplog: LogCaptureFixture):
     from socket import AF_UNIX, socket
 
     with socket(AF_UNIX) as sock:
@@ -56,7 +60,7 @@ def test_unix_socket_creation(caplog):
     app = Sanic(name="test")
 
     @app.listener("after_server_start")
-    def running(app, loop):
+    def running(app: Sanic, loop: AbstractEventLoop):
         assert os.path.exists(SOCKPATH)
         assert ino != os.stat(SOCKPATH).st_ino
         app.stop()
@@ -73,7 +77,7 @@ def running(app, loop):
 
 
 @pytest.mark.parametrize("path", (".", "no-such-directory/sanictest.sock"))
-def test_invalid_paths(path):
+def test_invalid_paths(path: str):
     app = Sanic(name="test")
 
     with pytest.raises((FileExistsError, FileNotFoundError)):
@@ -87,7 +91,7 @@ def test_dont_replace_file():
     app = Sanic(name="test")
 
     @app.listener("after_server_start")
-    def stop(app, loop):
+    def stop(app: Sanic, loop: AbstractEventLoop):
         app.stop()
 
     with pytest.raises(FileExistsError):
@@ -104,7 +108,7 @@ def test_dont_follow_symlink():
     app = Sanic(name="test")
 
     @app.listener("after_server_start")
-    def stop(app, loop):
+    def stop(app: Sanic, loop: AbstractEventLoop):
         app.stop()
 
     with pytest.raises(FileExistsError):
@@ -115,7 +119,7 @@ def test_socket_deleted_while_running():
     app = Sanic(name="test")
 
     @app.listener("after_server_start")
-    async def hack(app, loop):
+    async def hack(app: Sanic, loop: AbstractEventLoop):
         os.unlink(SOCKPATH)
         app.stop()
 
@@ -126,7 +130,7 @@ def test_socket_replaced_with_file():
     app = Sanic(name="test")
 
     @app.listener("after_server_start")
-    async def hack(app, loop):
+    async def hack(app: Sanic, loop: AbstractEventLoop):
         os.unlink(SOCKPATH)
         with open(SOCKPATH, "w") as f:
             f.write("Not a socket")
@@ -139,11 +143,11 @@ def test_unix_connection():
     app = Sanic(name="test")
 
     @app.get("/")
-    def handler(request):
+    def handler(request: Request):
         return text(f"{request.conn_info.server}")
 
     @app.listener("after_server_start")
-    async def client(app, loop):
+    async def client(app: Sanic, loop: AbstractEventLoop):
         if httpx_version >= (0, 20):
             transport = httpx.AsyncHTTPTransport(uds=SOCKPATH)
         else:
@@ -162,11 +166,11 @@ async def client(app, loop):
 app_multi = Sanic(name="test")
 
 
-def handler(request):
+def handler(request: Request):
     return text(f"{request.conn_info.server}")
 
 
-async def client(app, loop):
+async def client(app: Sanic, loop: AbstractEventLoop):
     try:
         async with httpx.AsyncClient(uds=SOCKPATH) as client:
             r = await client.get("http://myhost.invalid/")
@@ -229,7 +233,7 @@ def spawn():
         ino = os.stat(SOCKPATH).st_ino
         task = asyncio.get_event_loop().create_task(client())
         start_time = current_time()
-        while current_time() < start_time + 4:
+        while current_time() < start_time + 6:
             # Start a new one and wait until the socket is replaced
             processes.append(spawn())
             while ino == os.stat(SOCKPATH).st_ino: