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
|
import asyncio
import pytest
from flask import Blueprint
from flask import Flask
from flask import request
from flask.views import MethodView
from flask.views import View
pytest.importorskip("asgiref")
class AppError(Exception):
pass
class BlueprintError(Exception):
pass
class AsyncView(View):
methods = ["GET", "POST"]
async def dispatch_request(self):
await asyncio.sleep(0)
return request.method
class AsyncMethodView(MethodView):
async def get(self):
await asyncio.sleep(0)
return "GET"
async def post(self):
await asyncio.sleep(0)
return "POST"
@pytest.fixture(name="async_app")
def _async_app():
app = Flask(__name__)
@app.route("/", methods=["GET", "POST"])
@app.route("/home", methods=["GET", "POST"])
async def index():
await asyncio.sleep(0)
return request.method
@app.errorhandler(AppError)
async def handle(_):
return "", 412
@app.route("/error")
async def error():
raise AppError()
blueprint = Blueprint("bp", __name__)
@blueprint.route("/", methods=["GET", "POST"])
async def bp_index():
await asyncio.sleep(0)
return request.method
@blueprint.errorhandler(BlueprintError)
async def bp_handle(_):
return "", 412
@blueprint.route("/error")
async def bp_error():
raise BlueprintError()
app.register_blueprint(blueprint, url_prefix="/bp")
app.add_url_rule("/view", view_func=AsyncView.as_view("view"))
app.add_url_rule("/methodview", view_func=AsyncMethodView.as_view("methodview"))
return app
@pytest.mark.parametrize("path", ["/", "/home", "/bp/", "/view", "/methodview"])
def test_async_route(path, async_app):
test_client = async_app.test_client()
response = test_client.get(path)
assert b"GET" in response.get_data()
response = test_client.post(path)
assert b"POST" in response.get_data()
@pytest.mark.parametrize("path", ["/error", "/bp/error"])
def test_async_error_handler(path, async_app):
test_client = async_app.test_client()
response = test_client.get(path)
assert response.status_code == 412
def test_async_before_after_request():
app_before_called = False
app_after_called = False
bp_before_called = False
bp_after_called = False
app = Flask(__name__)
@app.route("/")
def index():
return ""
@app.before_request
async def before():
nonlocal app_before_called
app_before_called = True
@app.after_request
async def after(response):
nonlocal app_after_called
app_after_called = True
return response
blueprint = Blueprint("bp", __name__)
@blueprint.route("/")
def bp_index():
return ""
@blueprint.before_request
async def bp_before():
nonlocal bp_before_called
bp_before_called = True
@blueprint.after_request
async def bp_after(response):
nonlocal bp_after_called
bp_after_called = True
return response
app.register_blueprint(blueprint, url_prefix="/bp")
test_client = app.test_client()
test_client.get("/")
assert app_before_called
assert app_after_called
test_client.get("/bp/")
assert bp_before_called
assert bp_after_called
|
