/
test_create_task.py
112 lines (77 loc) · 2.58 KB
/
test_create_task.py
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
import asyncio
import sys
from threading import Event
from unittest.mock import Mock
import pytest
from sanic.exceptions import SanicException
from sanic.response import text
def test_create_task(app):
e = Event()
async def coro():
await asyncio.sleep(0.05)
e.set()
@app.route("/early")
def not_set(request):
return text(str(e.is_set()))
@app.route("/late")
async def set(request):
await asyncio.sleep(0.1)
return text(str(e.is_set()))
app.add_task(coro)
request, response = app.test_client.get("/early")
assert response.body == b"False"
app.signal_router.reset()
app.add_task(coro)
request, response = app.test_client.get("/late")
assert response.body == b"True"
def test_create_task_with_app_arg(app):
@app.after_server_start
async def setup_q(app, _):
app.ctx.q = asyncio.Queue()
@app.route("/")
async def not_set(request):
return text(await request.app.ctx.q.get())
async def coro(app):
await app.ctx.q.put(app.name)
app.add_task(coro)
_, response = app.test_client.get("/")
assert response.text == "test_create_task_with_app_arg"
@pytest.mark.skipif(sys.version_info < (3, 8), reason="Not supported in 3.7")
def test_create_named_task(app):
async def dummy():
...
@app.before_server_start
async def setup(app, _):
app.add_task(dummy, name="dummy_task")
@app.after_server_start
async def stop(app, _):
task = app.get_task("dummy_task")
assert app._task_registry
assert isinstance(task, asyncio.Task)
assert task.get_name() == "dummy_task"
app.stop()
app.run()
def test_named_task_called(app):
e = Event()
async def coro():
e.set()
@app.route("/")
async def isset(request):
await asyncio.sleep(0.05)
return text(str(e.is_set()))
@app.before_server_start
async def setup(app, _):
app.add_task(coro, name="dummy_task")
request, response = app.test_client.get("/")
assert response.body == b"True"
@pytest.mark.skipif(sys.version_info < (3, 8), reason="Not supported in 3.7")
def test_create_named_task_fails_outside_app(app):
async def dummy():
...
message = "Cannot name task outside of a running application"
with pytest.raises(RuntimeError, match=message):
app.add_task(dummy, name="dummy_task")
assert not app._task_registry
message = 'Registered task named "dummy_task" not found.'
with pytest.raises(SanicException, match=message):
app.get_task("dummy_task")