def test_pushProducer(self):
        class PushProducer(object):
            implements(IPushProducer)
            resumed = False

            def __init__(self, toProduce):
                self.toProduce = toProduce

            def resumeProducing(self):
                self.resumed = True

            def start(self, consumer):
                self.consumer = consumer
                consumer.registerProducer(self, True)
                self._produceAndSchedule()

            def _produceAndSchedule(self):
                if self.toProduce:
                    self.consumer.write(self.toProduce.pop(0))
                    reactor.callLater(0, self._produceAndSchedule)
                else:
                    self.consumer.unregisterProducer()
        d = self._producertest(PushProducer)

        def finished((client, server)):
            self.failIf(
                server.producer.resumed,
                "Streaming producer should not have been resumed.")
        d.addCallback(finished)
        return d
Return this component.
dataset/ETHPy150Open kuri65536/python-for-android/python-modules/twisted/twisted/test/test_loopback.py/LoopbackAsyncTestCase.test_pushProducer--crossover with--dataset/ETHPy150Open openstack/solum/solum/api/handlers/component_handler.py/ComponentHandler.get