/usr/lib/python2.7/dist-packages/celery/tests/compat_modules/test_compat.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 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 __future__ import absolute_import
from datetime import timedelta
import sys
sys.modules.pop('celery.task', None)
from celery.schedules import schedule # noqa
from celery.task import ( # noqa
periodic_task,
PeriodicTask
)
from celery.utils.timeutils import timedelta_seconds # noqa
from celery.tests.case import AppCase, depends_on_current_app # noqa
class test_Task(AppCase):
def test_base_task_inherits_magic_kwargs_from_app(self):
from celery.task import Task as OldTask
class timkX(OldTask):
abstract = True
with self.Celery(set_as_current=False,
accept_magic_kwargs=True) as app:
timkX.bind(app)
# see #918
self.assertFalse(timkX.accept_magic_kwargs)
from celery import Task as NewTask
class timkY(NewTask):
abstract = True
timkY.bind(app)
self.assertFalse(timkY.accept_magic_kwargs)
@depends_on_current_app
class test_periodic_tasks(AppCase):
def setup(self):
@periodic_task(app=self.app, shared=False,
run_every=schedule(timedelta(hours=1), app=self.app))
def my_periodic():
pass
self.my_periodic = my_periodic
def now(self):
return self.app.now()
def test_must_have_run_every(self):
with self.assertRaises(NotImplementedError):
type('Foo', (PeriodicTask, ), {'__module__': __name__})
def test_remaining_estimate(self):
s = self.my_periodic.run_every
self.assertIsInstance(
s.remaining_estimate(s.maybe_make_aware(self.now())),
timedelta)
def test_is_due_not_due(self):
due, remaining = self.my_periodic.run_every.is_due(self.now())
self.assertFalse(due)
# This assertion may fail if executed in the
# first minute of an hour, thus 59 instead of 60
self.assertGreater(remaining, 59)
def test_is_due(self):
p = self.my_periodic
due, remaining = p.run_every.is_due(
self.now() - p.run_every.run_every,
)
self.assertTrue(due)
self.assertEqual(remaining,
timedelta_seconds(p.run_every.run_every))
def test_schedule_repr(self):
p = self.my_periodic
self.assertTrue(repr(p.run_every))
|