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
|
import asyncio
import pytest
from ninja import NinjaAPI
from ninja.security import APIKeyQuery
from ninja.testing import TestAsyncClient
@pytest.mark.asyncio
async def test_asyncio_operations():
api = NinjaAPI()
class KeyQuery(APIKeyQuery):
def authenticate(self, request, key):
if key == "secret":
return key
@api.get("/async", auth=KeyQuery())
async def async_view(request, payload: int):
await asyncio.sleep(0)
return {"async": True}
@api.post("/async")
def sync_post_to_async_view(request):
return {"sync": True}
client = TestAsyncClient(api)
# Actual tests --------------------------------------------------
# without auth:
res = await client.get("/async?payload=1")
assert res.status_code == 401
# async successful
res = await client.get("/async?payload=1&key=secret")
assert res.json() == {"async": True}
# async innvalid input
res = await client.get("/async?payload=str&key=secret")
assert res.status_code == 422
# async call to sync method for path that have async operations
res = await client.post("/async")
assert res.json() == {"sync": True}
# invalid method
res = await client.put("/async")
assert res.status_code == 405
|
