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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
|
from __future__ import annotations
import asyncio
from typing import cast
from unittest.mock import Mock
import pytest
from hypercorn.typing import HTTPScope
from werkzeug.datastructures import Headers
from werkzeug.exceptions import BadRequest
from quart.app import Quart
from quart.ctx import after_this_request
from quart.ctx import AppContext
from quart.ctx import copy_current_app_context
from quart.ctx import copy_current_request_context
from quart.ctx import copy_current_websocket_context
from quart.ctx import has_app_context
from quart.ctx import has_request_context
from quart.ctx import RequestContext
from quart.globals import g
from quart.globals import request
from quart.globals import websocket
from quart.routing import QuartRule
from quart.testing import make_test_headers_path_and_query_string
from quart.testing import no_op_push
from quart.wrappers import Request
async def test_request_context_match(http_scope: HTTPScope) -> None:
app = Quart(__name__)
url_adapter = Mock()
rule = QuartRule("/", methods={"GET"}, endpoint="index")
url_adapter.match.return_value = (rule, {"arg": "value"})
app.create_url_adapter = lambda *_: url_adapter # type: ignore
request = Request(
"GET",
"http",
"/",
b"",
Headers([("host", "quart.com")]),
"",
"1.1",
http_scope,
send_push_promise=no_op_push,
)
async with RequestContext(app, request):
assert request.url_rule == rule
assert request.view_args == {"arg": "value"}
async def test_bad_request_if_websocket_route(http_scope: HTTPScope) -> None:
app = Quart(__name__)
url_adapter = Mock()
url_adapter.match.side_effect = BadRequest()
app.create_url_adapter = lambda *_: url_adapter # type: ignore
request = Request(
"GET",
"http",
"/",
b"",
Headers([("host", "quart.com")]),
"",
"1.1",
http_scope,
send_push_promise=no_op_push,
)
async with RequestContext(app, request):
assert isinstance(request.routing_exception, BadRequest)
async def test_after_this_request(http_scope: HTTPScope) -> None:
app = Quart(__name__)
headers, path, query_string = make_test_headers_path_and_query_string(app, "/")
async with RequestContext(
Quart(__name__),
Request(
"GET",
"http",
path,
query_string,
headers,
"",
"1.1",
http_scope,
send_push_promise=no_op_push,
),
) as context:
after_this_request(lambda: "hello") # type: ignore
assert context._after_request_functions[0]() == "hello" # type: ignore
async def test_has_request_context(http_scope: HTTPScope) -> None:
app = Quart(__name__)
headers, path, query_string = make_test_headers_path_and_query_string(app, "/")
request = Request(
"GET",
"http",
path,
query_string,
headers,
"",
"1.1",
http_scope,
send_push_promise=no_op_push,
)
async with RequestContext(Quart(__name__), request):
assert has_request_context() is True
assert has_app_context() is True
assert has_request_context() is False
assert has_app_context() is False
async def test_has_app_context() -> None:
async with AppContext(Quart(__name__)):
assert has_app_context() is True
assert has_app_context() is False
async def test_copy_current_app_context() -> None:
app = Quart(__name__)
@app.route("/")
async def index() -> str:
g.foo = "bar"
@copy_current_app_context
async def within_context() -> None:
assert g.foo == "bar"
await asyncio.ensure_future(within_context())
return ""
test_client = app.test_client()
response = await test_client.get("/")
assert response.status_code == 200
def test_copy_current_app_context_error() -> None:
with pytest.raises(RuntimeError):
copy_current_app_context(lambda: None)()
async def test_copy_current_request_context() -> None:
app = Quart(__name__)
@app.route("/")
async def index() -> str:
@copy_current_request_context
async def within_context() -> None:
assert request.path == "/"
await asyncio.ensure_future(within_context())
return ""
test_client = app.test_client()
response = await test_client.get("/")
assert response.status_code == 200
def test_copy_current_request_context_error() -> None:
with pytest.raises(RuntimeError):
copy_current_request_context(lambda: None)()
async def test_works_without_copy_current_request_context() -> None:
app = Quart(__name__)
@app.route("/")
async def index() -> str:
async def within_context() -> None:
assert request.path == "/"
await asyncio.ensure_future(within_context())
return ""
test_client = app.test_client()
response = await test_client.get("/")
assert response.status_code == 200
async def test_copy_current_websocket_context() -> None:
app = Quart(__name__)
@app.websocket("/")
async def index() -> None:
@copy_current_websocket_context
async def within_context() -> str:
return websocket.path
data = await asyncio.ensure_future(within_context())
await websocket.send(data.encode())
test_client = app.test_client()
async with test_client.websocket("/") as test_websocket:
data = await test_websocket.receive()
assert cast(bytes, data) == b"/"
def test_copy_current_websocket_context_error() -> None:
with pytest.raises(RuntimeError):
copy_current_websocket_context(lambda: None)()
|
