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() 对于队列线程内的任何内容都可以正常工作，但是在主线程中由测试异步调用它，因此结果可能不正确！

- Jonathan Livni

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

- Jonathan Livni

---

这可能是我在SO上看到的最实用（即不仅仅是重复目标）的自问自答问题。 - Jared Smith

Answer 2

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

- Senthil Kumaran

---

你知道这方面的教程或文档吗？ - Jonathan Livni

我所链接的代码本身更易于理解，你可以在python.org文档中查找线程文档以获得一些帮助。 - Senthil Kumaran