from __future__ import absolute_import import celery from celery.app.task import Task as ModernTask from celery.task.base import Task as CompatTask from celery.tests.case import AppCase, depends_on_current_app @depends_on_current_app class test_MagicModule(AppCase): def test_class_property_set_without_type(self): self.assertTrue(ModernTask.__dict__['app'].__get__(CompatTask())) def test_class_property_set_on_class(self): self.assertIs(ModernTask.__dict__['app'].__set__(None, None), ModernTask.__dict__['app']) def test_class_property_set(self): class X(CompatTask): pass ModernTask.__dict__['app'].__set__(X(), self.app) self.assertIs(X.app, self.app) def test_dir(self): self.assertTrue(dir(celery.messaging)) def test_direct(self): self.assertTrue(celery.task) def test_app_attrs(self): self.assertEqual(celery.task.control.broadcast, celery.current_app.control.broadcast) def test_decorators_task(self): @celery.decorators.task def _test_decorators_task(): pass self.assertTrue(_test_decorators_task.accept_magic_kwargs) def test_decorators_periodic_task(self): @celery.decorators.periodic_task(run_every=3600) def _test_decorators_ptask(): pass self.assertTrue(_test_decorators_ptask.accept_magic_kwargs)