celery/celery/tests/backends/test_backends.py

from celery import backends
from celery.backends.rpc import RPCBackend
from celery.backends.cache import CacheBackend
from celery.exceptions import ImproperlyConfigured
from celery.tests.case import AppCase, patch


class test_backends(AppCase):

    def test_get_backend_aliases(self):
        expects = [('rpc://', RPCBackend),
                   ('cache+memory://', CacheBackend)]

        for url, expect_cls in expects:
            backend, url = backends.get_backend_by_url(url, self.app.loader)
            self.assertIsInstance(
                backend(app=self.app, url=url),
                expect_cls,
            )

    def test_unknown_backend(self):
        with self.assertRaises(ImportError):
            backends.get_backend_cls('fasodaopjeqijwqe', self.app.loader)

    def test_backend_by_url(self, url='redis://localhost/1'):
        from celery.backends.redis import RedisBackend
        backend, url_ = backends.get_backend_by_url(url, self.app.loader)
        self.assertIs(backend, RedisBackend)
        self.assertEqual(url_, url)

    def test_sym_raises_ValuError(self):
        with patch('celery.backends.symbol_by_name') as sbn:
            sbn.side_effect = ValueError()
            with self.assertRaises(ImproperlyConfigured):
                backends.get_backend_cls('xxx.xxx:foo', self.app.loader)