test_session.py 文件源码

def test_wrapProcessProtocol_Protocol(self):
        """
        L{wrapPRocessProtocol}, when passed a L{Protocol} should return
        something that follows the L{IProcessProtocol} interface, with
        connectionMade() mapping to connectionMade(), outReceived() mapping to
        dataReceived() and processEnded() mapping to connectionLost().
        """
        protocol = MockProtocol()
        protocol.transport = StubTransport()
        process_protocol = session.wrapProcessProtocol(protocol)
        process_protocol.connectionMade()
        process_protocol.outReceived(b'data')
        self.assertEqual(protocol.transport.buf, b'data~')
        process_protocol.processEnded(failure.Failure(
            error.ProcessTerminated(0, None, None)))
        protocol.reason.trap(error.ProcessTerminated)