1
from __future__ import absolute_import
2
from __future__ import with_statement
4
from celery import backends
5
from celery.backends.amqp import AMQPBackend
6
from celery.backends.cache import CacheBackend
7
from celery.tests.utils import Case
10
class TestBackends(Case):
12
def test_get_backend_aliases(self):
13
expects = [("amqp", AMQPBackend),
14
("cache", CacheBackend)]
15
for expect_name, expect_cls in expects:
16
self.assertIsInstance(backends.get_backend_cls(expect_name)(),
19
def test_get_backend_cache(self):
20
backends.get_backend_cls.clear()
21
hits = backends.get_backend_cls.hits
22
misses = backends.get_backend_cls.misses
23
self.assertTrue(backends.get_backend_cls("amqp"))
24
self.assertEqual(backends.get_backend_cls.misses, misses + 1)
25
self.assertTrue(backends.get_backend_cls("amqp"))
26
self.assertEqual(backends.get_backend_cls.hits, hits + 1)
28
def test_unknown_backend(self):
29
with self.assertRaises(ValueError):
30
backends.get_backend_cls("fasodaopjeqijwqe")