Python单元测试补丁在多线程时不起作用

Question

我试图为每个线程用不同的返回值修补资源，但有时修补后的函数在其他线程可以使用它之前就变成了未修补的。

file1.py:

import subprocess
def dont_run_me() -> str:
result = subprocess.run('rm -rf /'.split(), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
if result.returncode != 0:
raise FileNotFoundException(
f'Something went wrong while deleting everything, I wonder why: {result.stderr.decode()}')
else:
return 'Deleted everything'
def function_being_tested(identifier: str) -> str:
return dont_run_me()

测试文件1.py

import unittest
from unittest.mock import patch
from concurrent.futures import ThreadPoolExecutor
from file1 import function_being_tested
class TestFunction(unittest.TestCase):
def assert_function_not_called(self, thread_name):
with patch('file1.function_being_tested') as function_being_tested_patch:
function_being_tested_patch.return_value = thread_name
result = function_being_tested(thread_name)
print(result)
self.assertEqual(t1_response, thread_name)
def test_function_being_tested(self):
executor = ThreadPoolExecutor()
thread_1 = executor.submit(self.assert_function_not_called, 'thread_1')
thread_2 = executor.submit(self.assert_function_not_called, 'thread_2')
thread_1.result()
thread_2.result()

成功时的结果：

thread_1
thread_2

失败时的结果：

thread_1
Something went wrong while deleting everything, I wonder why: ...

在其他地方，我看到补丁不是线程安全的：https://stackoverflow.com/a/26877522/5263074我如何确保每个线程都可以用不同的返回值修补函数，而必须进行模拟。

Answer 1

由于补丁会全局修改，因此线程内的补丁不会是线程安全的。

与其在线程的上下文中进行嘲讽，不如在创建线程之前进行修补。利用side_effect根据输入返回不同的值。

举例：

import unittest
from unittest.mock import patch
from concurrent.futures import ThreadPoolExecutor
from file1 import function_being_tested
class TestFunction(unittest.TestCase):
def assert_function_not_called(self, thread_name):
result = function_being_tested(thread_name)
print(thread_name)
self.assertEqual(t1_response, thread_name)
def test_function_being_tested(self):
def _mocked_dont_run_me(thread_name: str):
return thread_name
with patch('file1.dont_run_me', side_effect=_mocked_dont_run_me):
executor = ThreadPoolExecutor()
thread_1 = executor.submit(self.assert_function_not_called, 'thread_1')
thread_2 = executor.submit(self.assert_function_not_called, 'thread_2')
thread_1.result()
thread_2.result()