1

这里是

  • 有类Body将消息发送到远程 Perspective Broker (PB) Friend

  • 测试是检查Me实例是否可以 通过调用 *speak_ to _friend* 或tell方法与Friend通信。

  • speak _ to _friend是Soul.speak虚函数的封装,被Body.tell方法覆盖。 Soul类被封装到Body中

  • tell方法通过等待接收到的回调函数中的gotAnswer事件来模拟阻塞调用。

  • 根据打印输出,对于单元测试和独立案例,Friend以相同的方式与我交流。

问题

...是在试用运行单元测试( unittest-me.py )时,接收到的回调永远不会被调用。而不是在客户端打印的愚蠢错误消息:

连接被干净地关闭

对我来说,在延迟队列完全处理之前,reactor 似乎已关闭。

如果我正在运行独立测试test-body.py,那么一切都会按预期进行:朋友与我交谈!

问题

  1. 为什么在这种情况下 unittest 的行为会有所不同?
  2. 如何解决?

单元测试输出

Python2.7\Scripts\trial.py ..\Soapbox\Node\unittest-me.py
unittest-me
  Trial_TestCase
    test_conversation ... wake upConnecting to localhost:18800

DBG: speak
DBG: say
                                               [FAIL]
Friend not willing to talk with me because:  [Failure instance: Traceback (failure with no frames): <class 'twisted.internet.error.ConnectionDone
'>: Connection was closed cleanly.
]

===============================================================================
[FAIL]
Traceback (most recent call last):
  File "..\Soapbox\Node\unittest-me.py", line 50, in test_conversation
    self.assertTrue(answer is not None, "Friend gave no answer")
twisted.trial.unittest.FailTest: Friend gave no answer

unittest-me.Trial_TestCase.test_conversation
-------------------------------------------------------------------------------
Ran 1 tests in 3.096s

FAILED (failures=1)

测试体输出

python test-body.py
connector:  <twisted.internet.tcp.Connector instance at 0x01C87940>
Press any key to continue...Starting reactor


wake upDBG: speak
DBG: say

DBG: received
Friend's reply: <username>
Press any key to continue...

That's all

朋友的输出

Dr. Livesey: I got your words: whoami
DBG: answer:  <username>

Dr. Livesey: I got your words: whoami
DBG: answer:  <username>

单元测试-me.py

from twisted.trial import unittest
from twisted.spread import pb
from twisted.internet import reactor
from body import Body

class Trial_TestCase(unittest.TestCase):

    def setUp(self):
        self.ref = None
        self.Me = Body('localhost', 18800)
        self.assertTrue(self.Me is not None, "Creating Me assertion failed")
        self.Me.start()

        self.Me.pbFactory = pb.PBClientFactory(self.Me)

        print "Connecting to " + self.Me.hostName + ":" + str(self.Me.portNo)
        connector = reactor.connectTCP(self.Me.hostName,
                                       self.Me.portNo,
                                       self.Me.pbFactory)
        self.addCleanup(connector.disconnect)

        # =====================================================================
        # unittest related parts
        # =====================================================================
        def gotRoot(ref):
            self.ref = ref

        df = self.Me.pbFactory.getRootObject()
        '''
            Without this dummy call, I can say no word - say not called
        '''
        obj = df.addCallback(gotRoot)
        return obj


    def tearDown(self):
        self.Me.remoteObj.broker.transport.loseConnection()
        self.Me.pbFactory.disconnect()
        pass


    def test_conversation(self):
        # answer = self.Me.speak_to_friend()
        answer = self.Me.tell(None)
        self.assertTrue(answer is not None, "Friend gave no answer")

        return self.Me.dm

if __name__ == "__main__":
    #import sys;sys.argv = ['', 'Test.testName']
    unittest.main()

测试body.py

if __name__ == '__main__':
    pass

from twisted.spread import pb
from twisted.internet import reactor, threads
import threading
import time
from body import Body

def controlThread(name, body):
    print "Press any key to continue..."
    raw_input()

    body.start()

    result = body.tell(None)

    print "Friend's reply: " + str(result)

    print "Press any key to continue..."
    raw_input()
    reactor.stop()
    pass


body = Body('127.0.0.1', 18800)
f = pb.PBClientFactory(body)
body.pbFactory = f

connector = reactor.connectTCP("127.0.0.1", 18800, f)
print "connector: ", connector

threading.Thread(target=controlThread, args=("Control", body)).start()
print "Starting reactor"
reactor.run()
print "That's all"

正文.py

import threading
import types

class Soul():
    def speak(self):
        '''
            virtual method must be overwritten by derived class
        '''
        raise NotImplementedError("I cannot speak without a body. "\
                                  "Please give me body first")
        pass

class Body(threading.Thread):
    def __init__(self, hostname, portno):
        self.hostName = hostname
        self.portNo   = portno

        self.df = None
        self.dg = None
        self.dm = None
        self.remoteObj = None
        self.pbFactory = None

        self.answer = None
        self.gotAnswer = threading.Event()

        self.Soul = Soul()
        self.Soul.speak = types.MethodType(self.tell, self.Soul)

        threading.Thread.__init__(self)
        pass

    def run(self):
        print "wake up"
        pass

    def speak_to_friend(self):
        print 'DBG: speak_to_friend'
        return self.Soul.speak()

    def tell(self, soul):
        print 'DBG: speak'
        self.df = self.pbFactory.getRootObject()

        self.answer = None
        self.gotAnswer.clear()
        self.dg = self.df.addCallback(self.say)
        self.gotAnswer.wait(timeout=3)
        return self.answer

    def say(self, remoteObj):
        print 'DBG: say'
        self.remoteObj = remoteObj
        self.dm = remoteObj.callRemote('talk', 'whoami')
        self.dm.addCallback(self.received)
        self.dm.addErrback(self.broken)
        pass

    def received(self, (answer, result)):
        print 'DBG: received'
        self.answer = '\n\t'.join(str.splitlines(answer))
        self.gotAnswer.set()
        pass

    def broken(self,reason):
        print "Friend not willing to talk with me because: ", reason.getErrorMessage()
        pass

朋友.py

from twisted.spread import pb
from twisted.internet import reactor
from twisted.internet import defer

from subprocess import Popen
import subprocess

from twisted.web import _newclient


class BrokenError(pb.Error):    pass

class Friend(pb.Root):
    def __init__(self,name):
        self.counter = 0
        self.name = str(name)

    def remote_broken(self):
        msg = "I don't hear you"
        print "Talk louder" % msg
        raise BrokenError(msg)

    def broken(self,reason):
        print "Communication is broken: ", reason.getErrorMessage()
        # print "printError > %r" % failure
        if reason.check(_newclient.RequestGenerationFailed):
            print "printError: RequestGenerationFailed"
            for f in reason.value.reasons:
                print "printError > %r" % f
                print f.getTraceback()
        pass


    def remote_talk(self,words):
        print self.name + ": I got your words: " + words

        process = Popen(words,stdout=subprocess.PIPE)
        answer = process.stdout.read()
        process.wait()

        print "DBG: answer: ", answer

        return (answer, process.returncode)

def main():
    reactor.listenTCP(18800, pb.PBServerFactory(Friend("Dr. Livesey")))
    reactor.run()

if __name__ == '__main__':
    main()
4

0 回答 0