python - 使用 TimeoutMixin 时扭曲的单元测试“Reactor Unclean”
问题描述
我正在实现一个可以用不同的超时实例化的协议,所以我使用TimeoutMixin
. 这个想法可以用这个虚拟类来表示:
我的协议.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)
这是官方文档关于该主题的说法:
对 reactor.callLater 的调用会创建 IDelayedCall 。这些需要在测试期间运行或取消,否则它们将超过测试。这会很糟糕,因为它们可能会干扰以后的测试,从而在不相关的测试中造成混乱的失败!为此,Trial 会检查 reactor 以确保在测试后,reactor 中没有剩余的 IDelayedCall ,如果有则测试失败。确保这一切正常工作的最简洁和最简单的方法是从测试中返回 Deferred。
但是我不知道在 TimeoutMixin 的情况下该怎么做。
解决方案
您已将协议连接到proto_helpers.StringTransport
. 在许多方面,这是一件好事,对测试有益。然而,一个缺点是它StringTransport
没有实现loseConnection
你可能的事情。它所做的只是记录调用该方法的事实。它不会将通知传递给您的协议。
幸运的是,确实有哪个StringTransportWithDisconnection
将通知传递给协议。如果你切换到这个,那么你的协议的方法将被调用。然后,您需要进行另一项更改。 调用时不会自动取消其超时。所以你需要在你的协议中添加一个调用.connectionLost
TimeoutMixin
connectionLost
connectionLost
self.setTimeout(None)
推荐阅读
- project - MS Project - 将资源分配给同一天发生的多个任务
- azure - 从 Presto DB 连接 Azure 数据湖存储:方案没有文件系统:adl
- php - 如何为 LDAP 配置 SSL 证书
- r - 创建一个新变量并将值分配给一个组
- python - 如何插入旋转网格?
- css - CSS 选择器,对同一个类使用多个
- php - 在 WordPress 中跨多个分类法搜索和过滤查询
- ruby-on-rails - Rails 控制台中的 Rails 5 ActsAsTaggable 批量更新标签?
- authentication - 如何生成启用扩展 TLS Web 客户端身份验证的 x509 客户端证书
- javascript - 在所有列上使用 JavaScript 过滤带有文本字段的表