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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
|
import os
import platform
import sys
from unittest.mock import Mock, patch
import pytest
from celery.utils.imports import (NotAPackage, cwd_in_path, find_module, gen_task_name, module_file, qualname,
reload_from_cwd)
def test_find_module():
def imp_side_effect(module):
if module == 'foo':
return None
else:
raise ImportError(module)
assert find_module('celery')
imp = Mock()
imp.side_effect = imp_side_effect
with pytest.raises(NotAPackage) as exc_info:
find_module('foo.bar.baz', imp=imp)
assert exc_info.value.args[0] == 'foo'
assert find_module('celery.worker.request')
def test_find_module_legacy_namespace_package(tmp_path, monkeypatch):
monkeypatch.chdir(str(tmp_path))
(tmp_path / 'pkg' / 'foo').mkdir(parents=True)
(tmp_path / 'pkg' / '__init__.py').write_text(
'from pkgutil import extend_path\n'
'__path__ = extend_path(__path__, __name__)\n')
(tmp_path / 'pkg' / 'foo' / '__init__.py').write_text('')
(tmp_path / 'pkg' / 'foo' / 'bar.py').write_text('')
with patch.dict(sys.modules):
for modname in list(sys.modules):
if modname == 'pkg' or modname.startswith('pkg.'):
del sys.modules[modname]
with pytest.raises(ImportError):
find_module('pkg.missing')
with pytest.raises(ImportError):
find_module('pkg.foo.missing')
assert find_module('pkg.foo.bar')
with pytest.raises(NotAPackage) as exc_info:
find_module('pkg.foo.bar.missing')
assert exc_info.value.args[0] == 'pkg.foo.bar'
def test_find_module_pep420_namespace_package(tmp_path, monkeypatch):
monkeypatch.chdir(str(tmp_path))
(tmp_path / 'pkg' / 'foo').mkdir(parents=True)
(tmp_path / 'pkg' / 'foo' / '__init__.py').write_text('')
(tmp_path / 'pkg' / 'foo' / 'bar.py').write_text('')
with patch.dict(sys.modules):
for modname in list(sys.modules):
if modname == 'pkg' or modname.startswith('pkg.'):
del sys.modules[modname]
with pytest.raises(ImportError):
find_module('pkg.missing')
with pytest.raises(ImportError):
find_module('pkg.foo.missing')
assert find_module('pkg.foo.bar')
with pytest.raises(NotAPackage) as exc_info:
find_module('pkg.foo.bar.missing')
assert exc_info.value.args[0] == 'pkg.foo.bar'
def test_qualname():
Class = type('Fox', (object,), {
'__module__': 'quick.brown',
})
assert qualname(Class) == 'quick.brown.Fox'
assert qualname(Class()) == 'quick.brown.Fox'
def test_reload_from_cwd(patching):
reload = patching('celery.utils.imports.reload')
reload_from_cwd('foo')
reload.assert_called()
def test_reload_from_cwd_custom_reloader():
reload = Mock()
reload_from_cwd('foo', reload)
reload.assert_called()
def test_module_file():
m1 = Mock()
m1.__file__ = '/opt/foo/xyz.pyc'
assert module_file(m1) == '/opt/foo/xyz.py'
m2 = Mock()
m2.__file__ = '/opt/foo/xyz.py'
assert module_file(m1) == '/opt/foo/xyz.py'
def test_cwd_in_path(tmp_path, monkeypatch):
now_cwd = os.getcwd()
t = str(tmp_path) + "/foo"
os.mkdir(t)
os.chdir(t)
with cwd_in_path():
assert os.path.exists(t) is True
if sys.platform == "win32" or "Windows" in platform.platform():
# If it is a Windows server, other processes cannot delete the current working directory being used by celery
# . If you want to delete it, you need to terminate the celery process. If it is a Linux server, the current
# working directory of celery can be deleted by other processes.
pass
else:
os.rmdir(t)
with cwd_in_path():
assert os.path.exists(t) is False
os.chdir(now_cwd)
class test_gen_task_name:
def test_no_module(self):
app = Mock()
app.name == '__main__'
assert gen_task_name(app, 'foo', 'axsadaewe')
|
