Skip to content
Advertisement

Twisted unittest “Reactor Unclean” when using TimeoutMixin

I’m implementing a protocol which may be instantiated with different timeouts, so I use TimeoutMixin. The idea can be represented with this dummy class:

my_protocol.py

from twisted.protocols import basic, policies
from twisted.internet import protocol

class MyProtocol(basic.LineReceiver, policies.TimeoutMixin):
    def __init__(self, timeout):
        self.setTimeout(timeout)

    def lineReceived(self, line):
        self.transport.write(line)

To test it with the Twisted Trial a made this unittest roughly following the official tutorial:

test_my_protocol.py

from twisted.trial import unittest
from twisted.test import proto_helpers

import my_protocol

class TestMyProtocol(unittest.TestCase):
    def setUp(self):
        self.protocol = my_protocol.MyProtocol(1)
        self.transport = proto_helpers.StringTransport()
        self.protocol.makeConnection(self.transport)

    def test_echo(self):
        self.protocol.lineReceived(b"test")
        self.assertEqual(self.transport.value(), b"test")

    def tearDown(self):
        self.protocol.transport.loseConnection()

I get the following error when running the test:

$ python -m twisted.trial test_my_protocol
test_my_protocol
  TestMyProtocol
    test_echo ...                                                       [ERROR]

===============================================================================
[ERROR]
Traceback (most recent call last):
Failure: twisted.trial.util.DirtyReactorAggregateError: Reactor was unclean.
DelayedCalls: (set twisted.internet.base.DelayedCall.debug = True to debug)
<DelayedCall 0x7ff1f9a8d070 [0.9994857311248779s] called=0 cancelled=0 TimeoutMixin.__timedOut()>

test_my_protocol.TestMyProtocol.test_echo
-------------------------------------------------------------------------------
Ran 1 tests in 0.011s

FAILED (errors=1)

This is what official docs says on the topic:

Calls to reactor.callLater create IDelayedCall s. These need to be run or cancelled during a test, otherwise they will outlive the test. This would be bad, because they could interfere with a later test, causing confusing failures in unrelated tests! For this reason, Trial checks the reactor to make sure there are no leftover IDelayedCall s in the reactor after a test, and will fail the test if there are. The cleanest and simplest way to make sure this all works is to return a Deferred from your test.

But I can’t figure what to do in case of TimeoutMixin.

Advertisement

Answer

You’ve hooked your protocol up to a proto_helpers.StringTransport. In many ways, this is a good thing and beneficial for the tests. However, one downside is that StringTransport doesn’t implement loseConnection how you might thing. All it does is record the fact that the method was called. It does not deliver the notification to your protocol.

Fortunately, there’s StringTransportWithDisconnection which does deliver the notification to the protocol. If you switch to this, then your protocol’s connectionLost method will be called. Then you have one other change required. TimeoutMixin does not automatically cancel its timeout when connectionLost is called. So you need to add a connectionLost to your protocol which calls self.setTimeout(None).

Advertisement