def loseConnectionWithProducer(self, writeBlockedOnRead):
        """
        Common code for tests involving writes by producer after
        loseConnection is called.
        """
        clientProtocol, tlsProtocol, producer = self.setupStreamingProducer()
        serverProtocol, serverTLSProtocol = buildTLSProtocol(server=True)

        if not writeBlockedOnRead:
            # Do the initial handshake before write:
            self.flushTwoTLSProtocols(tlsProtocol, serverTLSProtocol)
        else:
            # In this case the write below will trigger write-blocked-on-read
            # condition...
            pass

        # Now write, then lose connection:
        clientProtocol.transport.write(b"x ")
        clientProtocol.transport.loseConnection()
        self.flushTwoTLSProtocols(tlsProtocol, serverTLSProtocol)

        # Underlying transport should not have loseConnection called yet, nor
        # should producer be stopped:
        self.assertFalse(tlsProtocol.transport.disconnecting)
        self.assertFalse("stop" in producer.producerHistory)

        # Writes from client to server should continue to go through, since we
        # haven't unregistered producer yet:
        clientProtocol.transport.write(b"hello")
        clientProtocol.transport.writeSequence([b" ", b"world"])

        # Unregister producer; this should trigger TLS shutdown:
        clientProtocol.transport.unregisterProducer()
        self.assertNotEqual(tlsProtocol.transport.value(), b"")
        self.assertFalse(tlsProtocol.transport.disconnecting)

        # Additional writes should not go through:
        clientProtocol.transport.write(b"won't")
        clientProtocol.transport.writeSequence([b"won't!"])

        # Finish TLS close handshake:
        self.flushTwoTLSProtocols(tlsProtocol, serverTLSProtocol)
        self.assertTrue(tlsProtocol.transport.disconnecting)

        # Bytes made it through, as long as they were written before producer
        # was unregistered:
        self.assertEqual(b"".join(serverProtocol.received), b"x hello world")

        Common code for tests involving writes by producer after
        loseConnection is called.
        
stackoverflow/venv/lib/python3.6/site-packages/twisted/protocols/test/test_tls.py