概述
官方使用介绍
直接从官方文档中摘抄，内容参考如下：
1. You can set a global timeout in the `pytest configuration file`__
using the ``timeout`` option. E.g.:
.. code:: ini
[pytest]
timeout = 300
2. The ``PYTEST_TIMEOUT`` environment variable sets a global timeout
overriding a possible value in the configuration file.
3. The ``--timeout`` command line option sets a global timeout
overriding both the environment variable and configuration option.
4. Using the ``timeout`` marker_ on test items you can specify
timeouts on a per-item basis:
.. code:: python
@pytest.mark.timeout(300)
def test_foo():
pass
总结一下：
1.写入pytest.ini文件
2.使用环境变量PYTEST_TIMEOUT
3.pytest执行时指定参数--timeout
4.测试用例中使用timeout标记
源码解读
源码目录结构
从官网下载解压到本地：
root@Gavin:~/pytest_plugin# cd pytest-timeout-2.3.1/
root@Gavin:~/pytest_plugin/pytest-timeout-2.3.1# ll
total 116
drwxr-xr-x 3 gavin gavin 4096 Mar 8 05:03 ./
drwxr-xr-x 24 root root 4096 Jul 25 10:30 ../
-rw-r--r-- 1 gavin gavin 681 Jul 19 2021 failure_demo.py
-rw-r--r-- 1 gavin gavin 1085 Nov 21 2019 LICENSE
-rw-r--r-- 1 gavin gavin 107 Nov 21 2019 MANIFEST.in
-rw-r--r-- 1 gavin gavin 20008 Mar 8 05:03 PKG-INFO
-rw-r--r-- 1 gavin gavin 81 Oct 8 2023 pyproject.toml
drwxr-xr-x 2 gavin gavin 4096 Mar 8 05:03 pytest_timeout.egg-info/
-rw-r--r-- 1 gavin gavin 18801 Mar 8 04:51 pytest_timeout.py
-rw-r--r-- 1 gavin gavin 18738 Mar 8 05:02 README.rst
-rw-r--r-- 1 gavin gavin 1236 Mar 8 05:03 setup.cfg
-rw-r--r-- 1 gavin gavin 121 Oct 8 2023 setup.py
-rw-r--r-- 1 gavin gavin 15989 Mar 8 04:51 test_pytest_timeout.py
-rw-r--r-- 1 gavin gavin 428 Feb 9 03:02 tox.ini
root@Gavin:~/pytest_plugin/pytest-timeout-2.3.1#
核心文件是pytest_timeout.py.
核心文件代码解读
变量声明相关
__all__ = ("is_debugging", "Settings")
SESSION_TIMEOUT_KEY = pytest.StashKey[float]()
SESSION_EXPIRE_KEY = pytest.StashKey[float]()
HAVE_SIGALRM = hasattr(signal, "SIGALRM")
if HAVE_SIGALRM:
DEFAULT_METHOD = "signal"
else:
DEFAULT_METHOD = "thread"
TIMEOUT_DESC = """
Timeout in seconds before dumping the stacks. Default is 0 which
means no timeout.
""".strip()
METHOD_DESC = """
Timeout mechanism to use. 'signal' uses SIGALRM, 'thread' uses a timer
thread. If unspecified 'signal' is used on platforms which support
SIGALRM, otherwise 'thread' is used.
""".strip()
FUNC_ONLY_DESC = """
When set to True, defers the timeout evaluation to only the test
function body, ignoring the time it takes when evaluating any fixtures
used in the test.
""".strip()
DISABLE_DEBUGGER_DETECTION_DESC = """
When specified, disables debugger detection. breakpoint(), pdb.set_trace(), etc.
will be interrupted by the timeout.
""".strip()
SESSION_TIMEOUT_DESC = """
Timeout in seconds for entire session. Default is None which
means no timeout. Timeout is checked between tests, and will not interrupt a test
in progress.
""".strip()
# bdb covers pdb, ipdb, and possibly others
# pydevd covers PyCharm, VSCode, and possibly others
KNOWN_DEBUGGING_MODULES = {"pydevd", "bdb", "pydevd_frame_evaluator"}
Settings = namedtuple(
"Settings", ["timeout", "method", "func_only", "disable_debugger_detection"]
)
这段代码用于配置和处理与测试相关的超时设置，特别是与 pytest 集成的某种测试超时插件或工具。
-
导出的名称:
__all__ = ("is_debugging", "Settings")这个模块导出了两个名称：
is_debugging和Settings，这可用于在导入时明确模块提供的功能。 -
会话超时和过期的 StashKey:
SESSION_TIMEOUT_KEY = pytest.StashKey[float]() SESSION_EXPIRE_KEY = pytest.StashKey[float]()这里定义了两个
StashKey，分别用于存储会话超时和会话过期时间的键。pytest.StashKey可能是pytest用于存储测试相关数据的机制。 -
检查 SIGALRM 信号:
HAVE_SIGALRM = hasattr(signal, "SIGALRM")检查
Python的signal模块是否定义了SIGALRM信号。 -
默认方法设置:
if HAVE_SIGALRM: DEFAULT_METHOD = "signal" else: DEFAULT_METHOD = "thread"根据是否支持
SIGALRM信号，设置默认的超时机制。如果支持，则默认使用信号（"signal"），否则使用线程（"thread"）。 -
描述文本:
TIMEOUT_DESC = """ ... """.strip() METHOD_DESC = """ ... """.strip() FUNC_ONLY_DESC = """ ... """.strip() DISABLE_DEBUGGER_DETECTION_DESC = """ ... """.strip() SESSION_TIMEOUT_DESC = """ ... """.strip()定义了多段描述性文本，解释了超时设置、机制选择、函数仅超时评估、禁用调试器检测和会话超时等配置项的含义。
-
已知的调试模块:
KNOWN_DEBUGGING_MODULES = {"pydevd", "bdb", "pydevd_frame_evaluator"}定义了一个集合，包含了已知的调试器模块名称，如
pydevd（通常与 PyCharm IDE 相关）、bdb和pydevd_frame_evaluator。 -
设置命名元组:
Settings = namedtuple("Settings", ["timeout", "method", "func_only", "disable_debugger_detection"])使用
collections.namedtuple创建了一个名为Settings的命名元组，用于存储超时设置，包括：timeout：超时时间。method：超时机制，可以是"signal"或"thread"。func_only：是否仅在测试函数体中计算超时时间，忽略fixtures执行时间。disable_debugger_detection：是否禁用调试器检测。
参数相关
@pytest.hookimpl
def pytest_addoption(parser):
"""Add options to control the timeout plugin."""
group = parser.getgroup(
"timeout",
"Interrupt test run and dump stacks of all threads after a test times out",
)
group.addoption("--timeout", type=float, help=TIMEOUT_DESC)
group.addoption(
"--timeout_method",
action="store",
choices=["signal", "thread"],
help="Deprecated, use --timeout-method",
)
group.addoption(
"--timeout-method",
dest="timeout_method",
action="store",
choices=["signal", "thread"],
help=METHOD_DESC,
)
group.addoption(
"--timeout-disable-debugger-detection",
dest="timeout_disable_debugger_detection",
action="store_true",
help=DISABLE_DEBUGGER_DETECTION_DESC,
)
group.addoption(
"--session-timeout",
action="store",
dest="session_timeout",
default=None,
type=float,
metavar="SECONDS",
help=SESSION_TIMEOUT_DESC,
)
parser.addini("timeout", TIMEOUT_DESC)
parser.addini("timeout_method", METHOD_DESC)
parser.addini("timeout_func_only", FUNC_ONLY_DESC, type="bool", default=False)
parser.addini(
"timeout_disable_debugger_detection",
DISABLE_DEBUGGER_DETECTION_DESC,
type="bool",
default=False,
)
parser.addini("session_timeout", SESSION_TIMEOUT_DESC)
这个函数的主要作用是为 pytest 添加了丰富的配置选项，允许用户通过命令行或配置文件来控制测试的超时行为，包括超时时间、超时机制、调试器检测等，这为用户提供了灵活的测试控制能力。
-
钩子装饰器:
@pytest.hookimpl
这行代码使用 pytest.hookimpl 装饰器，将下面的函数注册为 pytest 的钩子。
-
函数定义:
def pytest_addoption(parser):
定义了 pytest_addoption 函数，它接收一个参数 parser，这是 pytest 的命令行解析器。
-
创建选项组:
group = parser.getgroup( "timeout", "Interrupt test run and dump stacks of all threads after a test times out", )
使用 parser.getgroup 方法创建或获取一个名为 “timeout” 的选项组，并为其提供一个描述。
-
添加命令行选项:
group.addoption("--timeout", type=float, help=TIMEOUT_DESC): 添加一个--timeout命令行选项，用于设置测试超时时间（以秒为单位）。这个选项的类型是浮点数，提供了帮助描述。
-
添加弃用的命令行选项:
group.addoption("--timeout_method", ... help="Deprecated, use --timeout-method"): 添加一个--timeout_method命令行选项，但指出它已被弃用，建议使用--timeout-method。
-
添加新的命令行选项:
group.addoption("--timeout-method", ... help=METHOD_DESC): 添加一个--timeout-method命令行选项，用于设置超时机制（使用信号或线程）。选项的值必须在choices中指定的选项中选择。
-
添加调试器检测禁用选项:
group.addoption("--timeout-disable-debugger-detection", ... help=DISABLE_DEBUGGER_DETECTION_DESC): 添加一个命令行选项，用于禁用调试器检测。
-
添加会话超时选项:
group.addoption("--session-timeout", ... help=SESSION_TIMEOUT_DESC): 添加一个--session-timeout命令行选项，用于设置整个测试会话的超时时间。
-
添加配置文件选项:
parser.addini("timeout", TIMEOUT_DESC): 添加一个配置文件选项，允许用户在pytest的配置文件中设置测试超时时间。parser.addini("timeout_method", METHOD_DESC): 添加一个配置文件选项，用于设置超时机制。parser.addini("timeout_func_only", FUNC_ONLY_DESC, type="bool", default=False): 添加一个配置文件选项，用于设置是否仅在测试函数体中计算超时时间。parser.addini("timeout_disable_debugger_detection", ... default=False): 添加一个配置文件选项，用于设置是否禁用调试器检测。parser.addini("session_timeout", SESSION_TIMEOUT_DESC): 添加一个配置文件选项，用于设置会话超时时间。
class TimeoutHooks
class TimeoutHooks:
"""Timeout specific hooks."""
@pytest.hookspec(firstresult=True)
def pytest_timeout_set_timer(item, settings):
"""Called at timeout setup.
'item' is a pytest node to setup timeout for.
Can be overridden by plugins for alternative timeout implementation strategies.
"""
@pytest.hookspec(firstresult=True)
def pytest_timeout_cancel_timer(item):
"""Called at timeout teardown.
'item' is a pytest node which was used for timeout setup.
Can be overridden by plugins for alternative timeout implementation strategies.
def pytest_addhooks(pluginmanager):
"""Register timeout-specific hooks."""
pluginmanager.add_hookspecs(TimeoutHooks)
TimeoutHooks类包含了与 pytest 超时相关的钩子（hooks），这些钩子允许插件或测试框架的其他部分介入测试用例的超时设置和取消逻辑。
-
类定义:
class TimeoutHooks:定义了一个名为
TimeoutHooks的类，用于封装与超时相关的钩子。 -
pytest_timeout_set_timer 钩子:
@pytest.hookspec(firstresult=True) def pytest_timeout_set_timer(item, settings):- 使用
@pytest.hookspec装饰器注册一个钩子，firstresult=True参数表示pytest会使用第一个插件返回的结果。 - 定义了
pytest_timeout_set_timer钩子函数，它在每个测试用例的超时设置阶段被调用。 - 参数
item是一个pytest节点（node），表示当前的测试用例。 - 参数
settings是一个包含超时设置的对象。
- 使用
-
钩子函数文档字符串:
def pytest_timeout_set_timer(item, settings): """Called at timeout setup. 'item' is a pytest node to setup timeout for. Can be overridden by plugins for alternative timeout implementation strategies. """- 钩子函数的文档字符串说明了它在测试用例的超时设置阶段被调用。
item参数是用于设置超时的pytest节点。- 提到这个钩子可以被插件覆盖，以实现替代的超时实现策略。
-
pytest_timeout_cancel_timer 钩子:
@pytest.hookspec(firstresult=True) def pytest_timeout_cancel_timer(item):- 使用
@pytest.hookspec装饰器注册另一个钩子。 - 定义了
pytest_timeout_cancel_timer钩子函数，它在每个测试用例的超时取消阶段被调用。
- 使用
-
钩子函数文档字符串:
def pytest_timeout_cancel_timer(item): """Called at timeout teardown. 'item' is a pytest node which was used for timeout setup. Can be overridden by plugins for alternative timeout implementation strategies. """- 钩子函数的文档字符串说明了它在测试用例的超时取消阶段被调用。
item参数是用于取消超时的pytest节点，通常与pytest_timeout_set_timer中的节点相同。- 提到这个钩子也可以被插件覆盖，以实现替代的超时取消策略。
pytest_configure方法
@pytest.hookimpl
def pytest_configure(config):
"""Register the marker so it shows up in --markers output."""
config.addinivalue_line(
"markers",
"timeout(timeout, method=None, func_only=False, "
"disable_debugger_detection=False): Set a timeout, timeout "
"method and func_only evaluation on just one test item. The first "
"argument, *timeout*, is the timeout in seconds while the keyword, "
"*method*, takes the same values as the --timeout-method option. The "
"*func_only* keyword, when set to True, defers the timeout evaluation "
"to only the test function body, ignoring the time it takes when "
"evaluating any fixtures used in the test. The "
"*disable_debugger_detection* keyword, when set to True, disables "
"debugger detection, allowing breakpoint(), pdb.set_trace(), etc. "
"to be interrupted",
)
settings = get_env_settings(config)
config._env_timeout = settings.timeout
config._env_timeout_method = settings.method
config._env_timeout_func_only = settings.func_only
config._env_timeout_disable_debugger_detection = settings.disable_debugger_detection
timeout = config.getoption("session_timeout")
if timeout is None:
ini = config.getini("session_timeout")
if ini:
timeout = _validate_timeout(config.getini("session_timeout"), "config file")
if timeout is not None:
expire_time = time.time() + timeout
else:
expire_time = 0
timeout = 0
config.stash[SESSION_TIMEOUT_KEY] = timeout
config.stash[SESSION_EXPIRE_KEY] = expire_time
这段代码是一个 pytest 插件钩子函数 pytest_configure 的实现，它在 pytest 配置阶段被调用，用于注册和处理与超时相关的设置。
-
钩子装饰器:
@pytest.hookimpl def pytest_configure(config):使用
@pytest.hookimpl装饰器，将函数注册为pytest的钩子函数。 -
注册标记:
config.addinivalue_line( "markers", "timeout(timeout, method=None, func_only=False, " "disable_debugger_detection=False): Set a timeout, timeout " "method and func_only evaluation on just one test item. The first " "argument, *timeout*, is the timeout in seconds while the keyword, " "*method*, takes the same values as the --timeout-method option. The " "*func_only* keyword, when set to True, defers the timeout evaluation " "to only the test function body, ignoring the time it takes when " "evaluating any fixtures used in the test. The " "*disable_debugger_detection* keyword, when set to True, disables " "debugger detection, allowing breakpoint(), pdb.set_trace(), etc. " "to be interrupted", )使用
config.addinivalue_line方法向pytest的配置文件中添加一个标记timeout的说明。这个标记允许用户在测试项上设置超时时间、超时方法、函数体评估和调试器检测禁用选项。 -
获取环境设置:
settings = get_env_settings(config)调用
get_env_settings函数（未在代码中定义）获取环境设置，这些设置可能包括超时相关的配置。 -
设置环境超时变量:
config._env_timeout = settings.timeout config._env_timeout_method = settings.method config._env_timeout_func_only = settings.func_only config._env_timeout_disable_debugger_detection = settings.disable_debugger_detection将获取到的环境设置存储在
config对象的属性中，这些属性可能用于后续的超时逻辑。 -
处理会话超时:
timeout = config.getoption("session_timeout")获取命令行选项
--session-timeout的值。 -
验证和设置会话超时:
if timeout is None: ini = config.getini("session_timeout") if ini: timeout = _validate_timeout(config.getini("session_timeout"), "config file")如果命令行中没有设置会话超时，则检查配置文件中是否有设置，并使用
_validate_timeout函数（未在代码中定义）验证超时值。 -
计算过期时间:
if timeout is not None: expire_time = time.time() + timeout else: expire_time = 0 timeout = 0如果设置了超时时间，则计算出过期时间；否则，将过期时间和超时时间设置为 0。
-
存储超时信息:
config.stash[SESSION_TIMEOUT_KEY] = timeout config.stash[SESSION_EXPIRE_KEY] = expire_time将超时时间和过期时间存储在
config.stash字典中，以便在测试执行过程中使用。
pytest_runtest_protocol方法
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_protocol(item):
"""Hook in timeouts to the runtest protocol.
If the timeout is set on the entire test, including setup and
teardown, then this hook installs the timeout. Otherwise
pytest_runtest_call is used.
"""
hooks = item.config.pluginmanager.hook
settings = _get_item_settings(item)
is_timeout = settings.timeout is not None and settings.timeout > 0
if is_timeout and settings.func_only is False:
hooks.pytest_timeout_set_timer(item=item, settings=settings)
yield
if is_timeout and settings.func_only is False:
hooks.pytest_timeout_cancel_timer(item=item)
# check session timeout
expire_time = item.session.config.stash[SESSION_EXPIRE_KEY]
if expire_time and (expire_time < time.time()):
timeout = item.session.config.stash[SESSION_TIMEOUT_KEY]
item.session.shouldfail = f"session-timeout: {timeout} sec exceeded"
这段代码是 pytest 插件中用于处理测试超时的钩子函数 pytest_runtest_protocol 的实现，主要作用是在 pytest 的测试运行协议中集成超时逻辑，确保测试项的执行时间不会超过用户指定的超时时间，并且处理会话级别的超时。通过这种方式，可以避免测试因等待过长而挂起，提高测试的效率和响应性。
-
钩子装饰器:
@pytest.hookimpl(hookwrapper=True) def pytest_runtest_protocol(item):使用
@pytest.hookimpl装饰器，将函数注册为pytest的钩子函数，并使用hookwrapper=True参数来包装其他钩子的执行。 -
函数文档字符串:
def pytest_runtest_protocol(item): """Hook in timeouts to the runtest protocol. """文档字符串说明了这个钩子函数的作用是将超时逻辑集成到测试协议中。
-
获取插件钩子:
hooks = item.config.pluginmanager.hook从测试项的配置管理器中获取插件钩子，这些钩子可以被其他插件用来介入测试执行的不同阶段。
-
获取测试项设置:
settings = _get_item_settings(item)调用
_get_item_settings函数（未在代码中定义）获取当前测试项的设置，这些设置可能包括超时时间等。 -
检查是否设置了超时:
is_timeout = settings.timeout is not None and settings.timeout > 0判断是否设置了超时时间，并且超时时间大于 0。
-
设置超时计时器:
if is_timeout and settings.func_only is False: hooks.pytest_timeout_set_timer(item=item, settings=settings)如果设置了超时时间并且
func_only设置为False（表示超时包括整个测试过程，不仅仅是测试函数体），则调用pytest_timeout_set_timer钩子来设置超时计时器。 -
挂起执行:
yield使用
yield挂起当前钩子的执行，允许其他钩子在测试运行的不同阶段介入。 -
取消超时计时器:
if is_timeout and settings.func_only is False: hooks.pytest_timeout_cancel_timer(item=item)在测试项执行完毕后，如果设置了超时计时器，则调用
pytest_timeout_cancel_timer钩子来取消超时计时器。 -
检查会话超时:
expire_time = item.session.config.stash[SESSION_EXPIRE_KEY] if expire_time and (expire_time < time.time()): timeout = item.session.config.stash[SESSION_TIMEOUT_KEY] item.session.shouldfail = f"session-timeout: {timeout} sec exceeded"- 获取会话的过期时间，如果设置了过期时间并且当前时间已经超过了过期时间，则表示会话超时。
- 获取会话超时时间，如果存在会话超时，则设置
shouldfail属性，表示测试会话因超时而失败。
pytest_runtest_call方法
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_call(item):
"""Hook in timeouts to the test function call only.
If the timeout is set on only the test function this hook installs
the timeout, otherwise pytest_runtest_protocol is used.
"""
hooks = item.config.pluginmanager.hook
settings = _get_item_settings(item)
is_timeout = settings.timeout is not None and settings.timeout > 0
if is_timeout and settings.func_only is True:
hooks.pytest_timeout_set_timer(item=item, settings=settings)
yield
if is_timeout and settings.func_only is True:
hooks.pytest_timeout_cancel_timer(item=item)
这段代码是 pytest 插件中用于处理测试函数调用时的超时逻辑的钩子函数 pytest_runtest_call 的实现，主要作用是在 pytest 的测试运行协议中仅对测试函数调用阶段集成超时逻辑，确保测试函数的执行时间不会超过用户指定的超时时间。这对于控制测试执行时间特别有用，尤其是在测试函数可能执行长时间运行操作时。通过这种方式，可以避免测试因执行时间过长而挂起，提高测试的效率和响应性。
-
钩子装饰器:
@pytest.hookimpl(hookwrapper=True) def pytest_runtest_call(item):使用
@pytest.hookimpl装饰器，将函数注册为pytest的钩子函数，并使用hookwrapper=True参数来包装其他钩子的执行。 -
函数文档字符串:
def pytest_runtest_call(item): """Hook in timeouts to the test function call only. """文档字符串说明了这个钩子函数的作用是仅在测试函数调用时介入超时逻辑。
-
获取插件钩子:
hooks = item.config.pluginmanager.hook从测试项的配置管理器中获取插件钩子，这些钩子可以被其他插件用来介入测试执行的不同阶段。
-
获取测试项设置:
settings = _get_item_settings(item)调用
_get_item_settings函数（未在代码中定义）获取当前测试项的设置，这些设置可能包括超时时间等。 -
检查是否设置了超时:
is_timeout = settings.timeout is not None and settings.timeout > 0判断是否设置了超时时间，并且超时时间大于 0。
-
设置超时计时器:
if is_timeout and settings.func_only is True: hooks.pytest_timeout_set_timer(item=item, settings=settings)如果设置了超时时间并且
func_only设置为True（表示超时仅适用于测试函数体，不包括setup和teardown），则调用pytest_timeout_set_timer钩子来设置超时计时器。 -
挂起执行:
yield使用
yield挂起当前钩子的执行，允许其他钩子在测试运行的不同阶段介入。 -
取消超时计时器:
if is_timeout and settings.func_only is True: hooks.pytest_timeout_cancel_timer(item=item)在测试函数调用完毕后，如果设置了超时计时器，则调用
pytest_timeout_cancel_timer钩子来取消超时计时器。
pytest_report_header方法
@pytest.hookimpl(tryfirst=True)
def pytest_report_header(config):
"""Add timeout config to pytest header."""
timeout_header = []
if config._env_timeout:
timeout_header.append(
"timeout: %ss\ntimeout method: %s\ntimeout func_only: %s"
% (
config._env_timeout,
config._env_timeout_method,
config._env_timeout_func_only,
)
)
session_timeout = config.getoption("session_timeout")
if session_timeout:
timeout_header.append("session timeout: %ss" % session_timeout)
if timeout_header:
return timeout_header
这段代码是一个 pytest 插件钩子函数 pytest_report_header 的实现，它用于在 pytest 测试报告的头部添加关于超时配置的信息，主要作用是在 pytest 测试报告的头部显示超时相关的配置信息，包括环境级别的超时设置和会话超时设置。这为用户提供了清晰的测试配置概览，有助于理解测试的执行环境和限制。通过在报告头部显示这些信息，用户可以快速了解测试的超时策略。
-
钩子装饰器:
@pytest.hookimpl(tryfirst=True) def pytest_report_header(config):使用
@pytest.hookimpl装饰器，将函数注册为pytest的钩子函数，并使用tryfirst=True参数来确保这个钩子是第一个执行的。 -
函数文档字符串:
def pytest_report_header(config): """Add timeout config to pytest header."""文档字符串说明了这个钩子函数的作用是在
pytest报告的头部添加超时配置的信息。 -
初始化超时头部信息列表:
timeout_header = []创建一个空列表
timeout_header，用于存储超时配置的文本信息。 -
检查环境超时设置:
if config._env_timeout:检查是否存在环境级别的超时设置。
-
添加环境超时信息:
timeout_header.append( "timeout: %ss\ntimeout method: %s\ntimeout func_only: %s" % ( config._env_timeout, config._env_timeout_method, config._env_timeout_func_only, ) )如果存在环境级别的超时设置，将其格式化为字符串并添加到
timeout_header列表中。 -
检查会话超时设置:
session_timeout = config.getoption("session_timeout") if session_timeout:获取命令行选项
--session-timeout的值，并检查是否设置了会话超时。 -
添加会话超时信息:
timeout_header.append("session timeout: %ss" % session_timeout)如果设置了会话超时，将其格式化为字符串并添加到
timeout_header列表中。 -
返回超时头部信息:
if timeout_header: return timeout_header如果
timeout_header列表中有信息，则返回这些信息，它们将被添加到pytest报告的头部。
pytest_exception_interact方法
@pytest.hookimpl(tryfirst=True)
def pytest_exception_interact(node):
"""Stop the timeout when pytest enters pdb in post-mortem mode."""
hooks = node.config.pluginmanager.hook
hooks.pytest_timeout_cancel_timer(item=node)
这段代码是 pytest 插件中用于处理异常交互的钩子函数 pytest_exception_interact 的实现。
-
钩子装饰器:
@pytest.hookimpl(tryfirst=True) def pytest_exception_interact(node):使用
@pytest.hookimpl装饰器，将函数注册为pytest的钩子函数，并使用tryfirst=True参数来确保这个钩子是第一个执行的。 -
函数文档字符串:
def pytest_exception_interact(node): """Stop the timeout when pytest enters pdb in post-mortem mode."""文档字符串说明了这个钩子函数的作用是在
pytest进入pdb（Python Debugger）的事后模式时停止超时计时器。 -
获取插件钩子:
hooks = node.config.pluginmanager.hook从测试节点的配置管理器中获取插件钩子，这些钩子可以被其他插件用来介入测试执行的不同阶段。
-
取消超时计时器:
hooks.pytest_timeout_cancel_timer(item=node)调用
pytest_timeout_cancel_timer钩子来取消与测试节点node关联的超时计时器。
这个钩子函数的主要作用是在 pytest 测试执行过程中，如果用户在异常发生后进入 pdb 进行调试（即进入事后模式），则停止超时计时器。这可以防止在调试过程中测试因为超时而中断。
详细说明如下：
-
事后模式（post-mortem mode）：当测试发生错误或异常时，
pdb会启动，允许用户对错误现场进行调试。这通常在测试失败后发生，用户可以通过它来检查变量状态、调用堆栈等信息。 -
取消超时：在用户进入 pdb 调试时，测试的执行已经暂停，因此继续计时是没有意义的。通过取消超时，可以防止测试因为超时而提前终止，从而允许用户不受干扰地进行调试。
-
pytest_exception_interact钩子：这个钩子在pytest捕获到异常并且用户选择与异常交互时被调用。通过这个钩子，插件可以介入并执行一些操作，比如这里实现的取消超时计时器。
通过这种方式，pytest-timeout 插件提供了一种机制，确保在用户需要调试时，不会因为超时而干扰调试过程。这对于开发和调试测试用例非常有用。
pytest_enter_pdb方法
def pytest_enter_pdb():
"""Stop the timeouts when we entered pdb.
This stops timeouts from triggering when pytest's builting pdb
support notices we entered pdb.
"""
# Since pdb.set_trace happens outside of any pytest control, we don't have
# any pytest ``item`` here, so we cannot use timeout_teardown. Thus, we
# need another way to signify that the timeout should not be performed.
global SUPPRESS_TIMEOUT
SUPPRESS_TIMEOUT = True
这段代码的作用是在进入 Python Debugger (pdb) 时停止超时计时器。这通常用于 pytest 测试中，当用户在测试失败后进入 pdb 进行调试时，防止测试因为超时而中断。
-
函数文档字符串:
def pytest_enter_pdb(): """Stop the timeouts when we entered pdb. This stops timeouts from triggering when pytest's builting pdb support notices we entered pdb. """文档字符串说明了函数的目的：在进入
pdb时停止超时，防止在调试时因超时而中断。 -
停止超时的逻辑:
# Since pdb.set_trace happens outside of any pytest control, we don't have # any pytest ``item`` here, so we cannot use timeout_teardown. Thus, we # need another way to signify that the timeout should not be performed. global SUPPRESS_TIMEOUT SUPPRESS_TIMEOUT = True- 这段注释说明了
pdb.set_trace()是在pytest控制之外触发的，因此在这个上下文中没有pytest的item对象。 - 由于没有
item对象，不能使用timeout_teardown钩子来取消超时，因此需要另一种方式来指示不应该执行超时。 - 声明了一个全局变量
SUPPRESS_TIMEOUT并将其设置为True。这个全局变量作为一个标志，告诉超时机制在调试时不要触发。
- 这段注释说明了
详细说明：
-
pdb.set_trace()：这是一个
Python内置函数，用于启动pdb调试器。当在代码中调用此函数时，程序会进入调试模式，允许用户检查当前的执行状态和变量值。 -
全局变量
SUPPRESS_TIMEOUT：通过设置这个全局变量，插件或测试框架的其他部分可以检查是否进入了pdb调试模式，并据此决定是否应该取消超时计时器。 -
调试模式下的超时处理：在调试模式下，测试的执行已经暂停，继续计时是没有意义的。通过设置
SUPPRESS_TIMEOUT，可以防止在调试过程中因超时而中断测试。
is_debugging方法
def is_debugging(trace_func=None):
"""Detect if a debugging session is in progress.
This looks at both pytest's builtin pdb support as well as
externally installed debuggers using some heuristics.
This is done by checking if either of the following conditions is
true:
1. Examines the trace function to see if the module it originates
from is in KNOWN_DEBUGGING_MODULES.
2. Check is SUPPRESS_TIMEOUT is set to True.
:param trace_func: the current trace function, if not given will use
sys.gettrace(). Used to unit-test this function.
"""
global SUPPRESS_TIMEOUT, KNOWN_DEBUGGING_MODULES
if SUPPRESS_TIMEOUT:
return True
if trace_func is None:
trace_func = sys.gettrace()
trace_module = None
if trace_func:
trace_module = inspect.getmodule(trace_func) or inspect.getmodule(
trace_func.__class__
)
if trace_module:
parts = trace_module.__name__.split(".")
for name in KNOWN_DEBUGGING_MODULES:
if any(part.startswith(name) for part in parts):
return True
return False
SUPPRESS_TIMEOUT = False
这段代码提供了一个用于检测是否正在进行调试会话的函数 is_debugging，并且定义了一个全局变量 SUPPRESS_TIMEOUT。
-
函数定义:
def is_debugging(trace_func=None):定义了一个名为
is_debugging的函数，它接受一个可选参数trace_func，该参数用于指定当前的跟踪函数。 -
函数文档字符串:
def is_debugging(trace_func=None): """Detect if a debugging session is in progress. """文档字符串说明了这个函数的目的是检测是否正在进行调试会话。
-
全局变量声明:
global SUPPRESS_TIMEOUT, KNOWN_DEBUGGING_MODULES声明了两个全局变量
SUPPRESS_TIMEOUT和KNOWN_DEBUGGING_MODULES，这两个变量在函数中将被使用。 -
检测 SUPPRESS_TIMEOUT:
if SUPPRESS_TIMEOUT: return True如果
SUPPRESS_TIMEOUT被设置为True，则函数直接返回True，表示正在调试。 -
获取当前跟踪函数:
if trace_func is None: trace_func = sys.gettrace()如果
trace_func参数没有被提供，则使用sys.gettrace()获取当前的跟踪函数。 -
获取跟踪函数的模块:
trace_module = inspect.getmodule(trace_func) or inspect.getmodule( trace_func.__class__ )使用
inspect.getmodule获取跟踪函数的模块，如果跟踪函数本身没有关联的模块，则尝试获取其类关联的模块。 -
检测已知调试模块:
if trace_module: parts = trace_module.__name__.split(".") for name in KNOWN_DEBUGGING_MODULES: if any(part.startswith(name) for part in parts): return True如果跟踪函数的模块存在，则检查其名称是否与已知的调试模块名称有匹配。如果有匹配，则返回
True，表示正在调试。 -
返回结果:
return False如果没有检测到调试会话，则函数返回
False。 -
全局变量初始化:
SUPPRESS_TIMEOUT = False初始化全局变量
SUPPRESS_TIMEOUT为False，表示默认情况下不抑制超时。
详细说明：
-
KNOWN_DEBUGGING_MODULES：这是一个全局集合，包含了已知的调试器模块名称，如"pydevd"、"bdb"和"pydevd_frame_evaluator"。这些名称用于检测是否正在使用这些调试器。 -
SUPPRESS_TIMEOUT：这是一个全局变量，用于指示是否应该抑制超时。如果设置为True，则在调试时不会触发超时。 -
trace_func：这个参数允许用户传递一个自定义的跟踪函数，主要用于单元测试。
pytest_timeout_set_timer方法
@pytest.hookimpl(trylast=True)
def pytest_timeout_set_timer(item, settings):
"""Setup up a timeout trigger and handler."""
timeout_method = settings.method
if (
timeout_method == "signal"
and threading.current_thread() is not threading.main_thread()
):
timeout_method = "thread"
if timeout_method == "signal":
def handler(signum, frame):
__tracebackhide__ = True
timeout_sigalrm(item, settings)
def cancel():
signal.setitimer(signal.ITIMER_REAL, 0)
signal.signal(signal.SIGALRM, signal.SIG_DFL)
item.cancel_timeout = cancel
signal.signal(signal.SIGALRM, handler)
signal.setitimer(signal.ITIMER_REAL, settings.timeout)
elif timeout_method == "thread":
timer = threading.Timer(settings.timeout, timeout_timer, (item, settings))
timer.name = "%s %s" % (__name__, item.nodeid)
def cancel():
timer.cancel()
timer.join()
item.cancel_timeout = cancel
timer.start()
return True
这段代码是 pytest 插件中用于设置测试用例超时计时器的钩子函数 pytest_timeout_set_timer 的实现，主要作用是在 pytest 测试执行过程中，根据用户配置的超时方法，设置相应的超时计时器。当测试用例的执行时间超过指定的超时时间时，超时处理函数会被触发，从而中断测试用例的执行。这对于防止测试用例因执行时间过长而挂起非常有用。
-
钩子装饰器:
@pytest.hookimpl(trylast=True) def pytest_timeout_set_timer(item, settings):使用
@pytest.hookimpl装饰器，将函数注册为pytest的钩子函数，并使用trylast=True参数来确保这个钩子是最后一个执行的。 -
函数文档字符串:
def pytest_timeout_set_timer(item, settings): """Setup up a timeout trigger and handler."""文档字符串说明了这个钩子函数的作用是设置一个超时触发器和处理器。
-
获取超时方法:
timeout_method = settings.method从测试项的设置中获取超时方法。
-
检查是否在主线程:
if ( timeout_method == "signal" and threading.current_thread() is not threading.main_thread() ): timeout_method = "thread"如果超时方法设置为
"signal"且当前线程不是主线程，则将超时方法改为"thread"。 -
处理信号超时:
if timeout_method == "signal":如果使用信号超时方法，则执行以下操作：
- 定义一个
handler函数来处理SIGALRM信号。 - 定义一个
cancel函数来取消信号超时。 - 设置
item.cancel_timeout为cancel函数，以便在需要时可以取消超时。 - 使用
signal.signal设置SIGALRM信号的处理函数。 - 使用
signal.setitimer设置实时定时器。
- 定义一个
-
处理线程超时:
elif timeout_method == "thread":如果使用线程超时方法，则执行以下操作：
- 创建一个
threading.Timer对象来触发超时。 - 定义一个
cancel函数来取消线程超时。 - 设置
item.cancel_timeout为cancel函数，以便在需要时可以取消超时。 - 启动定时器。
- 创建一个
-
返回结果:
return True函数返回
True，表示钩子逻辑正常结束。
详细说明：
-
信号超时 (
signal)：使用signal模块设置SIGALRM信号，当超时发生时，信号处理函数会被调用，触发超时处理。 -
线程超时 (
thread)：使用threading.Timer创建一个定时器线程，当超时发生时，定时器会触发超时处理。 -
取消超时：通过
item.cancel_timeout方法，可以在测试用例执行过程中取消超时，防止不必要的超时中断。 -
settings.timeout：超时时间，用于设置信号超时或线程超时。
pytest_timeout_cancel_timer方法
@pytest.hookimpl(trylast=True)
def pytest_timeout_cancel_timer(item):
"""Cancel the timeout trigger if it was set."""
# When skipping is raised from a pytest_runtest_setup function
# (as is the case when using the pytest.mark.skipif marker) we
# may be called without our setup counterpart having been
# called.
cancel = getattr(item, "cancel_timeout", None)
if cancel:
cancel()
return True
这段代码是 pytest 插件中用于取消测试用例超时计时器的钩子函数 pytest_timeout_cancel_timer 的实现，通过这个钩子函数，插件可以确保在不需要超时限制的情况下，及时取消超时计时器，避免对测试执行产生干扰。这对于提高测试的灵活性和可靠性非常重要。
-
钩子装饰器:
@pytest.hookimpl(trylast=True) def pytest_timeout_cancel_timer(item):使用
@pytest.hookimpl装饰器，将函数注册为pytest的钩子函数，并使用trylast=True参数来确保这个钩子是最后一个执行的。 -
函数文档字符串:
def pytest_timeout_cancel_timer(item): """Cancel the timeout trigger if it was set."""文档字符串说明了这个钩子函数的作用是取消之前设置的超时触发器（如果已经设置）。
-
处理跳过测试的情况:
cancel = getattr(item, "cancel_timeout", None)使用
getattr从测试项item中获取cancel_timeout方法。这个方法应该在设置超时时被赋值。如果没有找到，则默认为None。 -
取消超时:
if cancel: cancel()如果存在
cancel_timeout方法，则调用它来取消超时。这通常发生在测试用例被跳过或在测试用例的 setup 阶段发生错误时。 -
返回结果:
return True函数返回
True，表示钩子逻辑正常结束。
详细说明：
-
超时取消：这个钩子函数的主要目的是取消测试用例的超时计时器。这在测试用例被跳过或提前终止时非常有用，可以防止不必要的超时中断。
-
item.cancel_timeout：这是一个在设置超时时赋值的方法，用于取消超时。它在pytest_timeout_set_timer钩子函数中被设置。 -
跳过测试：当使用
pytest的skipif标记或其他原因导致测试用例被跳过时，pytest_runtest_setup钩子可能不会被调用，因此cancel_timeout方法可能不会被设置。这种情况下，getattr会返回None，取消操作将不会执行。
get_env_settings方法
def get_env_settings(config):
"""Return the configured timeout settings.
This looks up the settings in the environment and config file.
"""
timeout = config.getvalue("timeout")
if timeout is None:
timeout = _validate_timeout(
os.environ.get("PYTEST_TIMEOUT"), "PYTEST_TIMEOUT environment variable"
)
if timeout is None:
ini = config.getini("timeout")
if ini:
timeout = _validate_timeout(ini, "config file")
method = config.getvalue("timeout_method")
if method is None:
ini = config.getini("timeout_method")
if ini:
method = _validate_method(ini, "config file")
if method is None:
method = DEFAULT_METHOD
func_only = config.getini("timeout_func_only")
disable_debugger_detection = config.getvalue("timeout_disable_debugger_detection")
if disable_debugger_detection is None:
ini = config.getini("timeout_disable_debugger_detection")
if ini:
disable_debugger_detection = _validate_disable_debugger_detection(
ini, "config file"
)
return Settings(timeout, method, func_only, disable_debugger_detection)
这段代码定义了一个名为 get_env_settings 的函数，它用于从环境变量、配置文件和 pytest 配置中获取超时设置。这个函数的主要作用是统一获取和验证超时相关的设置，确保它们在测试执行中正确应用。通过从不同的来源（pytest 配置、环境变量、配置文件）获取设置，并提供默认值，它提供了灵活且健壮的配置管理。
-
函数定义:
def get_env_settings(config):
定义了一个函数 get_env_settings，它接收一个参数 config，这是 pytest 的配置对象。
-
函数文档字符串:
def get_env_settings(config): """Return the configured timeout settings. This looks up the settings in the environment and config file. """
文档字符串说明了这个函数的目的是返回配置的超时设置，它会在环境变量和配置文件中查找这些设置。
-
获取超时时间:
timeout = config.getvalue("timeout")
从 pytest 配置中获取 timeout 设置。
-
从环境变量获取超时时间:
if timeout is None: timeout = _validate_timeout( os.environ.get("PYTEST_TIMEOUT"), "PYTEST_TIMEOUT environment variable" )
如果 pytest 配置中没有设置超时时间，则尝试从环境变量 PYTEST_TIMEOUT 中获取，并使用 _validate_timeout 函数验证其有效性。
-
从配置文件获取超时时间:
if timeout is None: ini = config.getini("timeout") if ini: timeout = _validate_timeout(ini, "config file")
如果环境变量中也没有设置超时时间，则尝试从 pytest 配置文件中获取，并使用 _validate_timeout 函数验证其有效性。
-
获取超时方法:
method = config.getvalue("timeout_method")
从 pytest 配置中获取 timeout_method 设置。
-
从配置文件获取超时方法:
if method is None: ini = config.getini("timeout_method") if ini: method = _validate_method(ini, "config file")
如果 pytest 配置中没有设置超时方法，则尝试从配置文件中获取，并使用 _validate_method 函数验证其有效性。
-
设置默认超时方法:
if method is None: method = DEFAULT_METHOD
如果配置文件中也没有设置超时方法，则使用默认值 DEFAULT_METHOD。
-
获取函数体超时评估:
func_only = config.getini("timeout_func_only")
从 pytest 配置文件中获取 timeout_func_only 设置，这决定了是否仅在测试函数体中计算超时时间。
-
获取调试器检测禁用设置:
disable_debugger_detection = config.getvalue("timeout_disable_debugger_detection")
从 pytest 配置中获取 timeout_disable_debugger_detection 设置。
-
从配置文件获取调试器检测禁用设置:
if disable_debugger_detection is None: ini = config.getini("timeout_disable_debugger_detection") if ini: disable_debugger_detection = _validate_disable_debugger_detection( ini, "config file" )
如果 pytest 配置中没有设置调试器检测禁用，则尝试从配置文件中获取，并使用 _validate_disable_debugger_detection 函数验证其有效性。
-
返回设置:
return Settings(timeout, method, func_only, disable_debugger_detection)
返回一个 Settings 对象，包含超时时间、超时方法、函数体超时评估和调试器检测禁用设置。
详细说明：
-
_validate_timeout函数：用于验证超时时间的有效性，确保它是合理的数值。 -
_validate_method函数：用于验证超时方法的有效性，确保它是一个允许的值（如"signal"或"thread"）。 -
_validate_disable_debugger_detection函数：用于验证调试器检测禁用设置的有效性，确保它是布尔值。 -
Settings对象：是一个命名元组，用于存储和传递超时相关的设置。
_get_item_settings方法
def _get_item_settings(item, marker=None):
"""Return (timeout, method) for an item."""
timeout = method = func_only = disable_debugger_detection = None
if not marker:
marker = item.get_closest_marker("timeout")
if marker is not None:
settings = _parse_marker(item.get_closest_marker(name="timeout"))
timeout = _validate_timeout(settings.timeout, "marker")
method = _validate_method(settings.method, "marker")
func_only = _validate_func_only(settings.func_only, "marker")
disable_debugger_detection = _validate_disable_debugger_detection(
settings.disable_debugger_detection, "marker"
)
if timeout is None:
timeout = item.config._env_timeout
if method is None:
method = item.config._env_timeout_method
if func_only is None:
func_only = item.config._env_timeout_func_only
if disable_debugger_detection is None:
disable_debugger_detection = item.config._env_timeout_disable_debugger_detection
return Settings(timeout, method, func_only, disable_debugger_detection)
这个函数的主要作用是统一获取和验证超时相关的设置，确保它们在测试执行中正确应用。通过从不同的来源（测试项的标记、配置）获取设置，并提供默认值，它提供了灵活且健壮的配置管理。
-
函数定义:
def _get_item_settings(item, marker=None):
定义了一个函数 _get_item_settings，接收两个参数：item 是 pytest 的测试项对象，marker 是一个可选参数，用于指定特定的标记。
-
函数文档字符串:
def _get_item_settings(item, marker=None): """Return (timeout, method) for an item."""
文档字符串说明了这个函数的目的是返回测试项的超时和方法设置。
-
初始化设置变量:
timeout = method = func_only = disable_debugger_detection = None
初始化四个变量，分别用于存储超时时间、超时方法、函数体超时评估和调试器检测禁用设置。
-
获取标记:
if not marker: marker = item.get_closest_marker("timeout")
如果 marker 参数未提供，则尝试获取测试项上最近的 "timeout" 标记。
-
解析标记设置:
if marker is not None: settings = _parse_marker(item.get_closest_marker(name="timeout")) timeout = _validate_timeout(settings.timeout, "marker") method = _validate_method(settings.method, "marker") func_only = _validate_func_only(settings.func_only, "marker") disable_debugger_detection = _validate_disable_debugger_detection( settings.disable_debugger_detection, "marker" )
如果存在 "timeout" 标记，则解析该标记，并验证超时时间、方法、函数体超时评估和调试器检测禁用设置。
-
从配置获取超时时间:
if timeout is None: timeout = item.config._env_timeout
如果未从标记中获取到超时时间，则从测试项的配置中获取。
-
从配置获取超时方法:
if method is None: method = item.config._env_timeout_method
如果未从标记中获取到超时方法，则从测试项的配置中获取。
-
从配置获取函数体超时评估:
if func_only is None: func_only = item.config._env_timeout_func_only
如果未从标记中获取到函数体超时评估设置，则从测试项的配置中获取。
-
从配置获取调试器检测禁用设置:
if disable_debugger_detection is None: disable_debugger_detection = item.config._env_timeout_disable_debugger_detection
如果未从标记中获取到调试器检测禁用设置，则从测试项的配置中获取。
-
返回设置:
return Settings(timeout, method, func_only, disable_debugger_detection)
返回一个 Settings 对象，包含超时时间、超时方法、函数体超时评估和调试器检测禁用设置。
详细说明：
-
_parse_marker函数：用于解析"timeout"标记的内容，提取超时相关的设置。 -
_validate_timeout函数：验证超时时间的有效性，确保它是合理的数值。 -
_validate_method函数：验证超时方法的有效性，确保它是一个允许的值（如"signal"或"thread"）。 -
_validate_func_only函数：验证函数体超时评估设置的有效性，确保它是布尔值。 -
_validate_disable_debugger_detection函数：验证调试器检测禁用设置的有效性，确保它是布尔值。 -
Settings对象：是一个命名元组，用于存储和传递超时相关的设置。
_parse_marker方法
def _parse_marker(marker):
"""Return (timeout, method) tuple from marker.
Either could be None. The values are not interpreted, so
could still be bogus and even the wrong type.
"""
if not marker.args and not marker.kwargs:
raise TypeError("Timeout marker must have at least one argument")
timeout = method = func_only = NOTSET = object()
for kw, val in marker.kwargs.items():
if kw == "timeout":
timeout = val
elif kw == "method":
method = val
elif kw == "func_only":
func_only = val
else:
raise TypeError("Invalid keyword argument for timeout marker: %s" % kw)
if len(marker.args) >= 1 and timeout is not NOTSET:
raise TypeError("Multiple values for timeout argument of timeout marker")
elif len(marker.args) >= 1:
timeout = marker.args[0]
if len(marker.args) >= 2 and method is not NOTSET:
raise TypeError("Multiple values for method argument of timeout marker")
elif len(marker.args) >= 2:
method = marker.args[1]
if len(marker.args) > 2:
raise TypeError("Too many arguments for timeout marker")
if timeout is NOTSET:
timeout = None
if method is NOTSET:
method = None
if func_only is NOTSET:
func_only = None
return Settings(timeout, method, func_only, None)
这个函数的主要作用是将 pytest 标记中的超时设置解析出来，并提供一种一致的方式来获取这些设置。这对于实现超时功能的自定义和验证非常有用。
-
函数定义:
def _parse_marker(marker):
定义了一个函数 _parse_marker，它接收一个参数 marker，这是 pytest 的标记对象。
-
函数文档字符串:
def _parse_marker(marker): """Return (timeout, method) tuple from marker. Either could be None. The values are not interpreted, so could still be bogus and even the wrong type. """
文档字符串说明了这个函数的目的是从一个标记中返回一个包含超时时间和方法的元组。同时指出这些值尚未被解释，因此可能是错误的或甚至类型不正确。
-
检查标记参数:
if not marker.args and not marker.kwargs: raise TypeError("Timeout marker must have at least one argument")
如果标记没有位置参数 (args) 也没有关键字参数 (kwargs)，则抛出 TypeError。
-
初始化变量:
timeout = method = func_only = NOTSET = object()
初始化 timeout、method 和 func_only 变量，并设置为 NOTSET。NOTSET 是一个对象，用于标识这些值尚未被设置。
-
解析关键字参数:
for kw, val in marker.kwargs.items(): if kw == "timeout": timeout = val elif kw == "method": method = val elif kw == "func_only": func_only = val else: raise TypeError("Invalid keyword argument for timeout marker: %s" % kw)
遍历标记的关键字参数，并根据参数名称设置相应的变量。如果遇到无效的关键字参数，则抛出 TypeError。
-
解析位置参数:
if len(marker.args) >= 1 and timeout is not NOTSET: raise TypeError("Multiple values for timeout argument of timeout marker") elif len(marker.args) >= 1: timeout = marker.args[0]
如果标记有位置参数，并且 timeout 已经被设置，则抛出 TypeError。否则，将第一个位置参数赋值给 timeout。
-
解析方法参数:
if len(marker.args) >= 2 and method is not NOTSET: raise TypeError("Multiple values for method argument of timeout marker") elif len(marker.args) >= 2: method = marker.args[1]
如果标记有第二个位置参数，并且 method 已经被设置，则抛出 TypeError。否则，将第二个位置参数赋值给 method。
-
检查参数数量:
if len(marker.args) > 2: raise TypeError("Too many arguments for timeout marker")
如果标记有超过两个位置参数，则抛出 TypeError。
-
处理未设置的值:
if timeout is NOTSET: timeout = None if method is NOTSET: method = None if func_only is NOTSET: func_only = None
如果 timeout、method 或 func_only 仍然为 NOTSET，则将它们设置为 None。
-
返回设置:
return Settings(timeout, method, func_only, None)
返回一个 Settings 对象，包含超时时间、超时方法、函数体超时评估和调试器检测禁用设置。
详细说明：
-
参数解析：函数首先检查标记是否有参数，然后分别解析位置参数和关键字参数。
-
错误处理：如果标记的参数数量不正确或参数名称无效，函数会抛出
TypeError。 -
默认值处理：如果参数未在标记中设置，则在函数结束时将它们设置为
None。 -
Settings对象：返回的Settings对象包含了所有解析出的设置，方便后续使用。
几个私有validate方法
def _validate_timeout(timeout, where):
if timeout is None:
return None
try:
return float(timeout)
except ValueError:
raise ValueError("Invalid timeout %s from %s" % (timeout, where))
def _validate_method(method, where):
if method is None:
return None
if method not in ["signal", "thread"]:
raise ValueError("Invalid method %s from %s" % (method, where))
return method
def _validate_func_only(func_only, where):
if func_only is None:
return None
if not isinstance(func_only, bool):
raise ValueError("Invalid func_only value %s from %s" % (func_only, where))
return func_only
def _validate_disable_debugger_detection(disable_debugger_detection, where):
if disable_debugger_detection is None:
return None
if not isinstance(disable_debugger_detection, bool):
raise ValueError(
"Invalid disable_debugger_detection value %s from %s"
% (disable_debugger_detection, where)
)
return disable_debugger_detection
这段代码包含了一系列用于验证超时设置的函数，它们确保从 pytest 标记、环境变量或配置文件中获取的值是有效的。
-
_validate_timeout函数:def _validate_timeout(timeout, where):
用于验证超时时间设置：
-
timeout: 超时时间值。 -
where: 值来源的描述，例如 “marker”、“environment variable” 或 “config file”。
-
处理
None值:if timeout is None: return None
如果超时时间设置为 None，则直接返回 None。
-
转换为浮点数:
try: return float(timeout) except ValueError: raise ValueError("Invalid timeout %s from %s" % (timeout, where))
尝试将超时时间转换为浮点数。如果转换失败（例如，值不是数字），则抛出 ValueError。
-
_validate_method函数:def _validate_method(method, where):
用于验证超时方法设置：
-
method: 超时方法值。 -
where: 值来源的描述。
-
处理
None值:if method is None: return None
如果超时方法设置为 None，则直接返回 None。
-
检查有效性:
if method not in ["signal", "thread"]: raise ValueError("Invalid method %s from %s" % (method, where))
检查超时方法是否为 “signal” 或 “thread”。如果不是，则抛出 ValueError。
-
_validate_func_only函数:def _validate_func_only(func_only, where):
用于验证函数体超时评估设置。
-
func_only: 函数体超时评估值。 -
where: 值来源的描述。
-
处理
None值:if func_only is None: return None
如果函数体超时评估设置为 None，则直接返回 None。
-
检查布尔值:
if not isinstance(func_only, bool): raise ValueError("Invalid func_only value %s from %s" % (func_only, where))
检查函数体超时评估值是否为布尔类型。如果不是，则抛出 ValueError。
-
_validate_disable_debugger_detection函数:def _validate_disable_debugger_detection(disable_debugger_detection, where):
用于验证调试器检测禁用设置：
- disable_debugger_detection: 调试器检测禁用值。
- where: 值来源的描述。
-
处理
None值:if disable_debugger_detection is None: return None
如果调试器检测禁用设置为 None，则直接返回 None。
-
检查布尔值:
if not isinstance(disable_debugger_detection, bool): raise ValueError( "Invalid disable_debugger_detection value %s from %s" % (disable_debugger_detection, where) )
检查调试器检测禁用值是否为布尔类型。如果不是，则抛出 ValueError。
详细说明：
-
这些函数通过检查和转换确保超时相关的设置是有效的，并且符合预期的类型和值。
-
它们在获取设置时提供了一种一致的验证机制，有助于避免因配置错误而导致的测试执行问题。
-
通过在抛出异常时包含值来源的描述，可以更容易地定位和解决问题。
timeout_sigalrm方法
def timeout_sigalrm(item, settings):
"""Dump stack of threads and raise an exception.
This will output the stacks of any threads other then the
current to stderr and then raise an AssertionError, thus
terminating the test.
"""
if not settings.disable_debugger_detection and is_debugging():
return
__tracebackhide__ = True
nthreads = len(threading.enumerate())
terminal = item.config.get_terminal_writer()
if nthreads > 1:
terminal.sep("+", title="Timeout")
dump_stacks(terminal)
if nthreads > 1:
terminal.sep("+", title="Timeout")
pytest.fail("Timeout >%ss" % settings.timeout)
这个函数的主要作用是在测试用例超时时提供详细的调试信息，并明确地终止测试。这对于识别和处理测试执行过程中的长时间运行或死锁非常有用。
-
函数定义:
def timeout_sigalrm(item, settings):
定义了一个函数 timeout_sigalrm，接收两个参数：
-
item：pytest的测试项对象。 -
settings：包含超时设置的Settings对象。
-
函数文档字符串:
def timeout_sigalrm(item, settings): """Dump stack of threads and raise an exception. This will output the stacks of any threads other then the current to stderr and then raise an AssertionError, thus terminating the test. """
文档字符串说明了这个函数的作用是在超时发生时转储线程栈，并引发异常以终止测试。
-
调试器检测:
if not settings.disable_debugger_detection and is_debugging(): return
如果 settings.disable_debugger_detection 为 False 并且检测到正在调试（is_debugging() 返回 True），则函数返回，不执行超时处理。
-
设置 traceback 隐藏:
__tracebackhide__ = True
设置一个特殊的变量 __tracebackhide__ 为 True，这告诉 pytest 在显示 traceback 时隐藏这部分的堆栈跟踪。
-
获取线程数:
nthreads = len(threading.enumerate())
使用 threading.enumerate() 获取当前线程的列表，并计算线程的数量。
-
获取终端写入器:
terminal = item.config.get_terminal_writer()
从测试项的配置中获取终端写入器，用于向终端输出信息。
-
输出超时信息:
if nthreads > 1: terminal.sep("+", title="Timeout")
如果有多个线程，则在终端输出一个分隔线，并标注标题为 “Timeout”。
-
转储线程栈:
dump_stacks(terminal)
调用 dump_stacks 函数（未在代码中定义），将所有线程的栈信息输出到终端。
-
再次输出超时信息:
if nthreads > 1: terminal.sep("+", title="Timeout")
如果有多个线程，则再次在终端输出一个分隔线，并标注标题为 “Timeout”。
-
引发超时失败:
pytest.fail("Timeout >%ss" % settings.timeout)
使用 pytest 的 fail 函数引发一个失败，消息中包含超时时间。
详细说明：
-
线程栈转储：当测试用例执行超过指定的超时时间时，这个函数会被调用，输出所有线程的栈信息，有助于调试和诊断问题。
-
调试器检测：如果设置了不检测调试器，或者当前不在调试模式，则会跳过超时处理。
-
超时失败：最终，函数会引发一个
AssertionError，导致测试失败，并显示超时信息。
timeout_timer方法
def timeout_timer(item, settings):
"""Dump stack of threads and call os._exit().
This disables the capturemanager and dumps stdout and stderr.
Then the stacks are dumped and os._exit(1) is called.
"""
if not settings.disable_debugger_detection and is_debugging():
return
terminal = item.config.get_terminal_writer()
try:
capman = item.config.pluginmanager.getplugin("capturemanager")
if capman:
capman.suspend_global_capture(item)
stdout, stderr = capman.read_global_capture()
else:
stdout, stderr = None, None
terminal.sep("+", title="Timeout")
caplog = item.config.pluginmanager.getplugin("_capturelog")
if caplog and hasattr(item, "capturelog_handler"):
log = item.capturelog_handler.stream.getvalue()
if log:
terminal.sep("~", title="Captured log")
terminal.write(log)
if stdout:
terminal.sep("~", title="Captured stdout")
terminal.write(stdout)
if stderr:
terminal.sep("~", title="Captured stderr")
terminal.write(stderr)
dump_stacks(terminal)
terminal.sep("+", title="Timeout")
except Exception:
traceback.print_exc()
finally:
terminal.flush()
sys.stdout.flush()
sys.stderr.flush()
os._exit(1)
这个函数的主要作用是在测试用例超时时提供详细的调试信息（特别是当使用线程超时方法时），并强制终止测试。这对于识别和处理测试执行过程中的长时间运行或死锁非常有用。
-
函数定义:
def timeout_timer(item, settings):
定义了一个函数 timeout_timer，接收两个参数：
-
item：pytest的测试项对象。 -
settings：包含超时设置的Settings对象。
-
函数文档字符串:
def timeout_timer(item, settings): """Dump stack of threads and call os._exit(). This disables the capturemanager and dumps stdout and stderr. Then the stacks are dumped and os._exit(1) is called. """
文档字符串说明了这个函数的作用是在超时发生时转储线程栈，输出捕获的标准输出和标准错误，然后调用 os._exit(1) 终止进程。
-
调试器检测:
if not settings.disable_debugger_detection and is_debugging(): return
如果 settings.disable_debugger_detection 为 False 并且检测到正在调试（is_debugging() 返回 True），则函数返回，不执行超时处理。
-
获取终端写入器:
terminal = item.config.get_terminal_writer()
从测试项的配置中获取终端写入器，用于向终端输出信息。
-
获取捕获管理器:
capman = item.config.pluginmanager.getplugin("capturemanager")
尝试获取捕获管理器插件，该插件负责捕获测试中的输出。
-
暂停全局捕获:
if capman: capman.suspend_global_capture(item) stdout, stderr = capman.read_global_capture() else: stdout, stderr = None, None
如果捕获管理器存在，暂停全局捕获并读取捕获到的标准输出和标准错误。
-
输出超时信息:
terminal.sep("+", title="Timeout")
在终端输出一个分隔线，并标注标题为 “Timeout”。
-
获取日志捕获插件:
caplog = item.config.pluginmanager.getplugin("_capturelog")
尝试获取日志捕获插件。
-
输出捕获的日志:
if caplog and hasattr(item, "capturelog_handler"): log = item.capturelog_handler.stream.getvalue() if log: terminal.sep("~", title="Captured log") terminal.write(log)
如果日志捕获插件存在，输出捕获的日志。
-
输出捕获的标准输出:
if stdout: terminal.sep("~", title="Captured stdout") terminal.write(stdout)
如果存在捕获的标准输出，输出它。
-
输出捕获的标准错误:
if stderr: terminal.sep("~", title="Captured stderr") terminal.write(stderr)
如果存在捕获的标准错误，输出它。
-
转储线程栈:
dump_stacks(terminal)
调用 dump_stacks 函数（未在代码中定义），将所有线程的栈信息输出到终端。
-
再次输出超时信息:
terminal.sep("+", title="Timeout")
再次在终端输出一个分隔线，并标注标题为 “Timeout”。
-
异常处理:
except Exception: traceback.print_exc()
如果在处理过程中发生异常，输出异常的 traceback。
-
清理和退出:
finally: terminal.flush() sys.stdout.flush() sys.stderr.flush() os._exit(1)在
finally块中，确保终端、标准输出和标准错误的缓冲区被刷新，然后调用os._exit(1)立即终止进程。
详细说明：
-
捕获管理器：函数尝试获取捕获管理器插件，并在超时发生时暂停全局捕获，以便输出捕获的内容。
-
日志捕获：函数还尝试获取日志捕获插件，并输出捕获的日志。
-
线程栈转储：函数输出所有线程的栈信息，有助于调试和诊断问题。
-
立即退出：最终，函数通过
os._exit(1)立即终止进程，确保测试因超时而失败。
dump_stacks方法
def dump_stacks(terminal):
"""Dump the stacks of all threads except the current thread."""
current_ident = threading.current_thread().ident
for thread_ident, frame in sys._current_frames().items():
if thread_ident == current_ident:
continue
for t in threading.enumerate():
if t.ident == thread_ident:
thread_name = t.name
break
else:
thread_name = "<unknown>"
terminal.sep("~", title="Stack of %s (%s)" % (thread_name, thread_ident))
terminal.write("".join(traceback.format_stack(frame)))
这个函数用于转储除当前线程之外的所有线程的栈信息，通常在测试超时或其他需要调试多线程行为的情况下使用，提供了一种快速查看所有线程当前状态的方法。这对于识别死锁、长时间运行的任务或其他并发问题非常有用。
-
函数定义:
def dump_stacks(terminal):
定义了一个函数 dump_stacks，接收一个参数 terminal，这是用于输出信息的终端写入器。
-
函数文档字符串:
def dump_stacks(terminal): """Dump the stacks of all threads except the current thread."""
文档字符串说明了这个函数的作用是转储所有其他线程的栈信息，不包括当前线程。
-
获取当前线程标识:
current_ident = threading.current_thread().ident
使用 threading.current_thread().ident 获取当前线程的唯一标识符。
-
遍历所有线程的栈:
for thread_ident, frame in sys._current_frames().items():
使用 sys._current_frames() 获取一个字典，其中包含所有线程的标识符和它们的当前栈帧。
-
跳过当前线程:
if thread_ident == current_ident: continue
如果线程标识符与当前线程标识符相同，则跳过当前线程。
-
获取线程名称:
for t in threading.enumerate(): if t.ident == thread_ident: thread_name = t.name break else: thread_name = "<unknown>"
遍历所有线程，找到与当前标识符匹配的线程，并获取其名称。如果找不到，则将名称设置为 <unknown>。
-
输出线程栈信息:
terminal.sep("~", title="Stack of %s (%s)" % (thread_name, thread_ident)) terminal.write("".join(traceback.format_stack(frame)))
使用终端写入器输出一个分隔线，并标注标题为线程名称和标识符。然后输出该线程的栈信息，使用 traceback.format_stack(frame) 格式化栈帧。
详细说明：
-
线程栈转储：函数输出所有其他线程的栈信息，有助于调试和诊断问题，特别是在测试超时或其他长时间运行的问题时。
-
跳过当前线程：函数故意跳过当前线程的栈信息，因为当前线程可能不包含相关的上下文信息。
-
线程名称获取：尝试获取线程的名称以提供更友好的输出。如果无法获取名称，则使用
<unknown>作为占位符。 -
输出格式：使用终端写入器输出格式化的栈信息，使其易于阅读和理解。
结语
pytest-timeout 是一个 pytest 插件，它为测试提供了超时功能，以下是其关键特性的总结：
-
自动超时：允许你为测试用例设置一个最大执行时间限制，超过这个时间限制的测试将被自动终止。
-
灵活的超时机制：支持两种超时机制——使用信号 (
signal) 或线程 (thread) 来实现超时。 -
命令行和配置文件选项：可以通过命令行参数或
pytest配置文件来设置超时时间、方法和其他相关选项。 -
调试支持：提供了选项来禁用调试器检测，允许在调试模式下运行测试而不触发超时。
-
会话超时：除了单个测试用例的超时，还支持设置整个测试会话的超时时间。
-
易于集成：与
pytest紧密结合，易于安装和使用，为现有的测试流程添加了超时保护。 -
报告和日志：在测试报告中提供超时信息，并在必要时输出所有线程的栈跟踪信息，有助于调试。
总的来说，pytest-timeout 插件通过为测试用例添加超时限制，帮助确保测试不会因为意外情况而无限期地挂起，提高了测试的可靠性和效率。
