如何测试LoopingCall()? [英] How to test LoopingCall()?

问题描述

在我的代码中，我使用task.LoopingCall()每秒运行一些延迟函数.我想确保该函数为某些事物返回正确的值.因此，我以为可以使用task.clock()并在其上调用advance()方法.但是，我没有得到正确数量的预期答复.

Within my code, I use task.LoopingCall() to run some deferred function every second. I want to make sure that that function returns the right values for a certain number of things. So, I thought I could use a task.clock() and call the advance() method on it. However, I am not getting the right number of responses expected.

知道我在做什么错吗?

这是一个测试代码，可以显示我的意思.首先是服务器:

Here is a test code to show what I mean. First is the server:

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时，我得到:

When I run py.test on this, I get:

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的使用不正确.

Note that adding import time; time.sleep(5) does indeed make the test pass. So, I suspect that the problem is that the task.clock is not used correctly.

推荐答案

我相信我已经找到了问题.

I believe that I have found the problems.

1. LoopingCall默认情况下使用反应堆.我需要进行设置，以使其通过类变量clock使用自己的时钟.请参见 task.clock 类文档.
2. self.clock.advance(x)将时钟设置为时间x.它不会通过(x-1, x-2, ..., now)，因此应该在这些中间步骤上运行的所有延期操作都不会运行.因此，测试中的错误是正确的行为.在从0开始到seconds_elapsed结束的循环中调用self.clock.advance(1)确实达到了预期的效果.

1. LoopingCall is using the reactor by default. I needed to set it up so that it used my own clock via the class variable clock. See task.clock class documentation.
2. self.clock.advance(x) sets the clock to be at time x. It does not go through (x-1, x-2, ..., now) and thus any deferred that should run on those intermediate steps will not run. Hence, the error in the test is correct behaviour. Calling self.clock.advance(1) within a loop starting at 0 and ending in seconds_elapsed did have the desired effect.

单元测试上的扭曲"部分值得阅读几次，以便您熟悉正在发生的事情.如果您还有其他问题，请查看扭曲的内部单元测试！

The Twisted section on unit tests is worth reading a few times so you get familiar with what is going on. If you have more problems, look at the twisted internal unit tests!

这篇关于如何测试LoopingCall()?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！