test_worker.py 文件源码

def test_processEnded(self):
        """
        L{LocalWorker.processEnded} calls C{connectionLost} on itself and on
        the L{AMP} protocol.
        """

        class FakeStream(object):
            callNumber = 0

            def close(self):
                self.callNumber += 1


        transport = FakeTransport()
        protocol = FakeAMProtocol()
        localWorker = LocalWorker(protocol, '.', 'test.log')
        localWorker.makeConnection(transport)
        localWorker._outLog = FakeStream()
        localWorker.processEnded(Failure(CONNECTION_DONE))
        self.assertEqual(localWorker._outLog.callNumber, 1)
        self.assertIdentical(None, protocol.transport)
        return self.assertFailure(localWorker.endDeferred, ConnectionDone)