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
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
|
# -*- coding: utf-8 -*-
from __future__ import (absolute_import, division, print_function,
unicode_literals)
import mock
from redis import Redis
from rq.decorators import job
from rq.job import Job
from rq.queue import Queue
from rq.worker import DEFAULT_RESULT_TTL
from tests import RQTestCase
from tests.fixtures import decorated_job
class TestDecorator(RQTestCase):
def setUp(self):
super(TestDecorator, self).setUp()
def test_decorator_preserves_functionality(self):
"""Ensure that a decorated function's functionality is still preserved.
"""
self.assertEqual(decorated_job(1, 2), 3)
def test_decorator_adds_delay_attr(self):
"""Ensure that decorator adds a delay attribute to function that returns
a Job instance when called.
"""
self.assertTrue(hasattr(decorated_job, 'delay'))
result = decorated_job.delay(1, 2)
self.assertTrue(isinstance(result, Job))
# Ensure that job returns the right result when performed
self.assertEqual(result.perform(), 3)
def test_decorator_accepts_queue_name_as_argument(self):
"""Ensure that passing in queue name to the decorator puts the job in
the right queue.
"""
@job(queue='queue_name')
def hello():
return 'Hi'
result = hello.delay()
self.assertEqual(result.origin, 'queue_name')
def test_decorator_accepts_result_ttl_as_argument(self):
"""Ensure that passing in result_ttl to the decorator sets the
result_ttl on the job
"""
# Ensure default
result = decorated_job.delay(1, 2)
self.assertEqual(result.result_ttl, DEFAULT_RESULT_TTL)
@job('default', result_ttl=10)
def hello():
return 'Why hello'
result = hello.delay()
self.assertEqual(result.result_ttl, 10)
def test_decorator_accepts_ttl_as_argument(self):
"""Ensure that passing in ttl to the decorator sets the ttl on the job
"""
# Ensure default
result = decorated_job.delay(1, 2)
self.assertEqual(result.ttl, None)
@job('default', ttl=30)
def hello():
return 'Hello'
result = hello.delay()
self.assertEqual(result.ttl, 30)
def test_decorator_accepts_meta_as_argument(self):
"""Ensure that passing in meta to the decorator sets the meta on the job
"""
# Ensure default
result = decorated_job.delay(1, 2)
self.assertEqual(result.meta, {})
test_meta = {
'metaKey1': 1,
'metaKey2': 2,
}
@job('default', meta=test_meta)
def hello():
return 'Hello'
result = hello.delay()
self.assertEqual(result.meta, test_meta)
def test_decorator_accepts_result_depends_on_as_argument(self):
"""Ensure that passing in depends_on to the decorator sets the
correct dependency on the job
"""
# Ensure default
result = decorated_job.delay(1, 2)
self.assertEqual(result.dependency, None)
self.assertEqual(result._dependency_id, None)
@job(queue='queue_name')
def foo():
return 'Firstly'
foo_job = foo.delay()
@job(queue='queue_name', depends_on=foo_job)
def bar():
return 'Secondly'
bar_job = bar.delay()
self.assertEqual(foo_job._dependency_ids,[])
self.assertIsNone(foo_job._dependency_id)
self.assertEqual(foo_job.dependency, None)
self.assertEqual(bar_job.dependency, foo_job)
self.assertEqual(bar_job.dependency.id, foo_job.id)
def test_decorator_delay_accepts_depends_on_as_argument(self):
"""Ensure that passing in depends_on to the delay method of
a decorated function overrides the depends_on set in the
constructor.
"""
# Ensure default
result = decorated_job.delay(1, 2)
self.assertEqual(result.dependency, None)
self.assertEqual(result._dependency_id, None)
@job(queue='queue_name')
def foo():
return 'Firstly'
@job(queue='queue_name')
def bar():
return 'Firstly'
foo_job = foo.delay()
bar_job = bar.delay()
@job(queue='queue_name', depends_on=foo_job)
def baz():
return 'Secondly'
baz_job = bar.delay(depends_on=bar_job)
self.assertIsNone(foo_job._dependency_id)
self.assertIsNone(bar_job._dependency_id)
self.assertEqual(foo_job._dependency_ids,[])
self.assertEqual(bar_job._dependency_ids,[])
self.assertEqual(baz_job._dependency_id, bar_job.id)
self.assertEqual(baz_job.dependency, bar_job)
self.assertEqual(baz_job.dependency.id, bar_job.id)
@mock.patch('rq.queue.resolve_connection')
def test_decorator_connection_laziness(self, resolve_connection):
"""Ensure that job decorator resolve connection in `lazy` way """
resolve_connection.return_value = Redis()
@job(queue='queue_name')
def foo():
return 'do something'
self.assertEqual(resolve_connection.call_count, 0)
foo()
self.assertEqual(resolve_connection.call_count, 0)
foo.delay()
self.assertEqual(resolve_connection.call_count, 1)
def test_decorator_custom_queue_class(self):
"""Ensure that a custom queue class can be passed to the job decorator"""
class CustomQueue(Queue):
pass
CustomQueue.enqueue_call = mock.MagicMock(
spec=lambda *args, **kwargs: None,
name='enqueue_call'
)
custom_decorator = job(queue='default', queue_class=CustomQueue)
self.assertIs(custom_decorator.queue_class, CustomQueue)
@custom_decorator
def custom_queue_class_job(x, y):
return x + y
custom_queue_class_job.delay(1, 2)
self.assertEqual(CustomQueue.enqueue_call.call_count, 1)
def test_decorate_custom_queue(self):
"""Ensure that a custom queue instance can be passed to the job decorator"""
class CustomQueue(Queue):
pass
CustomQueue.enqueue_call = mock.MagicMock(
spec=lambda *args, **kwargs: None,
name='enqueue_call'
)
queue = CustomQueue()
@job(queue=queue)
def custom_queue_job(x, y):
return x + y
custom_queue_job.delay(1, 2)
self.assertEqual(queue.enqueue_call.call_count, 1)
def test_decorator_custom_failure_ttl(self):
"""Ensure that passing in failure_ttl to the decorator sets the
failure_ttl on the job
"""
# Ensure default
result = decorated_job.delay(1, 2)
self.assertEqual(result.failure_ttl, None)
@job('default', failure_ttl=10)
def hello():
return 'Why hello'
result = hello.delay()
self.assertEqual(result.failure_ttl, 10)
|
