在python中模拟线程模块

如何解决在python中模拟线程模块

我正在尝试模拟一个函数，该函数使用多线程运行另一个具有不同参数的函数并将返回结果保存到队列中。我尝试使用 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 == []

解决方法

根据Unittest: Where to patch，您需要从使用它的地方（或查找它的地方）修补Thread。在您的函数 run_threads 中，由于 __main__.Threads 的导入方式，您使用的是 threading.Threads 而不是 from threading import Thread。删除 mock_thread.return_value = None，现在您在 run_threads 中的所有线程都将成为不执行任何功能的 MagicMock。

您的下一个问题是在 res_queue 中模拟 run_threads。在 test_run_threads 中修补它时，您无法将 res_queue 替换为不同的队列，因为您只是将 queue.Queue 的所有新实例替换为 MagicMock。

最好重写此函数以更易于测试。 我建议将 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','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,result_list)