def test_protocolSwitch(self, switcher=SimpleSymmetricCommandProtocol,
                            spuriousTraffic=False,
                            spuriousError=False):
        """
        Verify that it is possible to switch to another protocol mid-connection and
        send data to it successfully.
        """
        self.testSucceeded = False

        serverDeferred = defer.Deferred()
        serverProto = switcher(serverDeferred)
        clientDeferred = defer.Deferred()
        clientProto = switcher(clientDeferred)
        c, s, p = connectedServerAndClient(ServerClass=lambda: serverProto,
                                           ClientClass=lambda: clientProto)

        if spuriousTraffic:
            wfdr = []           # remote
            c.callRemote(WaitForever).addErrback(wfdr.append)
        switchDeferred = c.switchToTestProtocol()
        if spuriousTraffic:
            self.assertRaises(amp.ProtocolSwitched, c.sendHello, b'world')

        def cbConnsLost(info):
            ((serverSuccess, serverData), (clientSuccess, clientData)) = info
            self.assertTrue(serverSuccess)
            self.assertTrue(clientSuccess)
            self.assertEqual(b''.join(serverData), SWITCH_CLIENT_DATA)
            self.assertEqual(b''.join(clientData), SWITCH_SERVER_DATA)
            self.testSucceeded = True

        def cbSwitch(proto):
            return defer.DeferredList(
                [serverDeferred, clientDeferred]).addCallback(cbConnsLost)

        switchDeferred.addCallback(cbSwitch)
        p.flush()
        if serverProto.maybeLater is not None:
            serverProto.maybeLater.callback(serverProto.maybeLaterProto)
            p.flush()
        if spuriousTraffic:
            # switch is done here; do this here to make sure that if we're
            # going to corrupt the connection, we do it before it's closed.
            if spuriousError:
                s.waiting.errback(amp.RemoteAmpError(
                        b"SPURIOUS",
                        "Here's some traffic in the form of an error."))
            else:
                s.waiting.callback({})
            p.flush()
        c.transport.loseConnection() # close it
        p.flush()
        self.assertTrue(self.testSucceeded)

        Verify that it is possible to switch to another protocol mid-connection and
        send data to it successfully.
        
stackoverflow/venv/lib/python3.6/site-packages/twisted/test/test_amp.py