我在 my_fuction.py 中有一个函数
def my_function():
# do something
print("doing something")
return True
我还有一个名为 my_thread.py 的 threading.Thread 继承类
from my_function import my_function
from threading import Thread, Event
import time
class MyThread(Thread):
def __init__(self):
super().__init__(name="MyThread")
self.running = Event()
def run(self):
self.running.set()
while self.running.is_set():
return_value = my_function()
print(return_value)
time.sleep(3)
我想测试模拟 my_function 的 MyThread run 方法,这样运行不会调用真正的 my_function 函数,而是调用模拟的函数
我有一个用于测试 MyThread 的单元测试脚本,名为 test_my_thread.py,如下所示:
from my_thread import MyThread
import unittest
import time
from unittest.mock import Mock, patch
def mock_my_function():
print("not doing something")
return False
class TestMyThread(unittest.TestCase):
def setUp(self):
self.my_thread = MyThread()
def test_my_thread_run_method(self):
with patch("my_function.my_function", side_effect=mock_my_function) as mock_func:
with patch.object(MyThread, "my_function", mock_my_function) as patched_my_function:
self.my_thread.run()
time.sleep(5)
self.my_thread.running.clear()
self.my_thread.join()
patched_my_function.assert_called_once() # this method will fail which i assume is because the function was not patched in the MyThread.run method were it is used
我如何在 MyThread 的运行中模拟“my_function”,或者是否有其他方法来测试它是否正在运行而不实际运行它(模拟它的替代方法)。
MyThread 已经导入了 my_function.my_function,现在它有了自己对该函数的引用,这就是 MyThread 将调用的函数。这就是您需要修补的参考。它是“my_thread.my_function”
您犯了一个常见的错误,即尝试修补函数定义的位置,但修补程序只是替换了名称,而不是实际的函数。您需要修补的是“my_thread.my_function”
以下是其他一些不相关的更改:
您在线程上调用 run(),而不是 start(),因此单元测试会等待并且永远不会进入睡眠或加入状态。
我还更改了睡眠时间,因为您的测试导致mock_func被调用两次,因此assert_used_once()正确失败并显示消息
“预计‘my_function’已被调用一次。被调用了 2 次。 调用:[call(), call()]。”
我需要添加对 unittest.main() 的调用,以便我可以使用“python test_my_thread.py”开始测试,您可能以另一种方式开始测试,所以忽略它。
import unittest
import time
from unittest.mock import patch
from my_thread import MyThread
def mock_my_function():
print("mock function called")
return False
class TestMyThread(unittest.TestCase):
def setUp(self):
self.my_thread = MyThread()
def test_my_thread_run_method(self):
with patch("my_thread.my_function", side_effect=mock_my_function) as mock_func:
self.my_thread.start()
time.sleep(1)
self.my_thread.running.clear()
self.my_thread.join()
mock_func.assert_called_once()
if __name__ == "__main__":
unittest.main()