Python多线程单元测试

Question

我使用多线程设计（别无选择），但我的大部分代码驻留在单个线程中，其中的所有事件都通过 队列。以这种方式，我的大部分代码的行为就好像它是单线程的，而且我不必担心锁、信号量等等。

唉，我已经到了需要对我的代码进行单元测试的地步（请不要批评我一开始就没有 TDD），而我不知所措 - 如何在另一个线程中测试某些东西？

例如，假设我有以下类：

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

如果您的设计假设所有内容都必须经过队列，那么不要与之对抗 - 让所有内容都经过队列！

将 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})

Answer 2

您可以查看 test_threading.py stdlib 测试执行与您尝试执行的操作类似的操作。基本思想是使用互斥锁和信号量保护线程执行，以便在断言测试条件之前完成执行。