test_process.py 文件源码

def test_closeHandles(self):
        """
        The win32 handles should be properly closed when the process exits.
        """
        import win32api

        connected = defer.Deferred()
        ended = defer.Deferred()

        class SimpleProtocol(protocol.ProcessProtocol):
            """
            A protocol that fires deferreds when connected and disconnected.
            """
            def makeConnection(self, transport):
                connected.callback(transport)

            def processEnded(self, reason):
                ended.callback(None)

        p = SimpleProtocol()
        pyArgs = [pyExe, b"-u", b"-c", b"print('hello')"]
        proc = reactor.spawnProcess(p, pyExe, pyArgs)

        def cbConnected(transport):
            self.assertIs(transport, proc)
            # perform a basic validity test on the handles
            win32api.GetHandleInformation(proc.hProcess)
            win32api.GetHandleInformation(proc.hThread)
            # And save their values for later
            self.hProcess = proc.hProcess
            self.hThread = proc.hThread
        connected.addCallback(cbConnected)

        def checkTerminated(ignored):
            # The attributes on the process object must be reset...
            self.assertIsNone(proc.pid)
            self.assertIsNone(proc.hProcess)
            self.assertIsNone(proc.hThread)
            # ...and the handles must be closed.
            self.assertRaises(win32api.error,
                              win32api.GetHandleInformation, self.hProcess)
            self.assertRaises(win32api.error,
                              win32api.GetHandleInformation, self.hThread)
        ended.addCallback(checkTerminated)

        return defer.gatherResults([connected, ended])