test_application.py 文件源码

def testUDP(self):
        if not interfaces.IReactorUDP(reactor, None):
            raise unittest.SkipTest, "This reactor does not support UDP sockets"
        p = protocol.DatagramProtocol()
        t = internet.TCPServer(0, p)
        t.startService()
        num = t._port.getHost().port
        def onStop(ignored):
            t = internet.TCPServer(num, p)
            t.startService()
            return t.stopService()
        return defer.maybeDeferred(t.stopService).addCallback(onStop)