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

How do you timeout a twisted test that uses pytest?

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

我试过使用 @pytest.mark.timeout,但唯一有效的方法是 'thread' 方法,但这会停止整个测试 运行 并且不会继续下一次测试。使用方法 'signal' 无法停止测试,但我可以看到 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。如有任何帮助或建议,我们将不胜感激。

时隔 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