__init__.py 1.1 KB

123456789101112131415161718192021222324252627282930
  1. from __future__ import with_statement
  2. from celery.tests.utils import unittest
  3. from celery import backends
  4. from celery.backends.amqp import AMQPBackend
  5. from celery.backends.cache import CacheBackend
  6. class TestBackends(unittest.TestCase):
  7. def test_get_backend_aliases(self):
  8. expects = [("amqp", AMQPBackend),
  9. ("cache", CacheBackend)]
  10. for expect_name, expect_cls in expects:
  11. self.assertIsInstance(backends.get_backend_cls(expect_name)(),
  12. expect_cls)
  13. def test_get_backend_cache(self):
  14. backends.get_backend_cls.clear()
  15. hits = backends.get_backend_cls.hits
  16. misses = backends.get_backend_cls.misses
  17. self.assertTrue(backends.get_backend_cls("amqp"))
  18. self.assertEqual(backends.get_backend_cls.misses, misses + 1)
  19. self.assertTrue(backends.get_backend_cls("amqp"))
  20. self.assertEqual(backends.get_backend_cls.hits, hits + 1)
  21. def test_unknown_backend(self):
  22. with self.assertRaises(ValueError):
  23. backends.get_backend_cls("fasodaopjeqijwqe")