def test_useReactorArgument(self):
        """
        L{twcgi.FilteredScript.runProcess} uses the reactor passed as an
        argument to the constructor.
        """
        class FakeReactor:
            """
            A fake reactor recording whether spawnProcess is called.
            """
            called = False
            def spawnProcess(self, *args, **kwargs):
                """
                Set the C{called} flag to C{True} if C{spawnProcess} is called.

                @param args: Positional arguments.
                @param kwargs: Keyword arguments.
                """
                self.called = True

        fakeReactor = FakeReactor()
        request = DummyRequest(['a', 'b'])
        resource = twcgi.FilteredScript("dummy-file", reactor=fakeReactor)
        _render(resource, request)

        self.assertTrue(fakeReactor.called) 

def launchWorkerProcesses(self, spawner, protocols, arguments):
        """
        Spawn processes from a list of process protocols.

        @param spawner: A C{IReactorProcess.spawnProcess} implementation.

        @param protocols: An iterable of C{ProcessProtocol} instances.

        @param arguments: Extra arguments passed to the processes.
        """
        workertrialPath = theSystemPath[
            'twisted.trial._dist.workertrial'].filePath.path
        childFDs = {0: 'w', 1: 'r', 2: 'r', _WORKER_AMP_STDIN: 'w',
                    _WORKER_AMP_STDOUT: 'r'}
        environ = os.environ.copy()
        # Add an environment variable containing the raw sys.path, to be used by
        # subprocesses to make sure it's identical to the parent. See
        # workertrial._setupPath.
        environ['TRIAL_PYTHONPATH'] = os.pathsep.join(sys.path)
        for worker in protocols:
            args = [sys.executable, workertrialPath]
            args.extend(arguments)
            spawner(worker, sys.executable, args=args, childFDs=childFDs,
                    env=environ) 

def _spawn(script, outputFD):
    """
    Start a script that is a peer of this test as a subprocess.

    @param script: the module name of the script in this directory (no
        package prefix, no '.py')
    @type script: C{str}

    @rtype: L{StartStopProcessProtocol}
    """
    pyExe = FilePath(sys.executable).asBytesMode().path
    env = bytesEnviron()
    env[b"PYTHONPATH"] = FilePath(
        pathsep.join(sys.path)).asBytesMode().path
    sspp = StartStopProcessProtocol()
    reactor.spawnProcess(
        sspp, pyExe, [
            pyExe,
            FilePath(__file__).sibling(script + ".py").asBytesMode().path,
            intToBytes(outputFD),
        ],
        env=env,
        childFDs={0: "w", 1: "r", 2: "r", outputFD: outputFD}
    )
    return sspp 

def test_stdio(self):
        """
        L{twisted.internet.stdio} test.
        """
        scriptPath = b"twisted.test.process_twisted"
        p = Accumulator()
        d = p.endedDeferred = defer.Deferred()
        reactor.spawnProcess(p, pyExe, [pyExe, b'-u', b"-m", scriptPath],
                             env=properEnv,
                             path=None, usePTY=self.usePTY)
        p.transport.write(b"hello, world")
        p.transport.write(b"abc")
        p.transport.write(b"123")
        p.transport.closeStdin()

        def processEnded(ign):
            self.assertEqual(p.outF.getvalue(), b"hello, worldabc123",
                              "Output follows:\n"
                              "%s\n"
                              "Error message from process_twisted follows:\n"
                              "%s\n" % (p.outF.getvalue(), p.errF.getvalue()))
        return d.addCallback(processEnded) 

def test_unsetPid(self):
        """
        Test if pid is None/non-None before/after process termination.  This
        reuses process_echoer.py to get a process that blocks on stdin.
        """
        finished = defer.Deferred()
        p = TrivialProcessProtocol(finished)
        scriptPath = b"twisted.test.process_echoer"
        procTrans = reactor.spawnProcess(p, pyExe,
                                         [pyExe, b'-u', b"-m", scriptPath],
                                         env=properEnv)
        self.assertTrue(procTrans.pid)

        def afterProcessEnd(ignored):
            self.assertIsNone(procTrans.pid)

        p.transport.closeStdin()
        return finished.addCallback(afterProcessEnd) 

def test_echo(self):
        """
        A spawning a subprocess which echoes its stdin to its stdout via
        L{IReactorProcess.spawnProcess} will result in that echoed output being
        delivered to outReceived.
        """
        finished = defer.Deferred()
        p = EchoProtocol(finished)

        scriptPath = b"twisted.test.process_echoer"
        reactor.spawnProcess(p, pyExe, [pyExe, b'-u', b"-m", scriptPath],
                             env=properEnv)

        def asserts(ignored):
            self.assertFalse(p.failure, p.failure)
            self.assertTrue(hasattr(p, 'buffer'))
            self.assertEqual(len(p.buffer), len(p.s * p.n))

        def takedownProcess(err):
            p.transport.closeStdin()
            return err

        return finished.addCallback(asserts).addErrback(takedownProcess) 

def test_executionError(self):
        """
        Raise an error during execvpe to check error management.
        """
        cmd = self.getCommand('false')

        d = defer.Deferred()
        p = TrivialProcessProtocol(d)
        def buggyexecvpe(command, args, environment):
            raise RuntimeError("Ouch")
        oldexecvpe = os.execvpe
        os.execvpe = buggyexecvpe
        try:
            reactor.spawnProcess(p, cmd, [b'false'], env=None,
                                 usePTY=self.usePTY)

            def check(ignored):
                errData = b"".join(p.errData + p.outData)
                self.assertIn(b"Upon execvpe", errData)
                self.assertIn(b"Ouch", errData)
            d.addCallback(check)
        finally:
            os.execvpe = oldexecvpe
        return d 

def test_mockFork(self):
        """
        Test a classic spawnProcess. Check the path of the client code:
        fork, exec, exit.
        """
        gc.enable()

        cmd = b'/mock/ouch'

        d = defer.Deferred()
        p = TrivialProcessProtocol(d)
        try:
            reactor.spawnProcess(p, cmd, [b'ouch'], env=None,
                                 usePTY=False)
        except SystemError:
            self.assertTrue(self.mockos.exited)
            self.assertEqual(
                self.mockos.actions, [("fork", False), "exec", ("exit", 1)])
        else:
            self.fail("Should not be here")

        # It should leave the garbage collector disabled.
        self.assertFalse(gc.isenabled()) 

def _mockForkInParentTest(self):
        """
        Assert that in the main process, spawnProcess disables the garbage
        collector, calls fork, closes the pipe file descriptors it created for
        the child process, and calls waitpid.
        """
        self.mockos.child = False
        cmd = b'/mock/ouch'

        d = defer.Deferred()
        p = TrivialProcessProtocol(d)
        reactor.spawnProcess(p, cmd, [b'ouch'], env=None,
                             usePTY=False)
        # It should close the first read pipe, and the 2 last write pipes
        self.assertEqual(set(self.mockos.closed), set([-1, -4, -6]))
        self.assertEqual(self.mockos.actions, [("fork", False), "waitpid"]) 

def test_mockSetUid(self):
        """
        Try creating a process with setting its uid: it's almost the same path
        as the standard path, but with a C{switchUID} call before the exec.
        """
        cmd = b'/mock/ouch'

        d = defer.Deferred()
        p = TrivialProcessProtocol(d)
        try:
            reactor.spawnProcess(p, cmd, [b'ouch'], env=None,
                                 usePTY=False, uid=8080)
        except SystemError:
            self.assertTrue(self.mockos.exited)
            self.assertEqual(
                self.mockos.actions,
                [('fork', False), ('setuid', 0), ('setgid', 0),
                 ('switchuid', 8080, 1234), 'exec', ('exit', 1)])
        else:
            self.fail("Should not be here") 

def test_mockPTYSetUidInParent(self):
        """
        When spawning a child process with PTY and a UID different from the UID
        of the current process, the current process does not have its UID
        changed.
        """
        self.mockos.child = False
        cmd = b'/mock/ouch'

        d = defer.Deferred()
        p = TrivialProcessProtocol(d)
        oldPTYProcess = process.PTYProcess
        try:
            process.PTYProcess = DumbPTYProcess
            reactor.spawnProcess(p, cmd, [b'ouch'], env=None,
                                 usePTY=True, uid=8080)
        finally:
            process.PTYProcess = oldPTYProcess
        self.assertEqual(self.mockos.actions, [('fork', False), 'waitpid']) 

def test_mockWithWaitError(self):
        """
        Test that reapProcess logs errors raised.
        """
        self.mockos.child = False
        cmd = b'/mock/ouch'
        self.mockos.waitChild = (0, 0)

        d = defer.Deferred()
        p = TrivialProcessProtocol(d)
        proc = reactor.spawnProcess(p, cmd, [b'ouch'], env=None,
                             usePTY=False)
        self.assertEqual(self.mockos.actions, [("fork", False), "waitpid"])

        self.mockos.raiseWaitPid = OSError()
        proc.reapProcess()
        errors = self.flushLoggedErrors()
        self.assertEqual(len(errors), 1)
        errors[0].trap(OSError) 

def test_mockErrorECHILDInReapProcess(self):
        """
        Test that reapProcess doesn't log anything when waitpid raises a
        C{OSError} with errno C{ECHILD}.
        """
        self.mockos.child = False
        cmd = b'/mock/ouch'
        self.mockos.waitChild = (0, 0)

        d = defer.Deferred()
        p = TrivialProcessProtocol(d)
        proc = reactor.spawnProcess(p, cmd, [b'ouch'], env=None,
                                    usePTY=False)
        self.assertEqual(self.mockos.actions, [("fork", False), "waitpid"])

        self.mockos.raiseWaitPid = OSError()
        self.mockos.raiseWaitPid.errno = errno.ECHILD
        # This should not produce any errors
        proc.reapProcess() 

def test_mockErrorInPipe(self):
        """
        If C{os.pipe} raises an exception after some pipes where created, the
        created pipes are closed and don't leak.
        """
        pipes = [-1, -2, -3, -4]
        def pipe():
            try:
                return pipes.pop(0), pipes.pop(0)
            except IndexError:
                raise OSError()
        self.mockos.pipe = pipe
        protocol = TrivialProcessProtocol(None)
        self.assertRaises(OSError, reactor.spawnProcess, protocol, None)
        self.assertEqual(self.mockos.actions, [])
        self.assertEqual(set(self.mockos.closed), set([-4, -3, -2, -1])) 

def test_stderr(self):
        """
        Bytes written to stderr by the spawned process are passed to the
        C{errReceived} callback on the C{ProcessProtocol} passed to
        C{spawnProcess}.
        """
        value = "42"

        p = Accumulator()
        d = p.endedDeferred = defer.Deferred()
        reactor.spawnProcess(p, pyExe,
                             [pyExe, b"-c",
                              networkString("import sys; sys.stderr.write"
                                            "('{0}')".format(value))],
                             env=None, path="/tmp",
                             usePTY=self.usePTY)

        def processEnded(ign):
            self.assertEqual(b"42", p.errF.getvalue())
        return d.addCallback(processEnded) 

def test_process(self):
        cmd = self.getCommand('gzip')
        s = b"there's no place like home!\n" * 3
        p = Accumulator()
        d = p.endedDeferred = defer.Deferred()
        reactor.spawnProcess(p, cmd, [cmd, b"-c"], env=None, path="/tmp",
                             usePTY=self.usePTY)
        p.transport.write(s)
        p.transport.closeStdin()

        def processEnded(ign):
            f = p.outF
            f.seek(0, 0)
            with gzip.GzipFile(fileobj=f) as gf:
                self.assertEqual(gf.read(), s)
        return d.addCallback(processEnded) 

def test_openingTTY(self):
        scriptPath = b"twisted.test.process_tty"
        p = Accumulator()
        d = p.endedDeferred = defer.Deferred()
        reactor.spawnProcess(p, pyExe, [pyExe, b"-u", b"-m", scriptPath],
                             env=properEnv, usePTY=self.usePTY)
        p.transport.write(b"hello world!\n")

        def processEnded(ign):
            self.assertRaises(
                error.ProcessExitedAlready, p.transport.signalProcess, 'HUP')
            self.assertEqual(
                p.outF.getvalue(),
                b"hello world!\r\nhello world!\r\n",
                ("Error message from process_tty "
                 "follows:\n\n%s\n\n" % (p.outF.getvalue(),)))
        return d.addCallback(processEnded) 

def test_encodableUnicodeEnvironment(self):
        """
        Test C{os.environ} (inherited by every subprocess on Windows) that
        contains an ascii-encodable Unicode string. This is different from
        passing Unicode environment explicitly to spawnProcess (which is not
        supported on Python 2).
        """
        os.environ[self.goodKey] = self.goodValue
        self.addCleanup(operator.delitem, os.environ, self.goodKey)

        p = GetEnvironmentDictionary.run(reactor, [], properEnv)
        def gotEnvironment(environ):
            self.assertEqual(
                environ[self.goodKey.encode('ascii')],
                self.goodValue.encode('ascii'))
        return p.getResult().addCallback(gotEnvironment) 

def test_twisted(self):
        """Invoking python -m twisted should execute twist."""
        cmd = sys.executable
        p = Accumulator()
        d = p.endedDeferred = defer.Deferred()
        reactor.spawnProcess(p, cmd, [cmd, '-m', 'twisted', '--help'], env=None)
        p.transport.closeStdin()

        def processEnded(ign):
            f = p.outF
            output = f.getvalue().replace(b'\r\n', b'\n')

            options = TwistOptions()
            message = '{}\n'.format(options).encode('utf-8')
            self.assertEqual(output, message)
        return d.addCallback(processEnded) 

def test_stdin(self):
        """
        Making sure getPassword accepts a password from standard input by
        running a child process which uses getPassword to read in a string
        which it then writes it out again.  Write a string to the child
        process and then read one and make sure it is the right string.
        """
        p = PasswordTestingProcessProtocol()
        p.finished = Deferred()
        reactor.spawnProcess(
            p, pyExe,
            [pyExe,
             b'-c',
             (b'import sys\n'
              b'from twisted.python.util import getPassword\n'
              b'sys.stdout.write(getPassword())\n'
              b'sys.stdout.flush()\n')],
            env={b'PYTHONPATH': os.pathsep.join(sys.path).encode("utf8")})

        def processFinished(result):
            (reason, output) = result
            reason.trap(ProcessDone)
            self.assertIn((1, b'secret'), output)

        return p.finished.addCallback(processFinished) 

def _spawn(script, outputFD):
    """
    Start a script that is a peer of this test as a subprocess.

    @param script: the module name of the script in this directory (no
        package prefix, no '.py')
    @type script: C{str}

    @rtype: L{StartStopProcessProtocol}
    """
    pyExe = FilePath(sys.executable).asBytesMode().path
    env = bytesEnviron()
    env[b"PYTHONPATH"] = FilePath(
        pathsep.join(sys.path)).asBytesMode().path
    sspp = StartStopProcessProtocol()
    reactor.spawnProcess(
        sspp, pyExe, [
            pyExe,
            FilePath(__file__).sibling(script + ".py").asBytesMode().path,
            intToBytes(outputFD),
        ],
        env=env,
        childFDs={0: "w", 1: "r", 2: "r", outputFD: outputFD}
    )
    return sspp 

def test_stdio(self):
        """
        L{twisted.internet.stdio} test.
        """
        scriptPath = b"twisted.test.process_twisted"
        p = Accumulator()
        d = p.endedDeferred = defer.Deferred()
        reactor.spawnProcess(p, pyExe, [pyExe, b'-u', b"-m", scriptPath],
                             env=properEnv,
                             path=None, usePTY=self.usePTY)
        p.transport.write(b"hello, world")
        p.transport.write(b"abc")
        p.transport.write(b"123")
        p.transport.closeStdin()

        def processEnded(ign):
            self.assertEqual(p.outF.getvalue(), b"hello, worldabc123",
                              "Output follows:\n"
                              "%s\n"
                              "Error message from process_twisted follows:\n"
                              "%s\n" % (p.outF.getvalue(), p.errF.getvalue()))
        return d.addCallback(processEnded) 

def test_unsetPid(self):
        """
        Test if pid is None/non-None before/after process termination.  This
        reuses process_echoer.py to get a process that blocks on stdin.
        """
        finished = defer.Deferred()
        p = TrivialProcessProtocol(finished)
        scriptPath = b"twisted.test.process_echoer"
        procTrans = reactor.spawnProcess(p, pyExe,
                                         [pyExe, b'-u', b"-m", scriptPath],
                                         env=properEnv)
        self.assertTrue(procTrans.pid)

        def afterProcessEnd(ignored):
            self.assertIsNone(procTrans.pid)

        p.transport.closeStdin()
        return finished.addCallback(afterProcessEnd) 

def startService(self):
        service.MultiService.startService(self)
        for i in range(max(1, multiprocessing.cpu_count() - 1)):
            checker = reactor.spawnProcess(
                CheckerProcessProtocol(), sys.executable,
                ['moira-checker', WORKER_PATH, "-n", str(i), "-c", config.CONFIG_PATH, "-l", config.LOG_DIRECTORY],
                childFDs={0: 'w', 1: 1, 2: 2}, env=os.environ)
            self.checkers.append(checker) 

def __start_spawn_proc(self, proc, cmd):
        env = os.environ
        try:
            running_p = reactor.spawnProcess(proc, cmd[0], cmd, env)
        except Exception as exc:
            self.log.error(
                'Error while spawning vpn process... {0!r}'.format(exc))
            raise exc
        return running_p 

def spawn(self, cmdline, timeout=0, lines_max=0):
        """Spawn the process and run on_exit on exit, timeout, or lines_max."""
        LOG.d("spawn: {}, timeout: {}, lines_max: {}".format(
            cmdline, timeout, lines_max))
        assert self.state == "idle"
        # Reinitialize vars without duplicating a lot of code
        self.__init__(self.on_exit)
        self.lines_max = lines_max
        if timeout:
            # noinspection PyUnresolvedReferences
            self.dc_timeout = reactor.callLater(timeout, self.stop, "timeout")
        else:
            self.dc_timeout = None
        # noinspection PyUnresolvedReferences
        reactor.spawnProcess(self, cmdline[0], cmdline) 

def test_pathInfo(self):
        """
        L{twcgi.CGIScript.render} sets the process environment I{PATH_INFO} from
        the request path.
        """
        class FakeReactor:
            """
            A fake reactor recording the environment passed to spawnProcess.
            """
            def spawnProcess(self, process, filename, args, env, wdir):
                """
                Store the C{env} L{dict} to an instance attribute.

                @param process: Ignored
                @param filename: Ignored
                @param args: Ignored
                @param env: The environment L{dict} which will be stored
                @param wdir: Ignored
                """
                self.process_env = env

        _reactor = FakeReactor()
        resource = twcgi.CGIScript(self.mktemp(), reactor=_reactor)
        request = DummyRequest(['a', 'b'])
        _render(resource, request)

        self.assertEqual(_reactor.process_env["PATH_INFO"],
                         "/a/b") 

def _callProtocolWithDeferred(protocol, executable, args, env, path, reactor=None):
    if reactor is None:
        from twisted.internet import reactor

    d = defer.Deferred()
    p = protocol(d)
    reactor.spawnProcess(p, executable, (executable,)+tuple(args), env, path)
    return d 

def test_noCompatibilityLayer(self):
        """
        If no compatibility layer is present, imports of gobject and friends
        are disallowed.

        We do this by running a process where we make sure gi.pygtkcompat
        isn't present.
        """
        if _PY3:
            raise SkipTest("Python3 always has the compatibility layer.")

        from twisted.internet import reactor
        if not IReactorProcess.providedBy(reactor):
            raise SkipTest("No process support available in this reactor.")

        result = Deferred()
        class Stdout(ProcessProtocol):
            data = b""

            def errReceived(self, err):
                print(err)

            def outReceived(self, data):
                self.data += data

            def processExited(self, reason):
                result.callback(self.data)

        path = FilePath(__file__).sibling(b"process_gireactornocompat.py").path
        pyExe = FilePath(sys.executable)._asBytesPath()
        # Pass in a PYTHONPATH that is the test runner's os.path, to make sure
        # we're running from a checkout
        reactor.spawnProcess(Stdout(), pyExe, [pyExe, path],
                             env={"PYTHONPATH": ":".join(sys.path)})
        result.addCallback(self.assertEqual, b"success")
        return result 

def execute(self, remoteCommand, process, sshArgs='', conchArgs=None):
        """
        As for L{OpenSSHClientTestCase.execute}, except it runs the 'conch'
        command line tool, not 'ssh'.
        """
        if conchArgs is None:
            conchArgs = []

        process.deferred = defer.Deferred()
        port = self.conchServer.getHost().port
        cmd = ('-p {} -l testuser '
               '--known-hosts kh_test '
               '--user-authentications publickey '
               '-a '
               '-i dsa_test '
               '-v '.format(port) + sshArgs +
               ' 127.0.0.1 ' + remoteCommand)
        cmds = _makeArgs(conchArgs + cmd.split())
        env = os.environ.copy()
        env['PYTHONPATH'] = os.pathsep.join(sys.path)
        encodedCmds = []
        encodedEnv = {}
        for cmd in cmds:
            if isinstance(cmd, unicode):
                cmd = cmd.encode("utf-8")
            encodedCmds.append(cmd)
        for var in env:
            val = env[var]
            if isinstance(var, unicode):
                var = var.encode("utf-8")
            if isinstance(val, unicode):
                val = val.encode("utf-8")
            encodedEnv[var] = val
        reactor.spawnProcess(process, sys.executable, encodedCmds, env=encodedEnv)
        return process.deferred