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
|
import pytest
from .. import views
pytestmark = pytest.mark.django_db
pytest_plugins = ("pytest_asyncio",)
class TestEnqueueSuccessfulTask:
def test(self):
response = views.enqueue_successful_task(None)
assert response.status_code == 201
class TestEnqueueFailingTask:
def test(self):
response = views.enqueue_failing_task(None)
assert response.status_code == 201
class TestEnqueueNestingTask:
def test(self):
response = views.enqueue_nesting_task(None)
assert response.status_code == 201
class TestRaiseException:
def test(self):
with pytest.raises(Exception) as e:
views.raise_exception(None)
assert str(e.value) == "This is a view raising an exception."
class TestLogWithStandardLogger:
def test(self):
response = views.log_with_standard_logger(None)
assert response.status_code == 200
@pytest.mark.asyncio
class TestAsyncView:
async def test(self, mocker):
mocker.patch("asyncio.sleep")
response = await views.async_view(None)
assert response.status_code == 200
class TestRevokeTask:
def test(self):
response = views.revoke_task(None)
assert response.status_code == 201
class TestEnqueueUnknownTask:
def test(self):
response = views.enqueue_unknown_task(None)
assert response.status_code == 201
class TestEnqueueRejectedTask:
def test(self):
response = views.enqueue_rejected_task(None)
assert response.status_code == 201
@pytest.mark.asyncio
class TestAsyncStreamingViewView:
async def test(self, mocker):
response = await views.async_streaming_view(None)
assert response.status_code == 200
mocker.patch("asyncio.sleep")
assert b"0" == await anext(response.streaming_content)
assert b"1" == await anext(response.streaming_content)
assert b"2" == await anext(response.streaming_content)
assert b"3" == await anext(response.streaming_content)
assert b"4" == await anext(response.streaming_content)
with pytest.raises(StopAsyncIteration):
await anext(response.streaming_content)
class TestSyncStreamingViewView:
def test(self, mocker):
response = views.sync_streaming_view(None)
assert response.status_code == 200
mocker.patch("time.sleep")
assert b"0" == next(response.streaming_content)
assert b"1" == next(response.streaming_content)
assert b"2" == next(response.streaming_content)
assert b"3" == next(response.streaming_content)
assert b"4" == next(response.streaming_content)
with pytest.raises(StopIteration):
next(response.streaming_content)
|
