-
-
Notifications
You must be signed in to change notification settings - Fork 25
/
test_wsgi.py
130 lines (109 loc) · 3.81 KB
/
test_wsgi.py
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
import sys
import pytest
from starlette.middleware.wsgi import WSGIMiddleware, build_environ
from starlette.testclient import TestClient
def hello_world(environ, start_response):
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)
return [output]
def echo_body(environ, start_response):
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)
return [output]
def raise_exception(environ, start_response):
raise RuntimeError("Something went wrong")
def return_exc_info(environ, start_response):
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, exc_info=sys.exc_info())
return [output]
def test_wsgi_get():
app = WSGIMiddleware(hello_world)
client = TestClient(app)
response = client.get("/")
assert response.status_code == 200
assert response.text == "Hello World!\n"
def test_wsgi_post():
app = WSGIMiddleware(echo_body)
client = TestClient(app)
response = client.post("/", json={"example": 123})
assert response.status_code == 200
assert response.text == '{"example": 123}'
def test_wsgi_exception():
# Note that we're testing the WSGI app directly here.
# The HTTP protocol implementations would catch this error and return 500.
app = WSGIMiddleware(raise_exception)
client = TestClient(app)
with pytest.raises(RuntimeError):
client.get("/")
def test_wsgi_exc_info():
# Note that we're testing the WSGI app directly here.
# The HTTP protocol implementations would catch this error and return 500.
app = WSGIMiddleware(return_exc_info)
client = TestClient(app)
with pytest.raises(RuntimeError):
response = client.get("/")
app = WSGIMiddleware(return_exc_info)
client = TestClient(app, raise_server_exceptions=False)
response = client.get("/")
assert response.status_code == 500
assert response.text == "Internal Server Error"
def test_build_environ():
scope = {
"type": "http",
"http_version": "1.1",
"method": "GET",
"scheme": "https",
"path": "/",
"query_string": b"a=123&b=456",
"headers": [
(b"host", b"www.example.org"),
(b"content-type", b"application/json"),
(b"content-length", b"18"),
(b"accept", b"application/json"),
(b"accept", b"text/plain"),
],
"client": ("134.56.78.4", 1453),
"server": ("www.example.org", 443),
}
body = b'{"example":"body"}'
environ = build_environ(scope, body)
stream = environ.pop("wsgi.input")
assert stream.read() == b'{"example":"body"}'
assert environ == {
"CONTENT_LENGTH": "18",
"CONTENT_TYPE": "application/json",
"HTTP_ACCEPT": "application/json,text/plain",
"HTTP_HOST": "www.example.org",
"PATH_INFO": "/",
"QUERY_STRING": "a=123&b=456",
"REMOTE_ADDR": "134.56.78.4",
"REQUEST_METHOD": "GET",
"SCRIPT_NAME": "",
"SERVER_NAME": "www.example.org",
"SERVER_PORT": 443,
"SERVER_PROTOCOL": "HTTP/1.1",
"wsgi.errors": sys.stdout,
"wsgi.multiprocess": True,
"wsgi.multithread": True,
"wsgi.run_once": False,
"wsgi.url_scheme": "https",
"wsgi.version": (1, 0),
}