make PYTEST_CURRENT_TEST thread-safe

Author: Liam-DeVoe

.gitignore

@@ -16,7 +16,6 @@ include/
 *.class
 *.orig
 *~
-.hypothesis/
 
 # autogenerated
 src/_pytest/_version.py
@@ -51,6 +50,7 @@ coverage.xml
 .vscode
 __pycache__/
 .python-version
+.claude
 
 # generated by pip
 pip-wheel-metadata/

changelog/13837.improvement.rst

@@ -0,0 +1 @@
+Pytest setting :ref:`PYTEST_CURRENT_TEST <pytest current test env>` internally is now thread-safe.

src/_pytest/runner.py

@@ -208,7 +208,13 @@ def _update_current_test_var(
         value = value.replace("\x00", "(null)")
         os.environ[var_name] = value
     else:
-        os.environ.pop(var_name)
+        # under multithreading, this may have already been popped by another thread.
+        # Note that os.environ inherits from MutableMapping and therefore .pop(var_name, None)
+        # is not atomic or thread-safe, unlike e.g. popping from a builtin dict.
+        try:
+            os.environ.pop(var_name)
+        except KeyError:
+            pass
 
 
 def pytest_report_teststatus(report: BaseReport) -> tuple[str, str, str] | None:

testing/threading/test_threading.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from _pytest.pytester import Pytester
+
+
+# a conftest that runs tests in a threadpool. You can think of this as "like pytest-xdist,
+# but using threads instead of processes". This is a nice conftest for flushing out
+# concurrency bugs in pytest itself.
+threaded_conftest = """
+import sys
+from queue import Queue, Empty
+from concurrent.futures import ThreadPoolExecutor
+
+# make thread switches more common
+sys.setswitchinterval(0.000001)
+n = 3
+
+def pytest_runtestloop(session):
+    queue = Queue()
+    for item in session.items:
+        queue.put(item)
+
+    def worker():
+        try:
+            item = queue.get_nowait()
+        except Empty:
+            return
+
+        while item is not None:
+            try:
+                next_item = queue.get_nowait()
+            except Empty:
+                next_item = None
+
+            item.config.hook.pytest_runtest_protocol(
+                item=item, nextitem=next_item
+            )
+            item = next_item
+
+    with ThreadPoolExecutor() as executor:
+        futures = [executor.submit(worker) for _ in range(n)]
+        for future in futures:
+            future.result()
+    return True
+"""
+
+
+def test_concurrent(pytester: Pytester) -> None:
+    pytester.makeconftest(threaded_conftest)
+    pytester.makepyfile(
+        """
+        import pytest
+
+        def do_work():
+            # arbitrary moderately-expensive work
+            for x in range(500):
+                _y = x**x
+
+        def test_1(): do_work()
+        def test_2(): do_work()
+        def test_3(): do_work()
+        def test_4(): do_work()
+        """
+    )
+    result = pytester.runpytest()
+    result.assert_outcomes(passed=4)