Drop private imports in tests/conftest.py (#2569)

Co-authored-by: Tom Christie <[email protected]>

Author: florimondmanca

tests/conftest.py

@@ -17,12 +17,9 @@
 from uvicorn.config import Config
 from uvicorn.server import Server
 
-from httpx import URL
+import httpx
 from tests.concurrency import sleep
 
-if typing.TYPE_CHECKING:  # pragma: no cover
-    from httpx._transports.asgi import _Receive, _Send
-
 ENVIRONMENT_VARIABLES = {
     "SSL_CERT_FILE",
     "SSL_CERT_DIR",
@@ -51,10 +48,15 @@ def clean_environ():
     os.environ.update(original_environ)
 
 
-_Scope = typing.Dict[str, typing.Any]
+Message = typing.Dict[str, typing.Any]
+Receive = typing.Callable[[], typing.Awaitable[Message]]
+Send = typing.Callable[
+    [typing.Dict[str, typing.Any]], typing.Coroutine[None, None, None]
+]
+Scope = typing.Dict[str, typing.Any]
 
 
-async def app(scope: _Scope, receive: "_Receive", send: "_Send") -> None:
+async def app(scope: Scope, receive: Receive, send: Send) -> None:
     assert scope["type"] == "http"
     if scope["path"].startswith("/slow_response"):
         await slow_response(scope, receive, send)
@@ -74,7 +76,7 @@ async def app(scope: _Scope, receive: "_Receive", send: "_Send") -> None:
         await hello_world(scope, receive, send)
 
 
-async def hello_world(scope: _Scope, receive: "_Receive", send: "_Send") -> None:
+async def hello_world(scope: Scope, receive: Receive, send: Send) -> None:
     await send(
         {
             "type": "http.response.start",
@@ -85,7 +87,7 @@ async def hello_world(scope: _Scope, receive: "_Receive", send: "_Send") -> None
     await send({"type": "http.response.body", "body": b"Hello, world!"})
 
 
-async def hello_world_json(scope: _Scope, receive: "_Receive", send: "_Send") -> None:
+async def hello_world_json(scope: Scope, receive: Receive, send: Send) -> None:
     await send(
         {
             "type": "http.response.start",
@@ -96,7 +98,7 @@ async def hello_world_json(scope: _Scope, receive: "_Receive", send: "_Send") ->
     await send({"type": "http.response.body", "body": b'{"Hello": "world!"}'})
 
 
-async def slow_response(scope: _Scope, receive: "_Receive", send: "_Send") -> None:
+async def slow_response(scope: Scope, receive: Receive, send: Send) -> None:
     await send(
         {
             "type": "http.response.start",
@@ -108,7 +110,7 @@ async def slow_response(scope: _Scope, receive: "_Receive", send: "_Send") -> No
     await send({"type": "http.response.body", "body": b"Hello, world!"})
 
 
-async def status_code(scope: _Scope, receive: "_Receive", send: "_Send") -> None:
+async def status_code(scope: Scope, receive: Receive, send: Send) -> None:
     status_code = int(scope["path"].replace("/status/", ""))
     await send(
         {
@@ -120,7 +122,7 @@ async def status_code(scope: _Scope, receive: "_Receive", send: "_Send") -> None
     await send({"type": "http.response.body", "body": b"Hello, world!"})
 
 
-async def echo_body(scope: _Scope, receive: "_Receive", send: "_Send") -> None:
+async def echo_body(scope: Scope, receive: Receive, send: Send) -> None:
     body = b""
     more_body = True
 
@@ -139,7 +141,7 @@ async def echo_body(scope: _Scope, receive: "_Receive", send: "_Send") -> None:
     await send({"type": "http.response.body", "body": body})
 
 
-async def echo_binary(scope: _Scope, receive: "_Receive", send: "_Send") -> None:
+async def echo_binary(scope: Scope, receive: Receive, send: Send) -> None:
     body = b""
     more_body = True
 
@@ -158,7 +160,7 @@ async def echo_binary(scope: _Scope, receive: "_Receive", send: "_Send") -> None
     await send({"type": "http.response.body", "body": body})
 
 
-async def echo_headers(scope: _Scope, receive: "_Receive", send: "_Send") -> None:
+async def echo_headers(scope: Scope, receive: Receive, send: Send) -> None:
     body = {
         name.capitalize().decode(): value.decode()
         for name, value in scope.get("headers", [])
@@ -173,7 +175,7 @@ async def echo_headers(scope: _Scope, receive: "_Receive", send: "_Send") -> Non
     await send({"type": "http.response.body", "body": json.dumps(body).encode()})
 
 
-async def redirect_301(scope: _Scope, receive: "_Receive", send: "_Send") -> None:
+async def redirect_301(scope: Scope, receive: Receive, send: Send) -> None:
     await send(
         {"type": "http.response.start", "status": 301, "headers": [[b"location", b"/"]]}
     )
@@ -227,9 +229,9 @@ def cert_encrypted_private_key_file(localhost_cert):
 
 class TestServer(Server):
     @property
-    def url(self) -> URL:
+    def url(self) -> httpx.URL:
         protocol = "https" if self.config.is_ssl else "http"
-        return URL(f"{protocol}://{self.config.host}:{self.config.port}/")
+        return httpx.URL(f"{protocol}://{self.config.host}:{self.config.port}/")
 
     def install_signal_handlers(self) -> None:
         # Disable the default installation of handlers for signals such as SIGTERM,