Repository navigation
Expand file tree
/
Copy pathtest_taskqueue.py
More file actions
123 lines (95 loc) · 2.93 KB
/
Copy pathtest_taskqueue.py
File metadata and controls
123 lines (95 loc) · 2.93 KB
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
import json
import os
import time
from six.moves import range
import pytest
from taskqueue import RegisteredTask, TaskQueue, MockTask, PrintTask, LocalTaskQueue
from taskqueue import QUEUE_NAME
TRAVIS_BRANCH = None if 'TRAVIS_BRANCH' not in os.environ else os.environ['TRAVIS_BRANCH']
if QUEUE_NAME != '':
pass
elif TRAVIS_BRANCH is None:
QUEUE_NAME = 'test-pull-queue'
elif TRAVIS_BRANCH == 'master':
QUEUE_NAME = 'travis-pull-queue-1'
else:
QUEUE_NAME = 'travis-pull-queue-2'
QTYPES = ('aws',)#, 'google')
QURL = 'https://sqs.us-east-1.amazonaws.com/098703261575/wms-test-pull-queue'
def test_task_creation():
task = MockTask(this="is", a=[1, 4, 2], simple={"test": "to", "check": 4},
serialization=('i', 's', 's', 'u', 'e', 's'), wow=4, with_kwargs=None)
task.wow = 5
payload = task.payload()
payload = json.dumps(payload)
task_deserialized = MockTask.deserialize(payload)
assert task_deserialized._args == {
"this": "is", "a": [1, 4, 2],
"simple": {"test": "to", "check": 4},
"serialization": ["i", "s", "s", "u", "e", "s"],
"wow": 5,
"with_kwargs": None
}
def test_get():
global QUEUE_NAME
for qtype in QTYPES:
tq = TaskQueue(n_threads=0, queue_name=QUEUE_NAME, queue_server=qtype, qurl=QURL)
n_inserts = 5
tq.purge()
for _ in range(n_inserts):
task = PrintTask()
tq.insert(task)
tq.wait()
tq.purge()
def test_single_threaded_insertion():
global QUEUE_NAME
for qtype in QTYPES:
tq = TaskQueue(n_threads=0, queue_name=QUEUE_NAME, queue_server=qtype, qurl=QURL)
tq.purge()
if qtype != 'aws':
assert tq.enqueued == 0
n_inserts = 5
for _ in range(n_inserts):
task = PrintTask()
tq.insert(task)
tq.wait()
if qtype != 'aws':
assert tq.enqueued == n_inserts
assert all(map(lambda x: type(x) == PrintTask, tq.list()))
tq.purge()
if qtype != 'aws':
assert tq.enqueued == 0
def test_multi_threaded_insertion():
global QUEUE_NAME
for qtype in QTYPES:
tq = TaskQueue(n_threads=40, queue_name=QUEUE_NAME, queue_server=qtype, qurl=QURL)
n_inserts = 100
tq.purge()
tq.wait()
if qtype != 'aws':
assert tq.enqueued == 0
for _ in range(n_inserts):
task = PrintTask()
tq.insert(task)
tq.wait()
if qtype == 'aws':
list_len = 10
else:
list_len = 100
lst = tq.list()
assert len(lst) == list_len # task list api only lists 100 items at a time
assert all(map(lambda x: type(x) == PrintTask, lst))
tq.purge()
if qtype != 'aws':
assert tq.enqueued == 0
def test_400_errors():
global QUEUE_NAME
for qtype in QTYPES:
with TaskQueue(n_threads=1, queue_name=QUEUE_NAME, queue_server=qtype, qurl=QURL) as tq:
tq.delete('nonexistent')
def test_local_taskqueue():
with LocalTaskQueue(parallel=True, progress=False) as tq:
for i in range(20000):
tq.insert(
MockTask(arg=i)
)