Make function-scope fixtures run in the same thread as the test function itself

[ogrisel]

Currently, pytest fixtures seem to always be executed in the main thread. This prevents initializing thread-local configuration to run a particular test in isolation of concurrently running tests with different thread-local settings.

See the reproducer below:

# test_simple.py
import threading

import pytest

thread_local = threading.local()


@pytest.fixture
def change_thread_local():
    print("inside fixture", threading.current_thread())
    setattr(thread_local, "value", 1)
    yield


def test_simple(change_thread_local):
    print("inside function", threading.current_thread())
    assert getattr(thread_local, "value", None) == 1

pytest --threads 1 --iterations 1 test_simple.py

resulting in:

============================================================================================================= test session starts ==============================================================================================================
platform linux -- Python 3.13.0rc3, pytest-8.3.3, pluggy-1.5.0
rootdir: /tmp/proj
plugins: xdist-3.6.1, freethreaded-0.1.0
collected 1 item                                                                                                                                                                                                                               

test_simple.py F                                                                                                                                                                                                                         [100%]

=================================================================================================================== FAILURES ===================================================================================================================
_________________________________________________________________________________________________________________ test_simple __________________________________________________________________________________________________________________

change_thread_local = None

    def test_simple(change_thread_local):
        print("inside function", threading.current_thread())
>       assert getattr(thread_local, "value", None) == 1
E       AssertionError: assert None == 1
E        +  where None = getattr(<_thread._local object at 0x7b3843ff32e0>, 'value', None)

test_simple.py:17: AssertionError
------------------------------------------------------------------------------------------------------------ Captured stdout setup -------------------------------------------------------------------------------------------------------------
inside fixture <_MainThread(MainThread, started 135481613739584)>
------------------------------------------------------------------------------------------------------------- Captured stdout call -------------------------------------------------------------------------------------------------------------
inside function <Thread(ThreadPoolExecutor-0_0, started 135481585043136)>
=========================================================================================================== short test summary info ============================================================================================================
FAILED test_simple.py::test_simple - AssertionError: assert None == 1
============================================================================================================== 1 failed in 0.03s ===============================================================================================================

Would it be possible to execute function-scope fixtures on the same thread as the test function?

Originally discussed in the context of testing scikit-learn in free threading mode: scikit-learn/scikit-learn#30007

[lesteve]

I bumped into pytest-parallel that has been archived and is no longer maintained (latest release in October 2021 and see kevlened/pytest-parallel#104 (comment)), but seems to run the fixture and the test function in the same thread. The following command runs fine:

pytest --tests-per-worker 2 test_simple.py

See their plugin code.

Full disclosure, I had to edit the plugin code to avoid an AttributeError (probably something has changed in pytest since):

self._log = print # py.log.Producer('pytest-parallel')

[ogrisel]

As discussed in the linked scikit-learn issue, it's possible to convert such thread-sensitive fixtures to regular function decorators and the problems go away.

So we have a workaround.

[ogrisel]

Note that the above workaround is not always valid. For instance, the standard tmpdir fixture does no longer ensure working in an isolated temporary directory for a parametrized test:

@pytest.mark.parametrize("value", range(10))
def test_stuff_in_isolated_folder(value, tmpdir):
    subfolder = tmpdir.mkdir("subfoler")  # fails with py.error.EEXIST: [File exists]...
    # Do test stuff in "subfolder" and "value".

[ogrisel]

While tmpdir is deprecated, the problem stays the same with its official tmp_path replacement:

import pytest


@pytest.mark.parametrize("value", list("abcd"))
def test_simple(tmp_path, value):
    file_path = tmp_path / "file.txt"
    assert not file_path.exists()
    file_path.write_text(value)
    read_value = file_path.read_text()
    assert read_value == value

$ pytest --threads 2 -x
=================================== test session starts ===================================
platform darwin -- Python 3.13.0, pytest-8.3.3, pluggy-1.5.0
rootdir: /private/tmp/testfolder
plugins: freethreaded-0.1.0
collected 4 items                                                                         

test_tmp_path.py F

======================================== FAILURES =========================================
_____________________________________ test_simple[a] ______________________________________

tmp_path = PosixPath('/private/var/folders/_y/lfnx34p13w3_sr2k12bjb05w0000gn/T/pytest-of-ogrisel/pytest-11/test_simple_a_0')
value = 'a'

    @pytest.mark.parametrize("value", list("abcd"))
    def test_simple(tmp_path, value):
        file_path = tmp_path / "file.txt"
>       assert not file_path.exists()
E       AssertionError: assert not True
E        +  where True = exists()
E        +    where exists = PosixPath('/private/var/folders/_y/lfnx34p13w3_sr2k12bjb05w0000gn/T/pytest-of-ogrisel/pytest-11/test_simple_a_0/file.txt').exists

test_tmp_path.py:7: AssertionError

The above exception was the direct cause of the following exception:

item = <Function test_simple[a]>

    @pytest.hookimpl()
    def pytest_runtest_call(item: pytest.Item):
        # Try item.runtest()
        config_threads = item.config.option.threads
        config_iterations = item.config.option.iterations
        freethreaded_mark = item.get_closest_marker(name="freethreaded")
    
        if freethreaded_mark:
            threads = freethreaded_mark.kwargs.get("threads", config_threads)
            iterations = freethreaded_mark.kwargs.get("iterations", config_iterations)
        else:
            iterations = config_iterations
            threads = config_threads
    
        logger.debug("Running test %s", item.name)
        executor = ThreadPoolExecutor(max_workers=threads)
        barrier = threading.Barrier(threads)
        last_round = iterations % threads
        last_barrier = threading.Barrier(last_round) if last_round else None
        results = list(
            executor.map(
                get_one_result,
                repeat(item, iterations),
                chain(
                    repeat(barrier, iterations - last_round),
                    repeat(last_barrier, last_round),
                ),
            )
        )
        exceptions = [r for r in results if isinstance(r, Exception)]
        if not exceptions:
            return results[0]
        if len(exceptions) == len(results):
            raise results[0]
>       raise ConcurrencyError(
            iterations=iterations, failures=len(exceptions), threads=threads
        ) from exceptions[0]
E       pytest_freethreaded.plugin.ConcurrencyError: 199 failures in 200 iterations across 2 threads

/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/pytest_freethreaded/plugin.py:104: ConcurrencyError
================================= short test summary info =================================
FAILED test_tmp_path.py::test_simple[a] - pytest_freethreaded.plugin.ConcurrencyError: 199 failures in 200 iterations across 2 t...
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
==================================== 1 failed in 0.03s ====================================

[ogrisel]

Note that this is a different problem than that reported in #19 as tmp_path utilization would be perfectly thread-safe it that fixture what initialized in the same thread that used to run the test it is used for instead of being initialized in the main pytest thread.

[ogrisel]

I wanted to see if the old, unmaintained pytest-parallel plugin (with @lesteve's patch) would handle the tmp_path fixture in a better way.

But it does not work either (most of the time), but it's broken in a different manner:

$ pytest --tests-per-worker 4 test_tmp_path.py

/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/pytest_parallel/__init__.py:221: PytestDeprecationWarning: The hookimpl ParallelRunner.pytest_sessionstart uses old-style configuration options (marks or attributes).
Please use the pytest.hookimpl(tryfirst=True) decorator instead
 to configure the hooks.
 See https://docs.pytest.org/en/latest/deprecations.html#configuring-hook-specs-impls-using-markers
  @pytest.mark.tryfirst
========================================================================== test session starts ==========================================================================
platform darwin -- Python 3.13.0, pytest-8.3.3, pluggy-1.5.0
rootdir: /private/tmp/testfolder
plugins: repeat-0.9.3, parallel-0.1.1, xdist-3.6.1
collected 4 items                                                                                                                                                       
pytest-parallel: 1 worker (process), 4 tests per worker (threads)
......EE
================================================================================ ERRORS =================================================================================
___________________________________________________________________ ERROR at setup of test_simple[c] ____________________________________________________________________

cls = <class '_pytest.runner.CallInfo'>, func = <function call_and_report.<locals>.<lambda> at 0x101dc5080>, when = 'setup'
reraise = (<class '_pytest.outcomes.Exit'>, <class 'KeyboardInterrupt'>)

    @classmethod
    def from_call(
        cls,
        func: Callable[[], TResult],
        when: Literal["collect", "setup", "call", "teardown"],
        reraise: type[BaseException] | tuple[type[BaseException], ...] | None = None,
    ) -> CallInfo[TResult]:
        """Call func, wrapping the result in a CallInfo.
    
        :param func:
            The function to call. Called without arguments.
        :type func: Callable[[], _pytest.runner.TResult]
        :param when:
            The phase in which the function is called.
        :param reraise:
            Exception or exceptions that shall propagate if raised by the
            function, instead of being wrapped in the CallInfo.
        """
        excinfo = None
        start = timing.time()
        precise_start = timing.perf_counter()
        try:
>           result: TResult | None = func()

/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/runner.py:341: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/runner.py:242: in <lambda>
    lambda: runtest_hook(item=item, **kwds), when=when, reraise=reraise
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/pluggy/_hooks.py:513: in __call__
    return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/pluggy/_manager.py:120: in _hookexec
    return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/unraisableexception.py:90: in pytest_runtest_setup
    yield from unraisable_exception_runtest_hook()
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/unraisableexception.py:70: in unraisable_exception_runtest_hook
    yield
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/logging.py:840: in pytest_runtest_setup
    yield from self._runtest_for(item, "setup")
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/logging.py:829: in _runtest_for
    yield
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/capture.py:875: in pytest_runtest_setup
    return (yield)
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/threadexception.py:87: in pytest_runtest_setup
    yield from thread_exception_runtest_hook()
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/threadexception.py:68: in thread_exception_runtest_hook
    yield
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/runner.py:160: in pytest_runtest_setup
    item.session._setupstate.setup(item)
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/runner.py:514: in setup
    col.setup()
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/python.py:1630: in setup
    self._request._fillfixtures()
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/fixtures.py:697: in _fillfixtures
    item.funcargs[argname] = self.getfixturevalue(argname)
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/fixtures.py:532: in getfixturevalue
    fixturedef = self._get_active_fixturedef(argname)
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/fixtures.py:617: in _get_active_fixturedef
    fixturedef.execute(request=subrequest)
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/fixtures.py:1091: in execute
    result = ihook.pytest_fixture_setup(fixturedef=self, request=request)
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/pluggy/_hooks.py:513: in __call__
    return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/pluggy/_manager.py:120: in _hookexec
    return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/setuponly.py:36: in pytest_fixture_setup
    return (yield)
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/fixtures.py:1140: in pytest_fixture_setup
    result = call_fixture_func(fixturefunc, request, kwargs)
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/fixtures.py:891: in call_fixture_func
    fixture_result = next(generator)
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/tmpdir.py:273: in tmp_path
    path = _mk_tmp(request, tmp_path_factory)
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/tmpdir.py:253: in _mk_tmp
    return factory.mktemp(name, numbered=True)
/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/tmpdir.py:132: in mktemp
    basename = self._ensure_relative_to_basetemp(basename)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = TempPathFactory(_given_basetemp=None, _trace=<pluggy._tracing.TagTracerSub object at 0x101a2cfc0>, _basetemp=PosixPath...folders/_y/lfnx34p13w3_sr2k12bjb05w0000gn/T/pytest-of-ogrisel/pytest-65'), _retention_count=3, _retention_policy='all')
basename = 'test_simple_c_'

    def _ensure_relative_to_basetemp(self, basename: str) -> str:
        basename = os.path.normpath(basename)
        if (self.getbasetemp() / basename).resolve().parent != self.getbasetemp():
>           raise ValueError(f"{basename} is not a normalized and relative path")
E           ValueError: test_simple_c_ is not a normalized and relative path

/Users/ogrisel/miniforge3/envs/py313/lib/python3.13/site-packages/_pytest/tmpdir.py:114: ValueError
======================================================================== short test summary info ========================================================================
ERROR test_tmp_path.py::test_simple[c] - ValueError: test_simple_c_ is not a normalized and relative path
====================================================================== 3 passed, 1 error in 0.16s =======================================================================

EDIT: after editing the fixture code to report more info:

E           ValueError: test_simple_b_ is not a normalized and relative path:
E           (self.getbasetemp() / basename).resolve().parent = PosixPath('/private/var/folders/_y/lfnx34p13w3_sr2k12bjb05w0000gn/T/pytest-of-ogrisel/pytest-69')
E           self.getbasetemp() = PosixPath('/private/var/folders/_y/lfnx34p13w3_sr2k12bjb05w0000gn/T/pytest-of-ogrisel/pytest-69')

This makes no sense, I suspect that getbasetemp is not thread-safe.