如何使使用 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 "/usr/lib64/python2.7/threading.py", line 784, in __bootstrap
self.__bootstrap_inner()
File "/usr/lib64/python2.7/threading.py", line 811, in __bootstrap_inner
self.run()
File "/usr/lib64/python2.7/threading.py", line 764, in run
self.__target(*self.__args, **self.__kwargs)
File "/usr/lib64/python2.7/site-packages/twisted/python/threadpool.py", line 190, in _worker
o = self.q.get()
File "/usr/lib64/python2.7/Queue.py", line 168, in get
self.not_empty.wait()
File "/usr/lib64/python2.7/threading.py", line 339, in wait
waiter.acquire()
~~~~~~~ Stack of PoolThread-twisted.internet.reactor-1 (139997702035200) ~~~~~~~
File "/usr/lib64/python2.7/threading.py", line 784, in __bootstrap
self.__bootstrap_inner()
File "/usr/lib64/python2.7/threading.py", line 811, in __bootstrap_inner
self.run()
File "/usr/lib64/python2.7/threading.py", line 764, in run
self.__target(*self.__args, **self.__kwargs)
File "/usr/lib64/python2.7/site-packages/twisted/python/threadpool.py", line 190, in _worker
o = self.q.get()
File "/usr/lib64/python2.7/Queue.py", line 168, in get
self.not_empty.wait()
File "/usr/lib64/python2.7/threading.py", line 339, in wait
waiter.acquire()
+++++++++++++++++++++++++++++++++++ Timeout ++++++++++++++++++++++++++++++++++++
Unhandled Error
Traceback (most recent call last):
File "/usr/lib64/python2.7/site-
packages/twisted/internet/base.py", line 1169, in run
self.mainLoop()
--- <exception caught here> ---
File "/usr/lib64/python2.7/site-
packages/twisted/internet/base.py", line 1181, in mainLoop
self.doIteration(t)
File "/usr/lib64/python2.7/site-
packages/twisted/internet/epollreactor.py", line 362, in doPoll
l = self._poller.poll(timeout, len(self._selectables))
File "/usr/lib/python2.7/site-packages/pytest_timeout.py", line 110, in handler
timeout_sigalrm(item, timeout)
File "/usr/lib/python2.7/site-packages/pytest_timeout.py", line 243, in timeout_sigalrm
pytest.fail('Timeout >%ss' % timeout)
File "/usr/lib/python2.7/site-packages/_pytest/outcomes.py", line 85, in fail
raise Failed(msg=msg, pytrace=pytrace)
builtins.Failed: Timeout >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
我正在努力将一些测试从使用 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 "/usr/lib64/python2.7/threading.py", line 784, in __bootstrap
self.__bootstrap_inner()
File "/usr/lib64/python2.7/threading.py", line 811, in __bootstrap_inner
self.run()
File "/usr/lib64/python2.7/threading.py", line 764, in run
self.__target(*self.__args, **self.__kwargs)
File "/usr/lib64/python2.7/site-packages/twisted/python/threadpool.py", line 190, in _worker
o = self.q.get()
File "/usr/lib64/python2.7/Queue.py", line 168, in get
self.not_empty.wait()
File "/usr/lib64/python2.7/threading.py", line 339, in wait
waiter.acquire()
~~~~~~~ Stack of PoolThread-twisted.internet.reactor-1 (139997702035200) ~~~~~~~
File "/usr/lib64/python2.7/threading.py", line 784, in __bootstrap
self.__bootstrap_inner()
File "/usr/lib64/python2.7/threading.py", line 811, in __bootstrap_inner
self.run()
File "/usr/lib64/python2.7/threading.py", line 764, in run
self.__target(*self.__args, **self.__kwargs)
File "/usr/lib64/python2.7/site-packages/twisted/python/threadpool.py", line 190, in _worker
o = self.q.get()
File "/usr/lib64/python2.7/Queue.py", line 168, in get
self.not_empty.wait()
File "/usr/lib64/python2.7/threading.py", line 339, in wait
waiter.acquire()
+++++++++++++++++++++++++++++++++++ Timeout ++++++++++++++++++++++++++++++++++++
Unhandled Error
Traceback (most recent call last):
File "/usr/lib64/python2.7/site-
packages/twisted/internet/base.py", line 1169, in run
self.mainLoop()
--- <exception caught here> ---
File "/usr/lib64/python2.7/site-
packages/twisted/internet/base.py", line 1181, in mainLoop
self.doIteration(t)
File "/usr/lib64/python2.7/site-
packages/twisted/internet/epollreactor.py", line 362, in doPoll
l = self._poller.poll(timeout, len(self._selectables))
File "/usr/lib/python2.7/site-packages/pytest_timeout.py", line 110, in handler
timeout_sigalrm(item, timeout)
File "/usr/lib/python2.7/site-packages/pytest_timeout.py", line 243, in timeout_sigalrm
pytest.fail('Timeout >%ss' % timeout)
File "/usr/lib/python2.7/site-packages/_pytest/outcomes.py", line 85, in fail
raise Failed(msg=msg, pytrace=pytrace)
builtins.Failed: Timeout >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