def test_pause(self):
        """
        When the streaming producer is paused, the underlying producer stops
        getting resumeProducing calls.
        """
        class PausingStringTransport(StringTransport):
            writes = 0

            def __init__(self):
                StringTransport.__init__(self)
                self.paused = Deferred()

            def write(self, data):
                self.writes += 1
                StringTransport.write(self, data)
                if self.writes == 3:
                    self.producer.pauseProducing()
                    d = self.paused
                    del self.paused
                    d.callback(None)


        consumer = PausingStringTransport()
        nsProducer = NonStreamingProducer(consumer)
        streamingProducer = _PullToPush(nsProducer, consumer)
        consumer.registerProducer(streamingProducer, True)

        # Make sure the consumer does not continue:
        def shouldNotBeCalled(ignore):
            self.fail("BUG: The producer should not finish!")
        nsProducer.result.addCallback(shouldNotBeCalled)

        done = consumer.paused
        def paused(ignore):
            # The CooperatorTask driving the producer was paused:
            self.assertEqual(streamingProducer._coopTask._pauseCount, 1)
        done.addCallback(paused)

        # Now, start streaming:
        streamingProducer.startStreaming()
        return done

        When the streaming producer is paused, the underlying producer stops
        getting resumeProducing calls.
        
stackoverflow/venv/lib/python3.6/site-packages/twisted/protocols/test/test_tls.py