forked from celery/celery
-
Notifications
You must be signed in to change notification settings - Fork 40
/
__init__.py
76 lines (61 loc) · 2.3 KB
/
__init__.py
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
import logging
import os
import sys
from importlib import import_module
config_module = os.environ.setdefault("CELERY_TEST_CONFIG_MODULE",
"celery.tests.config")
os.environ["CELERY_CONFIG_MODULE"] = config_module
os.environ["CELERY_LOADER"] = "default"
os.environ["EVENTLET_NOPATCH"] = "yes"
os.environ["GEVENT_NOPATCH"] = "yes"
try:
WindowsError = WindowsError
except NameError:
class WindowsError(Exception):
pass
def teardown():
# Don't want SUBDEBUG log messages at finalization.
try:
from multiprocessing.util import get_logger
except ImportError:
pass
else:
get_logger().setLevel(logging.WARNING)
# Make sure test database is removed.
import os
if os.path.exists("test.db"):
try:
os.remove("test.db")
except WindowsError:
pass
# Make sure there are no remaining threads at shutdown.
import threading
remaining_threads = [thread for thread in threading.enumerate()
if thread.getName() != "MainThread"]
if remaining_threads:
sys.stderr.write(
"\n\n**WARNING**: Remaining threads at teardown: %r...\n" % (
remaining_threads))
def find_distribution_modules(name=__name__, file=__file__):
current_dist_depth = len(name.split(".")) - 1
current_dist = os.path.join(os.path.dirname(file),
*([os.pardir] * current_dist_depth))
abs = os.path.abspath(current_dist)
dist_name = os.path.basename(abs)
for dirpath, dirnames, filenames in os.walk(abs):
package = (dist_name + dirpath[len(abs):]).replace("/", ".")
if "__init__.py" in filenames:
yield package
for filename in filenames:
if filename.endswith(".py") and filename != "__init__.py":
yield ".".join([package, filename])[:-3]
def import_all_modules(name=__name__, file=__file__,
skip=["celery.decorators", "celery.contrib.batches"]):
for module in find_distribution_modules(name, file):
if module not in skip:
try:
import_module(module)
except ImportError:
pass
if os.environ.get("COVER_ALL_MODULES") or "--with-coverage3" in sys.argv:
import_all_modules()