In meinem Code verwende ichtask.LoopingCall() um jede Sekunde eine verzögerte Funktion auszuführen. Ich möchte sicherstellen, dass diese Funktion die richtigen Werte für eine bestimmte Anzahl von Dingen zurückgibt. Also dachte ich, ich könnte einen gebrauchentask.clock() und ruf die anadvance() Methode darauf. Es werden jedoch nicht die richtigen Antworten erwartet.

Irgendeine Idee, was ich falsch mache?

Hier ist ein Testcode, um zu zeigen, was ich meine. Zunächst ist der 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()

Und jetzt der Testfall:

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)

Wenn ich darauf py.test starte, erhalte ich:

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

Beachten Sie das Hinzufügenimport time; time.sleep(5) macht in der Tat den Test bestanden. Ich vermute also, dass das Problem das isttask.clock wird nicht richtig verwendet.