123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263 |
- from __future__ import absolute_import, unicode_literals
- import sys
- from importlib import import_module
- from case import mock
- from celery.app.defaults import (_OLD_DEFAULTS, _OLD_SETTING_KEYS,
- _TO_NEW_KEY, _TO_OLD_KEY, DEFAULTS,
- NAMESPACES, SETTING_KEYS)
- from celery.five import values
- class test_defaults:
- def setup(self):
- self._prev = sys.modules.pop('celery.app.defaults', None)
- def teardown(self):
- if self._prev:
- sys.modules['celery.app.defaults'] = self._prev
- def test_option_repr(self):
- assert repr(NAMESPACES['broker']['url'])
- def test_any(self):
- val = object()
- assert self.defaults.Option.typemap['any'](val) is val
- @mock.sys_platform('darwin')
- @mock.pypy_version((1, 4, 0))
- def test_default_pool_pypy_14(self):
- assert self.defaults.DEFAULT_POOL == 'solo'
- @mock.sys_platform('darwin')
- @mock.pypy_version((1, 5, 0))
- def test_default_pool_pypy_15(self):
- assert self.defaults.DEFAULT_POOL == 'prefork'
- def test_compat_indices(self):
- assert not any(key.isupper() for key in DEFAULTS)
- assert not any(key.islower() for key in _OLD_DEFAULTS)
- assert not any(key.isupper() for key in _TO_OLD_KEY)
- assert not any(key.islower() for key in _TO_NEW_KEY)
- assert not any(key.isupper() for key in SETTING_KEYS)
- assert not any(key.islower() for key in _OLD_SETTING_KEYS)
- assert not any(value.isupper() for value in values(_TO_NEW_KEY))
- assert not any(value.islower() for value in values(_TO_OLD_KEY))
- for key in _TO_NEW_KEY:
- assert key in _OLD_SETTING_KEYS
- for key in _TO_OLD_KEY:
- assert key in SETTING_KEYS
- def test_find(self):
- find = self.defaults.find
- assert find('default_queue')[2].default == 'celery'
- assert find('task_default_exchange')[2] is None
- @property
- def defaults(self):
- return import_module('celery.app.defaults')
|