如何测试LoopingCall()?

Question

在我的代码中，我使用了 task.LoopingCall() 来每秒运行一个延迟函数。我想确保这个函数能返回一些特定的正确值。所以，我想到了使用 task.clock() 并调用它的 advance() 方法。然而，我得到的响应数量并不对。

你知道我哪里出错了吗？

这里有一段测试代码来说明我的意思。首先是服务器的部分：

from twisted.internet.protocol import Factory
from twisted.protocols.basic import LineReceiver
from twisted.internet import reactor
from twisted.internet import task
import time

class Chat(LineReceiver):

    def __init__(self):
        self.echo = None

    def connectionMade(self):
        self.echo = task.LoopingCall(self.echo_print)
        self.echo.start(1)

    def connectionLost(self, reason='whatever'):
        if self.echo is not None and self.echo.running:
            self.echo.stop()

    def lineReceived(self, line):
        if line == 'stop':
            self.echo.stop()

    def echo_print (self):
        self.sendLine("Echo")

class ChatFactory(Factory):

    def __init__(self):
        pass

    def buildProtocol(self, addr):
        return Chat()

if __name__ == "__main__":
    reactor.listenTCP(8123, ChatFactory())
    reactor.run()

接下来是测试案例：

from twisted.internet import task, base
from twisted.trial import unittest
from twisted.test import proto_helpers
from chat import ChatFactory

class TestChat (unittest.TestCase):

    def setUp (self):
        self.factory = ChatFactory()
        self.clock = task.Clock()
        self.proto = self.factory.buildProtocol(('127.0.0.1', 0))
        self.tr = proto_helpers.StringTransport()
        self.proto.callLater = self.clock.callLater
        self.proto.makeConnection(self.tr)

    def tearDown (self):
        if self.proto:
            self.proto.connectionLost()

    def test_echo (self):
        self.proto.dataReceived('ook\n')
        seconds_elapsed = 5
        self.clock.advance(seconds_elapsed)
        expected = 'Echo\r\n' * seconds_elapsed
        self.assertEqual(self.tr.value(), expected)

当我在这个上运行 py.test 时，我得到：

E           FailTest: not equal:
E           a = 'Echo\r\n'
E           b = 'Echo\r\nEcho\r\nEcho\r\nEcho\r\nEcho\r\n'

注意，添加 import time; time.sleep(5) 确实让测试通过。所以，我怀疑问题出在 task.clock 的使用上不太正确。

Answer 1

我觉得我找到了问题所在。

1. LoopingCall 默认是使用反应器（reactor）。我需要设置它，让它使用我自己的时钟，这个时钟通过类变量 clock 来定义。你可以查看 task.clock 类的文档了解更多信息。
2. self.clock.advance(x) 会把时钟设置到时间 x。它并不会逐步走过 (x-1, x-2, ..., now) 这些时间，所以在这些中间步骤上应该运行的任何延迟任务都不会执行。因此，测试中的错误其实是正常的行为。在一个从 0 开始，到 seconds_elapsed 结束的循环中调用 self.clock.advance(1) 确实达到了预期的效果。

Twisted 关于 单元测试 的部分值得多读几遍，这样你能更好地理解发生了什么。如果你还有其他问题，可以看看 Twisted 内部的单元测试！