encode/httpx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
Co-authored-by: Tom Christie <tom@tomchristie.com>
This commit is contained in:
Florimond Manca 2023-02-09 16:54:53 +01:00 committed by GitHub
parent 78d381fc7d
commit ef06f7d076
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 18 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
tests/conftest.py
View File

@ -17,12 +17,9 @@ from cryptography.hazmat.primitives.serialization import (
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,