如何为单元测试类创建超时包装器答案

【问题标题】：How to create timeout wrapper for unittest class如何为单元测试类创建超时包装器
【发布时间】：2018-07-10 23:48:03
【问题描述】：

我正在尝试使用这个问题作为模板为单元测试类创建一个超时包装器： PyUnit - How to unit test a method that runs into an infinite loop for some input?

如果从 Main 运行，上述解决方案效果很好，但我试图通过使用 unittest 类创建类似的超时，但它失败了。

PicklingError: Can't pickle : it's not found as TimeTest.test1

我愿意接受任何其他独立于平台并使用本机 2.7 库的解决方案

这里是代码

import unittest
import multiprocessing as mp
import functools as ft
import time

def timeout(seconds=5, error_message="Timeout"):
    def decorator(func):
        def wrapper(*args, **kwargs):
            process = mp.Process(None, func, None, args, kwargs)
            process.start()
            process.join(seconds)
            if process.is_alive():
                process.terminate()
                raise mp.TimeoutError(error_message)

        return ft.wraps(func)(wrapper)
    return decorator

class TestCases(unittest.TestCase):

    def setUp(self):
        self.a = 0

    @timeout(1)
    def test1(self):
        time.sleep(2)
        self.assertEquals(self.a, 0)

【问题讨论】：

标签： python-2.7 python-unittest

【解决方案1】：

我最终放弃了包装器的想法并创建了一个简单的函数和断言。

import unittest
import multiprocessing as mp
import time

TIME_LIMIT = 1

class TestCases(unittest.TestCase):

    def setUp(self):
        self.a = 0

    def my_func(self):
        time.sleep(2)
        self.q.put(self.a + 1)

    def run_case(self, func):
        self.q = mp.Queue()

        test_process = mp.Process(target=func)

        test_process.start()
        test_process.join(TIME_LIMIT)

        self.assertFalse(test_process.is_alive(), 'timeout exceeded')

    def test_case1(self):
        self.run_case(self.my_func)
        self.assertEquals(self.a + 1, self.q.get())

【讨论】：