timeoutautomated-teststwistedpytestnose

How do you timeout a twisted test that uses pytest?

I'm working on converting some tests from using Nose and twisted, to using Pytest and twisted, as Nose is no longer in development. The easiest way to convert the tests is by editing the custom decorator that each test has. This decorator is on every test, and defines a timeout for the individual test.

I've tried using @pytest.mark.timeout, but the only method that's worked is the 'thread' method, but this stops the entire test run and won't continue on to the next test. Using the method 'signal' fails to stop the test, but I can see an error present in the junitxml file. 

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

The tests themselves use twisted to start up and send messages to the software. I don't need the tests to cancel any twisted processes or locks. I would just like pytest to mark the test as a failure after the timeout, and then move onto the next test.

Below is the error I see in the xml file when using signal method of timeout. 

</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>

I have looked around for a similar solution, and the closest I could find was this question. Any help or suggestions would be appreciated.

Solution

  • Coming back to this after 4+ years with an answer. The problem seems to be the exception from the test getting caught by the twisted reactor. I was able to resolve this by updating the version of twisted. Twisted versions since 16.5 have a new Deferred function call addTimeout (Docs). Using that, I was able to modify the original decorator to the following. Now whenever a test times out, it simply raises an exception and moves on to the next one. May not be the most elegant, but I hope this helps someone else out!

        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