/usr/lib/python2.7/dist-packages/celery/tests/contrib/test_methods.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 | from __future__ import absolute_import
from celery.contrib.methods import task_method, task
from celery.tests.case import AppCase, patch
class test_task_method(AppCase):
def test_task_method(self):
class X(object):
def __init__(self):
self.state = 0
@self.app.task(shared=False, filter=task_method)
def add(self, x):
self.state += x
x = X()
x.add(2)
self.assertEqual(x.state, 2)
x.add(4)
self.assertEqual(x.state, 6)
self.assertTrue(X.add)
self.assertIs(x.add.__self__, x)
def test_task(self):
with patch('celery.contrib.methods.current_app') as curapp:
fun = object()
task(fun, x=1)
curapp.task.assert_called_with(fun, x=1, filter=task_method)
|