В моем коде я использую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 не используется правильно. "