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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
|
#!/usr/bin/env python3
# --------------------( LICENSE )--------------------
# Copyright (c) 2014-2025 Beartype authors.
# See "LICENSE" for further details.
'''
**Root test configuration** (i.e., early-time configuration guaranteed to be
run by :mod:`pytest` *before* passed command-line arguments are parsed) for
this test suite.
Caveats
----------
For safety, this configuration should contain *only* early-time hooks
absolutely required by :mod:`pytest` design to be defined in this
configuration. Hooks for which this is the case (e.g.,
:func:`pytest_addoption`) are explicitly annotated as such in official
:mod:`pytest` documentation with a note resembling:
Note
This function should be implemented only in plugins or ``conftest.py``
files situated at the tests root directory due to how pytest discovers
plugins during startup.
This file is the aforementioned ``conftest.py`` file "...situated at the tests
root directory."
'''
# ....................{ IMPORTS }....................
from collections.abc import Callable
from pytest import Function
from typing import Optional
# ....................{ HOOKS ~ ini }....................
def pytest_configure(config) -> None:
'''
Hook programmatically configuring the top-level ``"pytest.ini"`` file.
'''
# Programmatically add our custom "run_in_subprocess" mark, enabling tests
# to notify the pytest_pyfunc_call() hook that they require isolation to a
# Python subprocess of the current Python process.
config.addinivalue_line(
'markers',
f'{_MARK_NAME_SUBPROCESS}: mark test to run in an isolated subprocess',
)
# ....................{ HOOKS ~ test : run }....................
def pytest_pyfunc_call(pyfuncitem: Function) -> Optional[bool]:
'''
Hook intercepting the call to run the passed :mod:`pytest` test function.
Specifically, this test:
* If this test has been decorated by our custom
``@pytest.mark.run_in_subprocess`` marker, runs this test in a Python
subprocess of the current Python process isolated to this test.
* Else, runs this test in the current Python process by deferring to the
standard :mod:`pytest` logic for running this test.
Parameters
----------
pyfuncitem: Function
:mod:`pytest`-specific object encapsulating the current test function
being run.
Returns
-------
Optional[bool]
Either:
* If this hook ran this test, :data:`True`.
* If this hook did *not* run this test, :data:`None`.
See Also
--------
https://github.com/ansible/pytest-mp/issues/15#issuecomment-1342682418
GitHub comment by @pelson (Phil Elson) strongly inspiring this hook.
'''
# If this test has been decorated by our custom
# @pytest.mark.run_in_subprocess marker...
if _MARK_NAME_SUBPROCESS in pyfuncitem.keywords:
# Defer hook-specific imports.
from functools import partial
from multiprocessing import Process
from pytest import fail
# Test function to be run.
test_func = pyfuncitem.function
# Partial object encapsulating the _run_test_in_subprocess() helper
# bound to the current "pyfuncitem" object encapsulating this test.
#
# Note that this logic is robust under POSIX-compliant platforms (e.g.,
# Linux, macOS) but *EXTREMELY* fragile under Windows, where the
# "multiprocessing.Process" class sadly leverages the mostly unusable
# "pickle" module rather than the mostly usable third-party "dill"
# module. Notably, we intentionally:
# * Leverage a "partial" object (which "pickle" silently supports)
# rather than a closure (which "pickle" rejects with
# non-human-readable exceptions).
# * Pass the actual low-level test function being tested (which "pickle"
# silently supports) rather than the high-level "pyfuncitem" object
# encapsulating that function (which "pickle" rejects with
# non-human-readable exceptions).
test_func_in_subprocess = partial(_test_func_in_subprocess, test_func)
# Python subprocess tasked with running this test.
test_subprocess = Process(target=test_func_in_subprocess)
# Begin running this test in this subprocess.
test_subprocess.start()
# Block this parent Python process until this test completes.
test_subprocess.join()
# If this subprocess reports non-zero exit status, this test failed. In
# this case...
if test_subprocess.exitcode != 0:
# Human-readable exception message to be raised.
exception_message = (
f'Test "{pyfuncitem.name}" failed in isolated subprocess with:')
# Raise a pytest-compliant exception.
raise fail(exception_message, pytrace=False)
# Else, this subprocess reports zero exit status. In this case, this
# test succeeded.
# Notify pytest that this hook successfully ran this test.
return True
# Notify pytest that this hook avoided attempting to run this test, in which
# case pytest will continue to look for a suitable runner for this test.
return None
# ....................{ PRIVATE ~ globals }....................
_MARK_NAME_SUBPROCESS = 'run_in_subprocess'
'''
**Subprocess mark** (i.e., name of our custom :mod:`pytest` mark, enabling tests
to notify the :func:`.pytest_pyfunc_call` hook that they require isolation to a
Python subprocess of the current Python process).
'''
# ....................{ PRIVATE ~ classes }....................
class _UnbufferedOutputStream(object):
'''
**Unbuffered standard output stream** (i.e., proxy object encapsulating a
buffered standard output stream by forcefully flushing that stream on all
writes to that stream).
See Also
--------
https://github.com/ansible/pytest-mp/issues/15#issuecomment-1342682418
GitHub comment by @pelson (Phil Elson) strongly inspiring this class.
'''
def __init__(self, stream) -> None:
self.stream = stream
def write(self, data) -> None:
self.stream.write(data)
self.stream.flush()
def writelines(self, datas) -> None:
self.stream.writelines(datas)
self.stream.flush()
def __getattr__(self, attr: str) -> object:
return getattr(self.stream, attr)
# ....................{ PRIVATE ~ functions }....................
def _test_func_in_subprocess(test_func: Callable) -> object:
'''
Run the passed :mod:`pytest` test function isolated to a Python subprocess
of the current Python process.
Parameters
----------
test_func : Callable
Test function to be run.
Returns
-------
object
Arbitrary object returned by this test if any *or* :data:`None`.
'''
# Defer subpracess-specific imports.
import sys
# Monkey-patch the unbuffered standard error and output streams of this
# subprocess with buffered equivalents, ensuring that pytest will reliably
# capture *all* standard error and output emitted by running this test.
sys.stderr = _UnbufferedOutputStream(sys.stderr)
sys.stdout = _UnbufferedOutputStream(sys.stdout)
# Run this test and return the result of doing so.
return test_func()
|
