Python多线程单元测试

Question

我使用了多线程设计（没办法），但大部分代码都在一个线程里运行，所有事件都是通过一个队列来管理。这样一来，我的代码大部分表现得就像是单线程的，我就不用担心锁、信号量之类的问题。

可惜的是，我现在需要对我的代码进行单元测试（请不要因为我一开始没有采用测试驱动开发而责怪我），但我不知道该怎么做——怎么测试另一个线程里的东西呢？

比如，我有一个这样的类：

class MyClass():
    def __init__(self):
        self.a=0
        # register event to self.on_event

    def on_some_event(self, b):
        self.a += b

    def get(self):
        return self.a

我想测试：

import unittest
from queued_thread import ThreadedQueueHandler

class TestMyClass(unittest.TestCase):
    def setUp(self):
        # create the queued thread and assign the queue to self.queue

    def test_MyClass(self):
        mc = MyClass()
        self.queue.put({'event_name':'some_event', 'val':1})
        self.queue.put({'event_name':'some_event', 'val':2})
        self.queue.put({'event_name':'some_event', 'val':3})
        self.assertEqual(mc.get(),6)

if __name__ == '__main__':
    unittest.main()

MyClass.get() 在队列线程里运行得很好，但在测试中，它会被主线程异步调用，这样结果可能就不正确了！

Answer 1

你可以看看这个链接中的 test_threading.py 文件，它在标准库的测试中做了类似的事情。基本的想法是用互斥锁和信号量来保护线程的执行，确保在检查测试条件之前，线程的执行是完整的。

Answer 2

如果你的设计是让所有东西都必须通过队列，那就别纠结了——让所有东西都通过队列吧！

给你的队列事件处理器添加一个 on_call 事件，然后把下面这个函数注册到这个事件上：

def on_call(self, callback):
    callback()

然后把你的测试修改成：

def test_MyClass(self):
    def threaded_test():
        self.assertEqual(mc.get(),6)

    mc = MyClass()
    self.queue.put(1)
    self.queue.put(2)
    self.queue.put(3)
    self.queue.put({'event_name':'call','val':threaded_test})