From 8a20ce47f455a755a4df375fcae0603576fb19ec Mon Sep 17 00:00:00 2001 From: Valentin Gagarin Date: Mon, 9 Nov 2026 10:44:58 +0100 Subject: [PATCH 1/3] test(utils): Share the thread patching setup between threading tests --- tests/test_utils.py | 44 ++++++++++++++++++-------------------------- 1 file changed, 18 insertions(+), 26 deletions(-) diff --git a/tests/test_utils.py b/tests/test_utils.py index 77def0eb6e..75e4b9450d 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1180,50 +1180,42 @@ def target(): assert (thread.ident, thread.name) == results.get(timeout=1) -def test_get_current_thread_meta_bad_running_thread(): +def get_current_thread_meta_patched(**patch_kwargs): results = Queue(maxsize=1) def target(): - with mock.patch("threading.current_thread", side_effect=["fake thread"]): + with mock.patch("threading.current_thread", **patch_kwargs): results.put(get_current_thread_meta()) thread = threading.Thread(target=target) thread.start() thread.join() + return results.get(timeout=1) + +def test_get_current_thread_meta_bad_running_thread(): main_thread = threading.main_thread() - assert (main_thread.ident, main_thread.name) == results.get(timeout=1) + assert ( + main_thread.ident, + main_thread.name, + ) == get_current_thread_meta_patched(side_effect=["fake thread"]) def test_get_current_thread_meta_main_thread(): - results = Queue(maxsize=1) - - def target(): - # mock that somehow the current thread doesn't exist - with mock.patch("threading.current_thread", side_effect=[None]): - results.put(get_current_thread_meta()) - main_thread = threading.main_thread() - - thread = threading.Thread(target=target) - thread.start() - thread.join() - assert (main_thread.ident, main_thread.name) == results.get(timeout=1) + # mock that somehow the current thread doesn't exist + assert ( + main_thread.ident, + main_thread.name, + ) == get_current_thread_meta_patched(side_effect=[None]) def test_get_current_thread_meta_failed_to_get_main_thread(): - results = Queue(maxsize=1) - - def target(): - with mock.patch("threading.current_thread", return_value="fake thread"): - results.put(get_current_thread_meta()) - main_thread = threading.main_thread() - - thread = threading.Thread(target=target) - thread.start() - thread.join() - assert (main_thread.ident, main_thread.name) == results.get(timeout=1) + assert ( + main_thread.ident, + main_thread.name, + ) == get_current_thread_meta_patched(return_value="fake thread") @pytest.mark.parametrize( From fbcdcbddab84bd27bc9f69e795a832b877a63d46 Mon Sep 17 00:00:00 2001 From: Valentin Gagarin Date: Mon, 9 Nov 2026 10:45:23 +0100 Subject: [PATCH 2/3] test(utils): Expose that thread mocks break on other callers --- tests/test_utils.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/tests/test_utils.py b/tests/test_utils.py index 75e4b9450d..e2de5cec35 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1182,17 +1182,27 @@ def target(): def get_current_thread_meta_patched(**patch_kwargs): results = Queue(maxsize=1) + patched = threading.Event() + other_caller_done = threading.Event() def target(): with mock.patch("threading.current_thread", **patch_kwargs): + patched.set() + other_caller_done.wait() results.put(get_current_thread_meta()) thread = threading.Thread(target=target) thread.start() + patched.wait() + try: + thread.join(0) + finally: + other_caller_done.set() thread.join() return results.get(timeout=1) +@pytest.mark.xfail(strict=True, raises=StopIteration) def test_get_current_thread_meta_bad_running_thread(): main_thread = threading.main_thread() assert ( @@ -1201,6 +1211,7 @@ def test_get_current_thread_meta_bad_running_thread(): ) == get_current_thread_meta_patched(side_effect=["fake thread"]) +@pytest.mark.xfail(strict=True, raises=StopIteration) def test_get_current_thread_meta_main_thread(): main_thread = threading.main_thread() # mock that somehow the current thread doesn't exist From 69b2fa7dcacfb03e26cea627f3e6ed76490c2a56 Mon Sep 17 00:00:00 2001 From: Valentin Gagarin Date: Mon, 9 Nov 2026 10:45:44 +0100 Subject: [PATCH 3/3] test(utils): Make threading mocks answer only the thread under test --- tests/test_utils.py | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/tests/test_utils.py b/tests/test_utils.py index e2de5cec35..7d7e4b6993 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1180,13 +1180,19 @@ def target(): assert (thread.ident, thread.name) == results.get(timeout=1) -def get_current_thread_meta_patched(**patch_kwargs): +def get_current_thread_meta_patched(fake_thread): results = Queue(maxsize=1) patched = threading.Event() other_caller_done = threading.Event() + real_current_thread = threading.current_thread + + def current_thread(): + if threading.get_ident() == thread.ident: + return fake_thread + return real_current_thread() def target(): - with mock.patch("threading.current_thread", **patch_kwargs): + with mock.patch("threading.current_thread", current_thread): patched.set() other_caller_done.wait() results.put(get_current_thread_meta()) @@ -1202,23 +1208,21 @@ def target(): return results.get(timeout=1) -@pytest.mark.xfail(strict=True, raises=StopIteration) def test_get_current_thread_meta_bad_running_thread(): main_thread = threading.main_thread() assert ( main_thread.ident, main_thread.name, - ) == get_current_thread_meta_patched(side_effect=["fake thread"]) + ) == get_current_thread_meta_patched("fake thread") -@pytest.mark.xfail(strict=True, raises=StopIteration) def test_get_current_thread_meta_main_thread(): main_thread = threading.main_thread() # mock that somehow the current thread doesn't exist assert ( main_thread.ident, main_thread.name, - ) == get_current_thread_meta_patched(side_effect=[None]) + ) == get_current_thread_meta_patched(None) def test_get_current_thread_meta_failed_to_get_main_thread(): @@ -1226,7 +1230,7 @@ def test_get_current_thread_meta_failed_to_get_main_thread(): assert ( main_thread.ident, main_thread.name, - ) == get_current_thread_meta_patched(return_value="fake thread") + ) == get_current_thread_meta_patched("fake thread") @pytest.mark.parametrize(