python中的模拟线程模块

Question

我正在尝试模拟一个函数，该函数使用多线程运行另一个具有不同参数的函数，并将返回结果保存到队列中。我尝试使用pytest和unitest来模拟它，但当从测试函数调用时，它似乎仍然执行线程

from threading import Thread
import threading
import time
import queue
from unittest import mock
def threaded_function(name):
time.sleep(100)
return name
def run_threads():
thread_list = []
result_list = []
res_queue = queue.Queue()
args_list = [("A"), ("B"), ("C")]
for val in args_list:
thread = Thread(target=lambda q, arg1: q.put(threaded_function(arg1)), args=(res_queue, val))
thread.start()
thread_list.append(thread)
for thread in thread_list:
thread.join()
while not res_queue.empty():
result_list.append(res_queue.get())
return result_list

以下是我正在尝试的模拟函数：

@mock.patch("threading.Thread")
@mock.patch("queue.Queue")
def test_run_threads(mock_queue, mock_thread):
new_queue = queue.Queue()
new_queue.put("D")
mock_queue.return_value = new_queue
mock_thread.return_value = None
result = run_threads()
assert result == ["D"]

class MockThread:
def __init__(self):
pass
def start():
pass
def join():
pass

def test_run_threads2(monkeypatch):
mock_thread = MockThread()
monkeypatch.setattr(threading, "Thread", MockThread)
result = run_threads()
assert result == []

Answer 1

根据Unittest：在哪里进行补丁，您需要从使用它的地方(或查找它的地方(对Thread进行补丁。在函数run_threads中，由于from threading import Thread的导入方式，您使用的是__main__.Threads而不是threading.Threads。删除mock_thread.return_value = None，现在run_threads中的所有线程都将是不执行任何功能的MagicMock。

您的下一个问题是在run_threads中嘲笑res_queue。在test_run_threads中对其进行修补时，您无法用不同的队列替换res_queue，因为您只是用MagicMock替换了queue.Queue的所有新实例。

最好重写这个函数，以便更容易测试。我建议将run_threads()分解为两个函数。

create_thread_list(args_list, res_queue):将用于创建我们的线程列表。通过将其分离，我们可以将args_list更改为我们想要测试的任何参数列表。

def create_thread_list(args_list, res_queue):
thread_list = []
for val in args_list:
thread = Thread(target=lambda q, arg1: q.put(threaded_function(arg1)), args=(res_queue, val))
thread_list.append(thread)
return thread_list

run_threads_2(thread_list, res_queue):将用于启动线程。

def run_threads_2(thread_list, res_queue):
result_list = []
for th in thread_list:
th.start()
for th in thread_list:
th.join()
while not res_queue.empty():
result_list.append((res_queue.get()))
return result_list

通过将它们分离出来，您可以传递任何要为线程测试的参数。

下面是我现在将如何测试的一些例子：

import queue
import time
from unittest.mock import patch
class MockThread2:
def __init__(self, name, result_q):
self.name = name
self.result_q = result_q
def start(self):
self.result_q.put(self.name)
def join(self):
pass
class TestMultiThreadedFunctions(unittest.TestCase):
def test_run_threads_2(self):
arg_list = ['A', 'B', 'C']
result_q = queue.Queue()
# Testing if created threads actually call the target function
# without actually calling the function.
with patch('__main__.threaded_function') as mock_function:
thread_list = create_thread_list(args_list=arg_list, res_queue=result_q)
run_threads_2(thread_list=thread_list, res_queue=result_q)
# Check if all threads ran
self.assertEqual(len(arg_list), mock_function.call_count)
arg_list = ['C', 'A', 'D', 'B', 'E']
result_q = queue.Queue()
# Using the threaded function, but just patching sleep
with patch('time.sleep') as mock_sleep:
thread_list = create_thread_list(args_list=arg_list, res_queue=result_q)
result_list = run_threads_2(thread_list=thread_list, res_queue=result_q)
self.assertListEqual(arg_list, result_list)
def test_run_with_alternate_threads(self):
# testing with MockThread and expecting nothing in the result_q
result_q = queue.Queue()
thread_list = [MockThread() for _ in range(5)]
expected_list = []
result_list = run_threads_2(thread_list=thread_list, res_queue=result_q)
self.assertListEqual(expected_list, result_list)
# testing with MockThread2
result_q = queue.Queue()
thread_list = [MockThread2(str(name), result_q) for name in range(5)]
expected_list = ['0', '1', '2', '3', '4']
result_list = run_threads_2(thread_list=thread_list, res_queue=result_q)
self.assertListEqual(expected_list, result_list)