子进程支持¶
通常,coverage 通过一个相当标准的 atexit 处理程序写入数据。但是,如果子进程没有自行退出,则 atexit 处理程序可能不会运行。为什么会发生这种情况,最好留给那些喜欢冒险的人通过在 Python bug tracker 中摸索来发现。
pytest-cov 支持子进程,并解决了这些 atexit 限制。但是,有一些需要注意的陷阱。
但首先,pytest-cov 的子进程支持是如何工作的?
pytest-cov 打包会在安装中注入一个 pytest-cov.pth。此文件实际上在*每次* Python 启动时都会运行以下操作
if 'COV_CORE_SOURCE' in os.environ:
try:
from pytest_cov.embed import init
init()
except Exception as exc:
sys.stderr.write(
"pytest-cov: Failed to setup subprocess coverage. "
"Environ: {0!r} "
"Exception: {1!r}\n".format(
dict((k, v) for k, v in os.environ.items() if k.startswith('COV_CORE')),
exc
)
)
pytest 插件将设置此 COV_CORE_SOURCE
环境变量,因此任何继承环境变量的子进程(默认行为)都将运行 pytest_cov.embed.init
,后者又根据这些变量设置 coverage
COV_CORE_SOURCE
COV_CORE_CONFIG
COV_CORE_DATAFILE
COV_CORE_BRANCH
COV_CORE_CONTEXT
你可能想知道为什么它有 COV_CORE
?嗯,这主要出于历史原因:很久以前,pytest-cov 依赖于一个 cov-core 包,该包为 pytest-cov、nose-cov 和 nose2-cov 实现通用功能。该依赖项已消失,但约定保留了下来。它可以更改,但会破坏所有手动设置这些原本打算为内部但实际上并非如此的环境变量的项目。
Coverage 的子进程支持¶
既然您了解了 pytest-cov 的工作原理,您就可以轻松地发现使用 coverage 推荐的 处理子进程的方式,无论是将其放在 .pth
文件中还是 sitecustomize.py
中,都会破坏一切
import coverage; coverage.process_startup() # this will break pytest-cov
不要这样做,因为这会使用错误的选项重新启动 coverage。
如果您使用 multiprocessing
¶
pytest-cov 4.0 中放弃了对 multiprocessing 的内置支持。此支持大部分都能正常工作,但在某些情况下却非常糟糕(请参阅 issue 82408),并导致测试套件非常不稳定且速度缓慢。
但是,coverage 中有内置的多处理支持,您可以迁移到该支持。您只需要在首选的配置文件(例如 .coveragerc
)中添加以下内容即可
[run]
concurrency = multiprocessing
parallel = true
sigterm = true
现在作为旁注,通常最好使用 Pool.join()
正确关闭您的 Pool。
from multiprocessing import Pool
def f(x):
return x*x
if __name__ == '__main__':
p = Pool(5)
try:
print(p.map(f, [1, 2, 3]))
finally:
p.close() # Marks the pool as closed.
p.join() # Waits for workers to exit.
信号处理程序¶
pytest-cov 提供了信号处理例程,主要用于您具有自定义信号处理且不允许 atexit 正确运行以及现已消失的多处理支持的特殊情况
pytest_cov.embed.cleanup_on_sigterm()
pytest_cov.embed.cleanup_on_signal(signum)
(例如:cleanup_on_signal(signal.SIGHUP)
)
如果您使用 multiprocessing¶
不建议将这些信号处理程序与 multiprocessing 一起使用,因为注册信号处理程序会导致池中出现死锁,请参阅:https://bugs.python.org/issue38227)。
如果您有自定义信号处理¶
pytest-cov 2.6 有一个基本的 pytest_cov.embed.cleanup_on_sigterm
,您可以使用它来注册一个 SIGTERM 处理程序,该处理程序会刷新 coverage 数据。
pytest-cov 2.7 添加了一个 pytest_cov.embed.cleanup_on_signal
函数,并更改了实现以使其更健壮:处理程序将调用先前的处理程序(如果您之前注册了任何处理程序),并且是可重入的(如果在处理程序运行时传递,则会延迟额外的信号)。
例如,如果您在 SIGHUP 上重新加载,则应执行以下操作
import os
import signal
def restart_service(frame, signum):
os.exec( ... ) # or whatever your custom signal would do
signal.signal(signal.SIGHUP, restart_service)
try:
from pytest_cov.embed import cleanup_on_signal
except ImportError:
pass
else:
cleanup_on_signal(signal.SIGHUP)
请注意,cleanup_on_signal
和 cleanup_on_sigterm
都会运行先前的信号处理程序。
或者,您可以执行以下操作
import os
import signal
try:
from pytest_cov.embed import cleanup
except ImportError:
cleanup = None
def restart_service(frame, signum):
if cleanup is not None:
cleanup()
os.exec( ... ) # or whatever your custom signal would do
signal.signal(signal.SIGHUP, restart_service)
如果您使用 Windows¶
在 Windows 上,您可以为 SIGTERM 注册一个处理程序,但它实际上不起作用。如果您 os.kill(os.getpid(), signal.SIGTERM)(将 SIGTERM 发送到当前进程),它将起作用,但对于大多数目的而言,这完全没用。
因此,这意味着如果您使用 multiprocessing,则别无选择,只能使用上面描述的 close/join 模式。使用上下文管理器 API 或 terminate 将不起作用,因为它依赖于 SIGTERM。
但是,您可以为 SIGBREAK 创建一个有效的处理程序(有一些注意事项)
import os
import signal
def shutdown(frame, signum):
# your app's shutdown or whatever
signal.signal(signal.SIGBREAK, shutdown)
try:
from pytest_cov.embed import cleanup_on_signal
except ImportError:
pass
else:
cleanup_on_signal(signal.SIGBREAK)
这些 注意事项 大致如下
您需要传递
signal.CTRL_BREAK_EVENT
它会传递给整个进程组,这可能会产生意想不到的后果