|
@@ -7,8 +7,6 @@ from celery import loaders
|
|
|
from celery.loaders import base
|
|
|
from celery.loaders import default
|
|
|
|
|
|
-from djcelery.loaders import djangoapp
|
|
|
-
|
|
|
from celery.tests.utils import with_environ
|
|
|
|
|
|
|
|
@@ -17,13 +15,9 @@ class TestLoaders(unittest.TestCase):
|
|
|
|
|
|
def test_get_loader_cls(self):
|
|
|
|
|
|
- self.assertEqual(loaders.get_loader_cls("django"),
|
|
|
- djangoapp.Loader)
|
|
|
self.assertEqual(loaders.get_loader_cls("default"),
|
|
|
loaders.DefaultLoader)
|
|
|
# Execute cached branch.
|
|
|
- self.assertEqual(loaders.get_loader_cls("django"),
|
|
|
- djangoapp.Loader)
|
|
|
self.assertEqual(loaders.get_loader_cls("default"),
|
|
|
loaders.DefaultLoader)
|
|
|
|
|
@@ -66,35 +60,6 @@ class TestLoaderBase(unittest.TestCase):
|
|
|
[os, sys, task])
|
|
|
|
|
|
|
|
|
-class TestDjangoLoader(unittest.TestCase):
|
|
|
-
|
|
|
- def setUp(self):
|
|
|
- self.loader = djangoapp.Loader()
|
|
|
-
|
|
|
- def test_on_worker_init(self):
|
|
|
- from django.conf import settings
|
|
|
- old_imports = getattr(settings, "CELERY_IMPORTS", None)
|
|
|
- settings.CELERY_IMPORTS = ("xxx.does.not.exist", )
|
|
|
- try:
|
|
|
- self.assertRaises(ImportError, self.loader.on_worker_init)
|
|
|
- finally:
|
|
|
- settings.CELERY_IMPORTS = old_imports
|
|
|
-
|
|
|
- def test_race_protection(self):
|
|
|
- djangoapp._RACE_PROTECTION = True
|
|
|
- try:
|
|
|
- self.assertFalse(self.loader.on_worker_init())
|
|
|
- finally:
|
|
|
- djangoapp._RACE_PROTECTION = False
|
|
|
-
|
|
|
- def test_find_related_module_no_path(self):
|
|
|
- self.assertFalse(djangoapp.find_related_module("sys", "tasks"))
|
|
|
-
|
|
|
- def test_find_related_module_no_related(self):
|
|
|
- self.assertFalse(djangoapp.find_related_module("someapp",
|
|
|
- "frobulators"))
|
|
|
-
|
|
|
-
|
|
|
def modifies_django_env(fun):
|
|
|
|
|
|
def _protected(*args, **kwargs):
|