3

私のコードでは、task.LoopingCall()遅延関数を毎秒実行していました。その関数が特定の数に対して正しい値を返すことを確認したいと思います。だから、私は a を使ってメソッド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正しく使用されていないことにあると思われます。

4

1 に答える 1

2

私は問題を発見したと信じています。

  1. LoopingCallデフォルトでリアクターを使用しています。クラス変数を介して独自の時計を使用するように設定する必要がありましたclock。task.clockクラスのドキュメントを参照してください。
  2. self.clock.advance(x)時計を time に設定しますx。実行されない(x-1, x-2, ..., now)ため、これらの中間ステップで実行する必要がある遅延は実行されません。したがって、テストのエラーは正しい動作です。self.clock.advance(1)0 から始まり で終わるループ内で呼び出すとseconds_elapsed、望ましい効果が得られました。

単体テストの Twisted セクションは何回か読む価値があるので、何が起こっているのかを理解できます。さらに問題がある場合は、twisted internal unit tests を見てください。

于 2013-01-31T13:48:06.083 に答える