Python多线程单元测试

Question

我使用多线程设计(别无选择),但我的大部分代码都驻留在一个线程中,其中的所有事件都通过队列进行管理.以这种方式,我的大多数代码都表现得好像是单线程的,我不必担心锁,信号量等等.

唉,我已经到了需要对我的代码进行单元测试的地方(请不要首先考虑不是TDDing),而且我不知所措 - 你如何在另一个线程中测试一些东西？

例如,假设我有以下课程:

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