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
|
import asyncio
import pytest
from ninja import NinjaAPI
from ninja.security import APIKeyQuery, HttpBearer
from ninja.testing import TestAsyncClient, TestClient
@pytest.mark.asyncio
async def test_async_view_handles_async_auth_func():
api = NinjaAPI()
async def auth(request):
key = request.GET.get("key")
if key == "secret":
return key
@api.get("/async", auth=auth)
async def view(request):
await asyncio.sleep(0)
return {"key": request.auth}
client = TestAsyncClient(api)
# Actual tests --------------------------------------------------
# without auth:
res = await client.get("/async")
assert res.status_code == 401
# async successful
res = await client.get("/async?key=secret")
assert res.json() == {"key": "secret"}
@pytest.mark.asyncio
async def test_async_view_handles_async_auth_cls():
api = NinjaAPI()
class Auth:
async def __call__(self, request):
key = request.GET.get("key")
if key == "secret":
return key
@api.get("/async", auth=Auth())
async def view(request):
await asyncio.sleep(0)
return {"key": request.auth}
client = TestAsyncClient(api)
# Actual tests --------------------------------------------------
# without auth:
res = await client.get("/async")
assert res.status_code == 401
# async successful
res = await client.get("/async?key=secret")
assert res.json() == {"key": "secret"}
@pytest.mark.asyncio
async def test_async_view_handles_multi_auth():
api = NinjaAPI()
def auth_1(request):
return None
async def auth_2(request):
return None
async def auth_3(request):
key = request.GET.get("key")
if key == "secret":
return key
@api.get("/async", auth=[auth_1, auth_2, auth_3])
async def view(request):
await asyncio.sleep(0)
return {"key": request.auth}
client = TestAsyncClient(api)
res = await client.get("/async?key=secret")
assert res.json() == {"key": "secret"}
@pytest.mark.asyncio
async def test_async_view_handles_auth_errors():
api = NinjaAPI()
async def auth(request):
raise Exception("boom")
@api.get("/async", auth=auth)
async def view(request):
await asyncio.sleep(0)
return {"key": request.auth}
@api.exception_handler(Exception)
def on_custom_error(request, exc):
return api.create_response(request, {"custom": True}, status=401)
client = TestAsyncClient(api)
res = await client.get("/async?key=secret")
assert res.json() == {"custom": True}
@pytest.mark.asyncio
async def test_sync_authenticate_method():
class KeyAuth(APIKeyQuery):
async def authenticate(self, request, key):
await asyncio.sleep(0)
if key == "secret":
return key
api = NinjaAPI(auth=KeyAuth())
@api.get("/async")
async def async_view(request):
return {"auth": request.auth}
client = TestAsyncClient(api)
res = await client.get("/async") # NO key
assert res.json() == {"detail": "Unauthorized"}
res = await client.get("/async?key=secret")
assert res.json() == {"auth": "secret"}
def test_async_authenticate_method_in_sync_context():
class KeyAuth(APIKeyQuery):
async def authenticate(self, request, key):
await asyncio.sleep(0)
if key == "secret":
return key
api = NinjaAPI(auth=KeyAuth())
@api.get("/sync")
def sync_view(request):
return {"auth": request.auth}
client = TestClient(api)
res = client.get("/sync") # NO key
assert res.json() == {"detail": "Unauthorized"}
res = client.get("/sync?key=secret")
assert res.json() == {"auth": "secret"}
@pytest.mark.asyncio
async def test_async_with_bearer():
class BearerAuth(HttpBearer):
async def authenticate(self, request, key):
await asyncio.sleep(0)
if key == "secret":
return key
api = NinjaAPI(auth=BearerAuth())
@api.get("/async")
async def async_view(request):
return {"auth": request.auth}
client = TestAsyncClient(api)
res = await client.get("/async") # NO key
assert res.json() == {"detail": "Unauthorized"}
res = await client.get("/async", headers={"Authorization": "Bearer secret"})
assert res.json() == {"auth": "secret"}
|
