Coverage for /usr/local/lib/python3.7/site-packages/pytest_cov/embed.py : 8%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Activate coverage at python startup if appropriate.
3The python site initialisation will ensure that anything we import
4will be removed and not visible at the end of python startup. However
5we minimise all work by putting these init actions in this separate
6module and only importing what is needed when needed.
8For normal python startup when coverage should not be activated the pth
9file checks a single env var and does not import or call the init fn
10here.
12For python startup when an ancestor process has set the env indicating
13that code coverage is being collected we activate coverage based on
14info passed via env vars.
15"""
16import atexit
17import os
18import signal
20_active_cov = None
23def multiprocessing_start(_):
24 global _active_cov
25 cov = init()
26 if cov:
27 _active_cov = cov
28 multiprocessing.util.Finalize(None, cleanup, exitpriority=1000)
31try:
32 import multiprocessing.util
33except ImportError:
34 pass
35else:
36 multiprocessing.util.register_after_fork(multiprocessing_start, multiprocessing_start)
39def init():
40 # Only continue if ancestor process has set everything needed in
41 # the env.
42 global _active_cov
44 cov_source = os.environ.get('COV_CORE_SOURCE')
45 cov_config = os.environ.get('COV_CORE_CONFIG')
46 cov_datafile = os.environ.get('COV_CORE_DATAFILE')
47 cov_branch = True if os.environ.get('COV_CORE_BRANCH') == 'enabled' else None
49 if cov_datafile:
50 if _active_cov:
51 cleanup()
52 # Import what we need to activate coverage.
53 import coverage
55 # Determine all source roots.
56 if cov_source in os.pathsep:
57 cov_source = None
58 else:
59 cov_source = cov_source.split(os.pathsep)
60 if cov_config == os.pathsep:
61 cov_config = True
63 # Activate coverage for this process.
64 cov = _active_cov = coverage.Coverage(
65 source=cov_source,
66 branch=cov_branch,
67 data_suffix=True,
68 config_file=cov_config,
69 auto_data=True,
70 data_file=cov_datafile
71 )
72 cov.load()
73 cov.start()
74 cov._warn_no_data = False
75 cov._warn_unimported_source = False
76 return cov
79def _cleanup(cov):
80 if cov is not None:
81 cov.stop()
82 cov.save()
83 cov._auto_save = False # prevent autosaving from cov._atexit in case the interpreter lacks atexit.unregister
84 try:
85 atexit.unregister(cov._atexit)
86 except Exception:
87 pass
90def cleanup():
91 global _active_cov
92 global _cleanup_in_progress
93 global _pending_signal
95 _cleanup_in_progress = True
96 _cleanup(_active_cov)
97 _active_cov = None
98 _cleanup_in_progress = False
99 if _pending_signal:
100 pending_singal = _pending_signal
101 _pending_signal = None
102 _signal_cleanup_handler(*pending_singal)
105multiprocessing_finish = cleanup # in case someone dared to use this internal
107_previous_handlers = {}
108_pending_signal = None
109_cleanup_in_progress = False
112def _signal_cleanup_handler(signum, frame):
113 global _pending_signal
114 if _cleanup_in_progress:
115 _pending_signal = signum, frame
116 return
117 cleanup()
118 _previous_handler = _previous_handlers.get(signum)
119 if _previous_handler == signal.SIG_IGN:
120 return
121 elif _previous_handler and _previous_handler is not _signal_cleanup_handler:
122 _previous_handler(signum, frame)
123 elif signum == signal.SIGTERM:
124 os._exit(128 + signum)
125 elif signum == signal.SIGINT:
126 raise KeyboardInterrupt()
129def cleanup_on_signal(signum):
130 previous = signal.getsignal(signum)
131 if previous is not _signal_cleanup_handler:
132 _previous_handlers[signum] = previous
133 signal.signal(signum, _signal_cleanup_handler)
136def cleanup_on_sigterm():
137 cleanup_on_signal(signal.SIGTERM)