Витой юнит-тест «Reactor Unclean» при использовании TimeoutMixin

Я реализую протокол, который может быть создан с разными тайм-аутами, поэтому я использую 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.

Почему в Python есть оператор "pass"?
Почему в Python есть оператор "pass"?
Оператор pass в Python - это простая концепция, которую могут быстро освоить даже новички без опыта программирования.
Некоторые методы, о которых вы не знали, что они существуют в Python
Некоторые методы, о которых вы не знали, что они существуют в Python
Python - самый известный и самый простой в изучении язык в наши дни. Имея широкий спектр применения в области машинного обучения, Data Science,...
Основы Python Часть I
Основы Python Часть I
Вы когда-нибудь задумывались, почему в программах на Python вы видите приведенный ниже код?
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
Алиса и Боб имеют неориентированный граф из n узлов и трех типов ребер:
Оптимизация кода с помощью тернарного оператора Python
Оптимизация кода с помощью тернарного оператора Python
И последнее, что мы хотели бы показать вам, прежде чем двигаться дальше, это
Советы по эффективной веб-разработке с помощью Python
Советы по эффективной веб-разработке с помощью Python
Как веб-разработчик, Python может стать мощным инструментом для создания эффективных и масштабируемых веб-приложений.
1
0
280
1
Перейти к ответу Данный вопрос помечен как решенный

Ответы 1

Ответ принят как подходящий

Вы подключили свой протокол к proto_helpers.StringTransport. Во многих отношениях это хорошо и полезно для тестов. Однако один недостаток заключается в том, что StringTransport не реализует loseConnection то, что вы могли бы думать. Все, что он делает, это записывает тот факт, что метод был вызван. Он не доставляет уведомление вашему протоколу.

К счастью, есть StringTransportWithDisconnection, который доставляет уведомление протоколу. Если вы переключитесь на это, будет вызван метод connectionLost вашего протокола. Тогда вам нужно еще одно изменение. TimeoutMixin не отменяет тайм-аут автоматически при вызове connectionLost. Поэтому вам нужно добавить connectionLost в свой протокол, который вызывает self.setTimeout(None).

Большое спасибо! С небольшой дополнительной помощью из документов тест теперь проходит. После self.transport = proto_helpers.StringTransportWithDisconnection() пришлось добавить self.transport.protocol = self.protocol.

Simon 16.12.2020 13:57

Другие вопросы по теме