test_internet.py 文件源码

def testChildResolve(self):
        # I've seen problems with reactor.run under gtk2reactor. Spawn a
        # child which just does reactor.resolve after the reactor has
        # started, fail if it does not complete in a timely fashion.
        helperPath = os.path.abspath(self.mktemp())
        with open(helperPath, 'w') as helperFile:

            # Eeueuuggg
            reactorName = reactor.__module__

            helperFile.write(resolve_helper % {'reactor': reactorName})

        env = os.environ.copy()
        env['PYTHONPATH'] = os.pathsep.join(sys.path)

        helperDeferred = Deferred()
        helperProto = ChildResolveProtocol(helperDeferred)

        reactor.spawnProcess(helperProto, sys.executable, ("python", "-u", helperPath), env)

        def cbFinished(result):
            (reason, output, error) = result
            # If the output is "done 127.0.0.1\n" we don't really care what
            # else happened.
            output = b''.join(output)
            if _PY3:
                expected_output = (b'done 127.0.0.1' +
                                   os.linesep.encode("ascii"))
            else:
                expected_output = b'done 127.0.0.1\n'
            if output != expected_output:
                self.fail((
                    "The child process failed to produce the desired results:\n"
                    "   Reason for termination was: %r\n"
                    "   Output stream was: %r\n"
                    "   Error stream was: %r\n") % (reason.getErrorMessage(), output, b''.join(error)))

        helperDeferred.addCallback(cbFinished)
        return helperDeferred