/usr/lib/python2.7/dist-packages/celery/tests/contrib/test_abortable.py is in python-celery 3.1.20-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | from __future__ import absolute_import
from celery.contrib.abortable import AbortableTask, AbortableAsyncResult
from celery.tests.case import AppCase
class test_AbortableTask(AppCase):
def setup(self):
@self.app.task(base=AbortableTask, shared=False)
def abortable():
return True
self.abortable = abortable
def test_async_result_is_abortable(self):
result = self.abortable.apply_async()
tid = result.id
self.assertIsInstance(
self.abortable.AsyncResult(tid), AbortableAsyncResult,
)
def test_is_not_aborted(self):
self.abortable.push_request()
try:
result = self.abortable.apply_async()
tid = result.id
self.assertFalse(self.abortable.is_aborted(task_id=tid))
finally:
self.abortable.pop_request()
def test_is_aborted_not_abort_result(self):
self.abortable.AsyncResult = self.app.AsyncResult
self.abortable.push_request()
try:
self.abortable.request.id = 'foo'
self.assertFalse(self.abortable.is_aborted())
finally:
self.abortable.pop_request()
def test_abort_yields_aborted(self):
self.abortable.push_request()
try:
result = self.abortable.apply_async()
result.abort()
tid = result.id
self.assertTrue(self.abortable.is_aborted(task_id=tid))
finally:
self.abortable.pop_request()
|