Я реализую протокол, который может быть создан с разными тайм-аутами, поэтому я использую TimeoutMixin. Идею можно представить с помощью этого фиктивного класса:
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)
Чтобы проверить это с помощью Twisted Trial, я сделал этот модульный тест, примерно следуя официальному руководству:
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()
Я получаю следующую ошибку при запуске теста:
$ 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)
Вот что официальная документация говорит по теме:
Вызов реактора.callLater создает IDelayedCall s. Их нужно запускать или отменять во время теста, иначе они переживут тест. Это было бы плохо, потому что они могли бы помешать более позднему тесту, вызывая запутанные сбои в несвязанных тестах! По этой причине Trial проверяет реактор, чтобы убедиться, что после теста в реакторе не осталось IDelayedCall , и не пройдет тест, если они есть. Самый чистый и простой способ убедиться, что все это работает, — это вернуть Deferred из вашего теста.
Но я не могу понять, что делать в случае TimeoutMixin.






Вы подключили свой протокол к proto_helpers.StringTransport. Во многих отношениях это хорошо и полезно для тестов. Однако один недостаток заключается в том, что StringTransport не реализует loseConnection то, что вы могли бы думать. Все, что он делает, это записывает тот факт, что метод был вызван. Он не доставляет уведомление вашему протоколу.
К счастью, есть StringTransportWithDisconnection, который доставляет уведомление протоколу. Если вы переключитесь на это, будет вызван метод connectionLost вашего протокола. Тогда вам нужно еще одно изменение. TimeoutMixin не отменяет тайм-аут автоматически при вызове connectionLost. Поэтому вам нужно добавить connectionLost в свой протокол, который вызывает self.setTimeout(None).
Большое спасибо! С небольшой дополнительной помощью из документов тест теперь проходит. После
self.transport = proto_helpers.StringTransportWithDisconnection()пришлось добавитьself.transport.protocol = self.protocol.