1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
|
from __future__ import annotations
import io
import sys
from typing import AsyncGenerator, Callable
import a2wsgi
import httpx
import pytest
from uvicorn._types import Environ, HTTPRequestEvent, HTTPScope, StartResponse
from uvicorn.middleware import wsgi
def hello_world(environ: Environ, start_response: StartResponse) -> list[bytes]:
status = "200 OK"
output = b"Hello World!\n"
headers = [
("Content-Type", "text/plain; charset=utf-8"),
("Content-Length", str(len(output))),
]
start_response(status, headers, None)
return [output]
def echo_body(environ: Environ, start_response: StartResponse) -> list[bytes]:
status = "200 OK"
output = environ["wsgi.input"].read()
headers = [
("Content-Type", "text/plain; charset=utf-8"),
("Content-Length", str(len(output))),
]
start_response(status, headers, None)
return [output]
def raise_exception(environ: Environ, start_response: StartResponse) -> list[bytes]:
raise RuntimeError("Something went wrong")
def return_exc_info(environ: Environ, start_response: StartResponse) -> list[bytes]:
try:
raise RuntimeError("Something went wrong")
except RuntimeError:
status = "500 Internal Server Error"
output = b"Internal Server Error"
headers = [
("Content-Type", "text/plain; charset=utf-8"),
("Content-Length", str(len(output))),
]
start_response(status, headers, sys.exc_info()) # type: ignore[arg-type]
return [output]
@pytest.fixture(params=[wsgi._WSGIMiddleware, a2wsgi.WSGIMiddleware])
def wsgi_middleware(request: pytest.FixtureRequest) -> Callable:
return request.param
@pytest.mark.anyio
async def test_wsgi_get(wsgi_middleware: Callable) -> None:
transport = httpx.ASGITransport(wsgi_middleware(hello_world))
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
response = await client.get("/")
assert response.status_code == 200
assert response.text == "Hello World!\n"
@pytest.mark.anyio
async def test_wsgi_post(wsgi_middleware: Callable) -> None:
transport = httpx.ASGITransport(wsgi_middleware(echo_body))
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
response = await client.post("/", json={"example": 123})
assert response.status_code == 200
assert response.text in ['{"example": 123}', '{"example":123}']
@pytest.mark.anyio
async def test_wsgi_put_more_body(wsgi_middleware: Callable) -> None:
async def generate_body() -> AsyncGenerator[bytes, None]:
for _ in range(1024):
yield b"123456789abcdef\n" * 64
transport = httpx.ASGITransport(wsgi_middleware(echo_body))
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
response = await client.put("/", content=generate_body())
assert response.status_code == 200
assert response.text == "123456789abcdef\n" * 64 * 1024
@pytest.mark.anyio
async def test_wsgi_exception(wsgi_middleware: Callable) -> None:
# Note that we're testing the WSGI app directly here.
# The HTTP protocol implementations would catch this error and return 500.
transport = httpx.ASGITransport(wsgi_middleware(raise_exception))
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
with pytest.raises(RuntimeError):
await client.get("/")
@pytest.mark.anyio
async def test_wsgi_exc_info(wsgi_middleware: Callable) -> None:
app = wsgi_middleware(return_exc_info)
transport = httpx.ASGITransport(
app=app,
raise_app_exceptions=False,
)
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
response = await client.get("/")
assert response.status_code == 500
assert response.text == "Internal Server Error"
def test_build_environ_encoding() -> None:
scope: HTTPScope = {
"asgi": {"version": "3.0", "spec_version": "2.0"},
"scheme": "http",
"raw_path": b"/\xe6\x96\x87%2Fall",
"type": "http",
"http_version": "1.1",
"method": "GET",
"path": "/文/all",
"root_path": "/文",
"client": None,
"server": None,
"query_string": b"a=123&b=456",
"headers": [(b"key", b"value1"), (b"key", b"value2")],
"extensions": {},
}
message: HTTPRequestEvent = {
"type": "http.request",
"body": b"",
"more_body": False,
}
environ = wsgi.build_environ(scope, message, io.BytesIO(b""))
assert environ["SCRIPT_NAME"] == "/文".encode().decode("latin-1")
assert environ["PATH_INFO"] == b"/all".decode("latin-1")
assert environ["HTTP_KEY"] == "value1,value2"
|
