test_stdio.py 文件源码

def testConsumer(self):
        p = StandardIOTestProcessProtocol()
        d = p.onCompletion

        junkPath = self._junkPath()

        self._spawnProcess(p, 'stdio_test_consumer.py', junkPath)

        def processEnded(reason):
            self.assertEquals(p.data, {1: file(junkPath).read()})
            reason.trap(error.ProcessDone)
        return self._requireFailure(d, processEnded)