1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
|
import sys
from importlib import import_module
from celery.app.defaults import (_OLD_DEFAULTS, _OLD_SETTING_KEYS, _TO_NEW_KEY, _TO_OLD_KEY, DEFAULTS, NAMESPACES,
SETTING_KEYS)
class test_defaults:
def setup_method(self):
self._prev = sys.modules.pop('celery.app.defaults', None)
def teardown_method(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
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 _TO_NEW_KEY.values())
assert not any(value.islower() for value in _TO_OLD_KEY.values())
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')
|
