如何使使用 pytest 的扭曲测试超时？

Question

我正在努力将一些测试从使用 Nose 和 twisted 转换为使用 Pytest 和 twisted，因为 Nose 不再处于开发阶段。转换测试的最简单方法是编辑每个测试具有的自定义装饰器。这个装饰器在每个测试中，并为单个测试定义超时。

我尝试过使用@pytest.mark.timeout，但唯一有效的方法是“线程”方法，但这会停止整个测试运行并且不会继续进行下一个测试。使用“信号”方法无法停止测试，但我可以看到 junitxml 文件中存在错误。

def inlineCallbacksTest ( timeout = None ):

    def decorator ( method ):
        @wraps ( method )
        @pytest.mark.timeout(timeout = timeout, method = 'signal' )
        @pytest.inlineCallbacks
        def testMethod ( *args, **kwargs ):
            return method(*args, **kwargs)           
        return testMethod
    return decorator

测试本身使用twisted 来启动并向软件发送消息。我不需要测试来取消任何扭曲的进程或锁。我希望 pytest 在超时后将测试标记为失败，然后进行下一个测试。

以下是我在使用超时信号方法时在 xml 文件中看到的错误。

</system-out><system-err>
+++++++++++++++++++++++++++++++++++ Timeout ++++++++++++++++++++++++++++++++++++

~~~~~~~ Stack of PoolThread-twisted.internet.reactor-2 (139997693642496) ~~~~~~~
  File &quot;/usr/lib64/python2.7/threading.py&quot;, line 784, in __bootstrap
    self.__bootstrap_inner()
  File &quot;/usr/lib64/python2.7/threading.py&quot;, line 811, in __bootstrap_inner
    self.run()
  File &quot;/usr/lib64/python2.7/threading.py&quot;, line 764, in run
    self.__target(*self.__args, **self.__kwargs)
  File &quot;/usr/lib64/python2.7/site-packages/twisted/python/threadpool.py&quot;, line 190, in _worker
    o = self.q.get()
  File &quot;/usr/lib64/python2.7/Queue.py&quot;, line 168, in get
self.not_empty.wait()
  File &quot;/usr/lib64/python2.7/threading.py&quot;, line 339, in wait
waiter.acquire()

~~~~~~~ Stack of PoolThread-twisted.internet.reactor-1 (139997702035200) ~~~~~~~
      File &quot;/usr/lib64/python2.7/threading.py&quot;, line 784, in __bootstrap
        self.__bootstrap_inner()
  File &quot;/usr/lib64/python2.7/threading.py&quot;, line 811, in __bootstrap_inner
    self.run()
  File &quot;/usr/lib64/python2.7/threading.py&quot;, line 764, in run
    self.__target(*self.__args, **self.__kwargs)
  File &quot;/usr/lib64/python2.7/site-packages/twisted/python/threadpool.py&quot;, line 190, in _worker
    o = self.q.get()
  File &quot;/usr/lib64/python2.7/Queue.py&quot;, line 168, in get
    self.not_empty.wait()
  File &quot;/usr/lib64/python2.7/threading.py&quot;, line 339, in wait
    waiter.acquire()

+++++++++++++++++++++++++++++++++++ Timeout ++++++++++++++++++++++++++++++++++++
Unhandled Error
Traceback (most recent call last):
  File &quot;/usr/lib64/python2.7/site-
packages/twisted/internet/base.py&quot;, line 1169, in run
    self.mainLoop()
--- &lt;exception caught here&gt; ---
  File &quot;/usr/lib64/python2.7/site-
packages/twisted/internet/base.py&quot;, line 1181, in mainLoop
    self.doIteration(t)
  File &quot;/usr/lib64/python2.7/site-
packages/twisted/internet/epollreactor.py&quot;, line 362, in doPoll
    l = self._poller.poll(timeout, len(self._selectables))
  File &quot;/usr/lib/python2.7/site-packages/pytest_timeout.py&quot;, line 110, in handler
    timeout_sigalrm(item, timeout)
  File &quot;/usr/lib/python2.7/site-packages/pytest_timeout.py&quot;, line 243, in timeout_sigalrm
    pytest.fail(&apos;Timeout &gt;%ss&apos; % timeout)
  File &quot;/usr/lib/python2.7/site-packages/_pytest/outcomes.py&quot;, line 85, in fail
    raise Failed(msg=msg, pytrace=pytrace)
builtins.Failed: Timeout &gt;5.0s
</system-err>

我四处寻找类似的解决方案，我能找到的最接近的是question。任何帮助或建议将不胜感激。

Answer 1

4 年多后回到这个问题并给出答案。问题似乎是测试被扭曲反应堆捕获的例外。我能够通过更新扭曲的版本来解决这个问题。自 16.5 起的 Twisted 版本有一个新的延迟函数调用 addTimeout (Docs)。使用它，我能够将原始装饰器修改为以下内容。现在，每当测试超时时，它只会引发异常并继续进行下一个异常。可能不是最优雅的，但我希望这对其他人有帮助！

    import twisted.internet.defer as defer
    import pytest_twisted as pt
    from functools import wraps
    
    def inlineCallbacksTest ( timeout = None ):
    
        def testDecorator ( testFunc ):


            def timeoutError ( value, timeout ):

                raise Exception ( "Test Timeout: {} secs have expired".format ( timeout ) ) 
    
            @wraps ( testFunc )
            def wrapper ( *args, **kwargs ):

                testDefer = pt.inlineCallbacks ( testFunc )( *args, **kwargs )
                testDefer.addTimeout ( timeout, reactor, timeoutError )
                return testDefer
    
            return wrapper
        return testDecorator