在我的代码中,我使用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()
现在测试用例:
^{pr2}$当我跑的时候py.测试关于这个,我得到:
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
没有正确使用。在
我相信我已经发现了问题。在
LoopingCall
默认使用reactor。我需要设置它,以便它通过类变量clock
使用我自己的时钟。请参阅task.clock类文档。在self.clock.advance(x)
将时钟设置为时间x
。它不经过(x-1, x-2, ..., now)
,因此任何应该在这些中间步骤上运行的延迟操作都不会运行。因此,测试中的错误是正确的行为。在从0开始到结束于seconds_elapsed
的循环中调用self.clock.advance(1)
确实取得了预期的效果。在关于unit tests的扭曲部分值得阅读几次,这样您就可以熟悉正在发生的事情。如果你有更多的问题,看看扭曲的内部单元测试!在
相关问题 更多 >
编程相关推荐