def test_streamingProducerBothTransportsDecideToPause(self):
        """
        pauseProducing() events can come from both the TLS transport layer and
        the underlying transport. In this case, both decide to pause,
        underlying first.
        """
        class PausingStringTransport(StringTransport):
            _didPause = False

            def write(self, data):
                if not self._didPause and self.producer is not None:
                    self._didPause = True
                    self.producer.pauseProducing()
                StringTransport.write(self, data)


        class TLSConnection(object):
            def __init__(self):
                self.l = []

            def send(self, data):
                # on first write, don't send all bytes:
                if not self.l:
                    data = data[:-1]
                # pause on second write:
                if len(self.l) == 1:
                    self.l.append("paused")
                    raise WantReadError()
                # otherwise just take in data:
                self.l.append(data)
                return len(data)

            def set_connect_state(self):
                pass

            def do_handshake(self):
                pass

            def bio_write(self, data):
                pass

            def bio_read(self, size):
                return b'X'

            def recv(self, size):
                raise WantReadError()

        transport = PausingStringTransport()
        clientProtocol, tlsProtocol, producer = self.setupStreamingProducer(
            transport, fakeConnection=TLSConnection())
        self.assertEqual(producer.producerState, 'producing')

        # Shove in fake TLSConnection that will raise WantReadError the second
        # time send() is called. This will allow us to have bytes written to
        # to the PausingStringTransport, so it will pause the producer. Then,
        # WantReadError will be thrown, triggering the TLS transport's
        # producer code path.
        clientProtocol.transport.write(b"hello")
        self.assertEqual(producer.producerState, 'paused')
        self.assertEqual(producer.producerHistory, ['pause'])

        # Now, underlying transport resumes, and then we deliver some data to
        # TLS transport so that it will resume:
        tlsProtocol.transport.producer.resumeProducing()
        self.assertEqual(producer.producerState, 'producing')
        self.assertEqual(producer.producerHistory, ['pause', 'resume'])
        tlsProtocol.dataReceived(b"hello")
        self.assertEqual(producer.producerState, 'producing')
        self.assertEqual(producer.producerHistory, ['pause', 'resume'])

        pauseProducing() events can come from both the TLS transport layer and
        the underlying transport. In this case, both decide to pause,
        underlying first.
        
stackoverflow/venv/lib/python3.6/site-packages/twisted/protocols/test/test_tls.py