test_threads.py 文件源码

def testCallMultiple(self):
        L = []
        N = 10
        d = defer.Deferred()

        def finished():
            self.assertEquals(L, range(N))
            d.callback(None)

        threads.callMultipleInThread([
            (L.append, (i,), {}) for i in xrange(N)
            ] + [(reactor.callFromThread, (finished,), {})])
        return d