test_process.py 文件源码

def test_mockForkErrorGivenFDs(self):
        """
        When C{os.forks} raises an exception and that file descriptors have
        been specified with the C{childFDs} arguments of
        L{reactor.spawnProcess}, they are not closed.
        """
        self.mockos.raiseFork = OSError(errno.EAGAIN, None)
        protocol = TrivialProcessProtocol(None)
        self.assertRaises(OSError, reactor.spawnProcess, protocol, None,
            childFDs={0: -10, 1: -11, 2: -13})
        self.assertEqual(self.mockos.actions, [("fork", False)])
        self.assertEqual(self.mockos.closed, [])

        # We can also put "r" or "w" to let twisted create the pipes
        self.assertRaises(OSError, reactor.spawnProcess, protocol, None,
            childFDs={0: "r", 1: -11, 2: -13})
        self.assertEqual(set(self.mockos.closed), set([-1, -2]))