test_defaults.py 1.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455
  1. from __future__ import absolute_import
  2. from __future__ import with_statement
  3. import sys
  4. from importlib import import_module
  5. from mock import Mock, patch
  6. from celery.tests.utils import Case, pypy_version, sys_platform
  7. class test_defaults(Case):
  8. def setUp(self):
  9. self._prev = sys.modules.pop("celery.app.defaults", None)
  10. def tearDown(self):
  11. if self._prev:
  12. sys.modules["celery.app.defaults"] = self._prev
  13. def test_any(self):
  14. val = object()
  15. self.assertIs(self.defaults.Option.typemap["any"](val), val)
  16. def test_default_pool_pypy_14(self):
  17. with sys_platform("darwin"):
  18. with pypy_version((1, 4, 0)):
  19. self.assertEqual(self.defaults.DEFAULT_POOL, "solo")
  20. def test_default_pool_pypy_15(self):
  21. with sys_platform("darwin"):
  22. with pypy_version((1, 5, 0)):
  23. self.assertEqual(self.defaults.DEFAULT_POOL, "processes")
  24. def test_deprecated(self):
  25. source = Mock()
  26. source.BROKER_INSIST = True
  27. with patch("celery.utils.warn_deprecated") as warn:
  28. self.defaults.find_deprecated_settings(source)
  29. self.assertTrue(warn.called)
  30. def test_default_pool_jython(self):
  31. with sys_platform("java 1.6.51"):
  32. self.assertEqual(self.defaults.DEFAULT_POOL, "threads")
  33. def test_find(self):
  34. find = self.defaults.find
  35. self.assertEqual(find("server_email")[2].default, "celery@localhost")
  36. self.assertEqual(find("default_queue")[2].default, "celery")
  37. self.assertEqual(find("celery_default_exchange")[2], "celery")
  38. @property
  39. def defaults(self):
  40. return import_module("celery.app.defaults")