Marking a test to be run in subprocess

[pchanial]

Hello, I tried to use pytest internals to create a marker whose goal is to run a test in a subprocess.
pytest-forked is not a solution, because the required test involves CUDA initialization that can only be done in a subprocess and pytest-xdist runs all the tests in a subprocess.

This is the current implementation :

@pytest.hookimpl(tryfirst=True)
def pytest_runtest_protocol(item: pytest.Item):
    if not item.get_closest_marker('insubprocess'):
        return  # normal test: standard execution
    if os.environ.get('PYTEST_IN_SUBPROCESS') == '1':
        return
    ihook = item.ihook
    ihook.pytest_runtest_logstart(nodeid=item.nodeid, location=item.location)
    reports = in_subprocess_run_report(item)
    for rep in reports:
        ihook.pytest_runtest_logreport(report=rep)
    ihook.pytest_runtest_logfinish(nodeid=item.nodeid, location=item.location)
    return True


def in_subprocess_run_report(item: pytest.Item):
    # Execution in subprocess
    env = os.environ.copy()
    env['PYTEST_IN_SUBPROCESS'] = '1'
    result = subprocess.run(
        [sys.executable, '-m', 'pytest', '--capture=no', item.nodeid],
        env=env,
        capture_output=True,
        encoding='utf-8',
    )

    keywords = {keyword: 1 for keyword in item.keywords}
    outcome = 'passed' if result.returncode == 0 else 'failed'

    sections = []
    if result.stdout:
        sections.append(('captured stdout', result.stdout))
    if result.stderr:
        sections.append(('captured stderr', result.stderr))

    reports = [
        runner.TestReport(
            item.nodeid,
            item.location,
            keywords,
            outcome,
            longrepr=None,
            when='call',
            sections=sections,
        )
    ]
    return reports

But using pytest 8.3.5, I get the following error message when running the next test:

self = <_pytest.runner.SetupState object at 0x748a287cb430>, item = <Function test_rmul>

    def setup(self, item: Item) -> None:
        """Setup objects along the collector chain to the item."""
        needed_collectors = item.listchain()
    
        # If a collector fails its setup, fail its entire subtree of items.
        # The setup is not retried for each item - the same exception is used.
        for col, (finalizers, exc) in self.stack.items():
>           assert col in needed_collectors, "previous item was not torn down properly"
E           AssertionError: previous item was not torn down properly

venv/lib/python3.10/site-packages/_pytest/runner.py:505: AssertionError

The runner's stack is:

(Pdb) for k, v in self.stack.items(): print(k, v)
<Session  exitstatus=<ExitCode.OK: 0> testsfailed=1 testscollected=16> ([<bound method Node.teardown of <Session  exitstatus=<ExitCode.OK: 0> testsfailed=1 testscollected=16>>, functools.partial(<bound method FixtureDef.finish of <FixtureDef argname='enable_x64' scope='session' baseid='tests'>>, request=<SubRequest 'enable_x64' for <Function test_inverse[IdentityOperator]>>)], None)
<Dir furax> ([<bound method Node.teardown of <Dir furax>>], None)
<Package tests> ([<bound method Node.teardown of <Package tests>>], None)
<Package core> ([<bound method Node.teardown of <Package core>>], None)
<Package base> ([<bound method Node.teardown of <Package base>>], None)
<Module test_inverse.py> ([<bound method Node.teardown of <Module test_inverse.py>>], None)

Note that the marked tests is in test_inverse.py, and the test that produce the error is in another module.

How should I propertly teardown the test, I've looked at pytest-forked implementation, but I didn't see any teardown.

[RonnyPfannschmidt]

Xdist is currently the only game in town that correctly passes reports between processes

It currently has no feature to spin up processes for single tests

Forked is a hack that can't quite work correctly which is why it was taken out of xdist and put into the unmaintained state