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
|
"""
test_async
~~~~~~~~~~
Tests using Flask async.
Make sure our decorators allow for async views
Make sure signal receivers can be async
:copyright: (c) 2023-2023 by J. Christopher Wagner (jwag).
:license: MIT, see LICENSE for more details.
"""
import asyncio
import base64
import pytest
from flask_principal import identity_changed
from flask_security import (
anonymous_user_required,
auth_token_required,
auth_required,
http_auth_required,
roles_required,
roles_accepted,
permissions_required,
permissions_accepted,
unauth_csrf,
)
from tests.test_utils import (
authenticate,
json_authenticate,
)
pytestmark = pytest.mark.flask_async()
def test_auth_required(app, client):
@app.route("/async_test")
@auth_required()
async def async_test():
await asyncio.sleep(0)
return "Access Granted"
authenticate(client)
response = client.get("/async_test")
assert b"Access Granted" in response.data
def test_auth_token_required(app, client):
@app.route("/async_test")
@auth_token_required
async def async_test():
await asyncio.sleep(0)
return "Access Granted"
@identity_changed.connect_via(app)
async def ic(myapp, identity, **extra_args):
await asyncio.sleep(0)
response = json_authenticate(client)
token = response.json["response"]["user"]["authentication_token"]
response = client.get("/async_test?auth_token=" + token)
assert b"Access Granted" in response.data
def test_auth_http_required(app, client):
@app.route("/async_test")
@http_auth_required
async def async_test():
await asyncio.sleep(0)
return "Access Granted"
response = client.get(
"/async_test",
headers={
"Authorization": "Basic %s"
% base64.b64encode(b"joe@lp.com:password").decode("utf-8")
},
)
assert b"Access Granted" in response.data
def test_roles_required(app, client):
@app.route("/async_test")
@roles_required("admin")
async def async_test():
await asyncio.sleep(0)
return "Access Granted"
authenticate(client)
response = client.get("/async_test")
assert b"Access Granted" in response.data
def test_roles_accepted(app, client):
@app.route("/async_test")
@roles_accepted("admin")
async def async_test():
await asyncio.sleep(0)
return "Access Granted"
authenticate(client)
response = client.get("/async_test")
assert b"Access Granted" in response.data
def test_permissions_required(app, client):
@app.route("/async_test")
@permissions_required("super")
async def async_test():
await asyncio.sleep(0)
return "Access Granted"
authenticate(client)
response = client.get("/async_test")
assert b"Access Granted" in response.data
def test_permissions_accepted(app, client):
@app.route("/async_test")
@permissions_accepted("super")
async def async_test():
await asyncio.sleep(0)
return "Access Granted"
authenticate(client)
response = client.get("/async_test")
assert b"Access Granted" in response.data
def test_anon(app, client):
@app.route("/async_test")
@anonymous_user_required
async def async_test():
await asyncio.sleep(0)
return "Access Granted"
response = client.get("/async_test")
assert b"Access Granted" in response.data
def test_unauth_csrf(app, client):
@app.route("/async_test")
@unauth_csrf()
async def async_test():
await asyncio.sleep(0)
return "Access Granted"
response = client.get("/async_test")
assert b"Access Granted" in response.data
|
