test_protocol.py 文件源码

def test_beginRequest_finishedNotifier_forwards_failures(self):
        '''Beginning a request retrieves a Deferred from that request that
        forwards failures to the protocol's connectionLost.

        '''
        self.protocol.request = self.request
        self.protocol.beginRequest()

        reason = connectionDone

        def assertConnectionLostCalled(ignored):
            recordedExceptions = [
                reason.value for reason in
                self.sessionMachineRecorder.connectionsLostReasons]
            self.assertEqual(recordedExceptions, [reason.value])

        finishedNotifier = self.protocol.finishedNotifier
        finishedNotifier.addCallback(assertConnectionLostCalled)

        def trapConnectionDone(failure):
            failure.trap(error.ConnectionDone)

        terminationDeferred = self.protocol.terminationDeferred
        terminationDeferred.addErrback(trapConnectionDone)

        self.request.processingFailed(reason)
        return DeferredList([finishedNotifier, terminationDeferred])