我正在实现一个协议,该协议可以用不同的超时来实例化,所以我使用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)
为了使用扭曲试验对其进行测试,a大致按照official tutorial进行了单元测试:
测试我的协议.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.
但我不知道如果超时混合该怎么办
您已将协议连接到
proto_helpers.StringTransport
。在许多方面,这是一件好事,对测试是有益的。然而,一个缺点是StringTransport
没有实现loseConnection
您可能会做的事情。它所做的只是记录调用该方法的事实。它不会将通知传递到您的协议幸运的是,有
StringTransportWithDisconnection
,将通知传递给协议。如果切换到该选项,则将调用协议的connectionLost
方法。那么您还需要另一个更改TimeoutMixin
在调用connectionLost
时不会自动取消其超时。因此,您需要将connectionLost
添加到调用self.setTimeout(None)
的协议中相关问题 更多 >
编程相关推荐