test_udp.py 文件源码

def testStartStop(self):
        client = Client()
        d = client.startedDeferred = defer.Deferred()
        port2 = reactor.connectUDP("127.0.0.1", 8888, client)

        def assertName():
            self.failUnless(repr(port2).find('test_udp.Client') >= 0)

        def cbStarted(ignored):
            self.assertEquals(client.started, 1)
            self.assertEquals(client.stopped, 0)
            assertName()
            d = defer.maybeDeferred(port2.stopListening)
            d.addCallback(lambda ign: assertName())
            return d

        return d.addCallback(cbStarted)