使用 TimeoutMixin 时扭曲单元测试 "Reactor Unclean"

Question

我正在实现一个可以用不同的超时实例化的协议，所以我使用 TimeoutMixin。这个想法可以用这个虚拟 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)

为了用 Twisted Trial 测试它，使这个单元测试大致遵循 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()

我在运行测试时得到以下错误：

$ 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)

这是 official docs 对主题的看法：

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.

但我不知道在 TimeoutMixin 的情况下该怎么做。

Answer 1

您已将协议连接到 proto_helpers.StringTransport。在许多方面，这是一件好事，对测试有益。但是，一个缺点是 StringTransport 没有实现 loseConnection 您可能的想法。它所做的只是记录调用该方法的事实。它不会向您的协议发送通知。

幸运的是，StringTransportWithDisconnection 将通知传递给协议。如果你切换到这个，那么你的协议的 connectionLost 方法将被调用。然后您需要进行另一项更改。 TimeoutMixin 不会在调用 connectionLost 时自动取消其超时。所以你需要在调用 self.setTimeout(None).

的协议中添加一个 connectionLost

使用 TimeoutMixin 时扭曲单元测试 "Reactor Unclean"

Twisted unittest "Reactor Unclean" when using TimeoutMixin

python

unit-testing

twisted

deferred

twisted.internet