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
|
from mock import Mock
from datetime import datetime, timedelta
from celery.result import AsyncResult
import celery.states as states
from tests import AsyncHTTPTestCase
class ApplyTests(AsyncHTTPTestCase):
def test_apply(self):
from mock import patch, PropertyMock
import json
result = 'result'
with patch('celery.result.AsyncResult.state', new_callable=PropertyMock) as mock_state:
with patch('celery.result.AsyncResult.result', new_callable=PropertyMock) as mock_result:
mock_state.return_value = states.SUCCESS
mock_result.return_value = result
ar = AsyncResult(123)
ar.get = Mock(return_value=result)
task = self._app.capp.tasks['foo'] = Mock()
task.apply_async = Mock(return_value=ar)
r = self.post('/api/task/apply/foo', body='')
self.assertEqual(200, r.code)
body = bytes.decode(r.body)
self.assertEqual(result, json.loads(body)['result'])
task.apply_async.assert_called_once_with(args=[], kwargs={})
class AsyncApplyTests(AsyncHTTPTestCase):
def test_async_apply(self):
task = self._app.capp.tasks['foo'] = Mock()
task.apply_async = Mock(return_value=AsyncResult(123))
r = self.post('/api/task/async-apply/foo', body={})
self.assertEqual(200, r.code)
task.apply_async.assert_called_once_with(args=[], kwargs={})
def test_async_apply_eta(self):
task = self._app.capp.tasks['foo'] = Mock()
task.apply_async = Mock(return_value=AsyncResult(123))
tomorrow = datetime.utcnow() + timedelta(days=1)
r = self.post('/api/task/async-apply/foo',
body='{"eta": "%s"}' % tomorrow)
self.assertEqual(200, r.code)
task.apply_async.assert_called_once_with(
args=[], kwargs={}, eta=tomorrow)
def test_async_apply_countdown(self):
task = self._app.capp.tasks['foo'] = Mock()
task.apply_async = Mock(return_value=AsyncResult(123))
r = self.post('/api/task/async-apply/foo',
body='{"countdown": "3"}')
self.assertEqual(200, r.code)
task.apply_async.assert_called_once_with(
args=[], kwargs={}, countdown=3)
def test_async_apply_expires(self):
task = self._app.capp.tasks['foo'] = Mock()
task.apply_async = Mock(return_value=AsyncResult(123))
r = self.post('/api/task/async-apply/foo',
body='{"expires": "60"}')
self.assertEqual(200, r.code)
task.apply_async.assert_called_once_with(
args=[], kwargs={}, expires=60)
def test_async_apply_expires_datetime(self):
task = self._app.capp.tasks['foo'] = Mock()
task.apply_async = Mock(return_value=AsyncResult(123))
tomorrow = datetime.utcnow() + timedelta(days=1)
r = self.post('/api/task/async-apply/foo',
body='{"expires": "%s"}' % tomorrow)
self.assertEqual(200, r.code)
task.apply_async.assert_called_once_with(
args=[], kwargs={}, expires=tomorrow)
|
