comparison tests/run-tests.py @ 25031:0adc22a0b6b3

python3: update killdaemons and run-tests print and exception syntax

test-run-tests.t still passes fine on Python 2.6. run-tests.py --local
no longer fails with syntax errors, and now fails looking for xrange.

Most changes done with

2to3 -w -f numliterals -f except -f print tests/run-tests.py tests/killdaemons.py

after which one import was fixed in run-tests and a __future__ import
was added.

25030:33298a8b472f

# (You could use any subset of the tests: test-s* happens to match
# enough that it's worth doing parallel runs, few enough that it
# completes fairly quickly, includes both shell and Python scripts, and
# includes some scripts that run daemon processes.)

from distutils import version
import difflib
import errno
import optparse
import os

import time
import random
import re
import threading
import killdaemons as killmod
import Queue as queue
from xml.dom import minidom
import unittest

try:
    import json

    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.bind(('localhost', port))
        s.close()
        return True
    except socket.error, exc:
        if not exc.errno == errno.EADDRINUSE:
            raise
        return False

closefds = os.name == 'posix'

    entries = dict()
    for filename in files:
        try:
            path = os.path.expanduser(os.path.expandvars(filename))
            f = open(path, "rb")
        except IOError, err:
            if err.errno != errno.ENOENT:
                raise
            if warn:
                print "warning: no such %s file: %s" % (listtype, filename)
            continue

        for line in f.readlines():
            line = line.split('#', 1)[0].strip()
            if line:

    Arguments are strings to print.
    """
    iolock.acquire()
    if verbose:
        print verbose,
    for m in msg:
        print m,
    print
    sys.stdout.flush()
    iolock.release()

def terminate(proc):
    """Terminate subprocess (with fallback for Python versions < 2.6)"""

        self._out = None
        self._skipped = None

        try:
            os.mkdir(self._threadtmp)
        except OSError, e:
            if e.errno != errno.EEXIST:
                raise

        self._testtmp = os.path.join(self._threadtmp,
                                     os.path.basename(self.path))

        # Remove any previous output files.
        if os.path.exists(self.errpath):
            try:
                os.remove(self.errpath)
            except OSError, e:
                # We might have raced another test to clean up a .err
                # file, so ignore ENOENT when removing a previous .err
                # file.
                if e.errno != errno.ENOENT:
                    raise

            try:
                self.runTest()
            except KeyboardInterrupt:
                self._aborted = True
                raise
            except SkipTest, e:
                result.addSkip(self, str(e))
                # The base class will have already counted this as a
                # test we "ran", but we want to exclude skipped tests
                # from those we count towards those run.
                result.testsRun -= 1
            except IgnoreTest, e:
                result.addIgnore(self, str(e))
                # As with skips, ignores also should be excluded from
                # the number of tests executed.
                result.testsRun -= 1
            except WarnTest, e:
                result.addWarn(self, str(e))
            except self.failureException, e:
                # This differs from unittest in that we don't capture
                # the stack trace. This is for historical reasons and
                # this decision could be revisited in the future,
                # especially for PythonTest instances.
                if result.addFailure(self, str(e)):

        stdout, stderr = proc.communicate()
        ret = proc.wait()
        if wifexited(ret):
            ret = os.WEXITSTATUS(ret)
        if ret == 2:
            print stdout
            sys.exit(1)

        return ret == 0

    def _parsetest(self, lines):

        self._portoffset = 0
        self._ports = {}

    def run(self, args, parser=None):
        """Run the test suite."""
        oldmask = os.umask(022)
        try:
            parser = parser or getparser()
            options, args = parseargs(args, parser)
            self.options = options

            slow = 'svn gendoc check-code-hg'.split()
            def sortkey(f):
                # run largest tests first, as they tend to take the longest
                try:
                    val = -os.stat(f).st_size
                except OSError, e:
                    if e.errno != errno.ENOENT:
                        raise
                    return -1e9 # file does not exist, tell early
                for kw in slow:
                    if kw in f:

            tmpdir = self.options.tmpdir
            if os.path.exists(tmpdir):
                # Meaning of tmpdir has changed since 1.3: we used to create
                # HGTMP inside tmpdir; now HGTMP is tmpdir.  So fail if
                # tmpdir already exists.
                print "error: temp dir %r already exists" % tmpdir
                return 1

                # Automatically removing tmpdir sounds convenient, but could
                # really annoy anyone in the habit of using "--tmpdir=/tmp"
                # or "--tmpdir=$HOME".

                while tests:
                    if os.path.exists(tests[0] + ".err"):
                        break
                    tests.pop(0)
                if not tests:
                    print "running all tests"
                    tests = orig

            tests = [self._gettest(t, i) for i, t in enumerate(tests)]

            failed = False

            if self.options.anycoverage:
                self._outputcoverage()
        except KeyboardInterrupt:
            failed = True
            print "\ninterrupted!"

        if failed:
            return 1
        if warned:
            return 80

            mypython = os.path.join(self._tmpbindir, pyexename)
            try:
                if os.readlink(mypython) == sys.executable:
                    return
                os.unlink(mypython)
            except OSError, err:
                if err.errno != errno.ENOENT:
                    raise
            if self._findprogram(pyexename) != sys.executable:
                try:
                    os.symlink(sys.executable, mypython)
                    self._createdfiles.append(mypython)
                except OSError, err:
                    # child processes may race, which is harmless
                    if err.errno != errno.EEXIST:
                        raise
        else:
            exedir, exename = os.path.split(sys.executable)

            path = os.environ['PATH'].split(os.pathsep)
            while exedir in path:
                path.remove(exedir)
            os.environ['PATH'] = os.pathsep.join([exedir] + path)
            if not self._findprogram(pyexename):
                print "WARNING: Cannot find %s in search path" % pyexename

    def _installhg(self):
        """Install hg into the test environment.

        This will also configure hg with the appropriate testing settings.

        # setuptools requires install directories to exist.
        def makedirs(p):
            try:
                os.makedirs(p)
            except OSError, e:
                if e.errno != errno.EEXIST:
                    raise
        makedirs(self._pythondir)
        makedirs(self._bindir)

                                    '"%~dp0python" "%~dp0hg" %*')
                f = open(hgbat, 'wb')
                f.write(data)
                f.close()
            else:
                print 'WARNING: cannot fix hg.bat reference to python.exe'

        if self.options.anycoverage:
            custom = os.path.join(self._testdir, 'sitecustomize.py')
            target = os.path.join(self._pythondir, 'sitecustomize.py')
            vlog('# Installing coverage trigger to %s' % target)

            vlog('# Installing coverage rc to %s' % rc)
            os.environ['COVERAGE_PROCESS_START'] = rc
            covdir = os.path.join(self._installdir, '..', 'coverage')
            try:
                os.mkdir(covdir)
            except OSError, e:
                if e.errno != errno.EEXIST:
                    raise

            os.environ['COVERAGE_DIR'] = covdir

                p += '.exe'
            found = self._findprogram(p)
            if found:
                vlog("# Found prerequisite", p, "at", found)
            else:
                print "WARNING: Did not find prerequisite tool: %s " % p

if __name__ == '__main__':
    runner = TestRunner()

    try:

25031:0adc22a0b6b3

# (You could use any subset of the tests: test-s* happens to match
# enough that it's worth doing parallel runs, few enough that it
# completes fairly quickly, includes both shell and Python scripts, and
# includes some scripts that run daemon processes.)

from __future__ import print_function

from distutils import version
import difflib
import errno
import optparse
import os

import time
import random
import re
import threading
import killdaemons as killmod
try:
    import Queue as queue
except ImportError:
    import queue
from xml.dom import minidom
import unittest

try:
    import json

    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.bind(('localhost', port))
        s.close()
        return True
    except socket.error as exc:
        if not exc.errno == errno.EADDRINUSE:
            raise
        return False

closefds = os.name == 'posix'

    entries = dict()
    for filename in files:
        try:
            path = os.path.expanduser(os.path.expandvars(filename))
            f = open(path, "rb")
        except IOError as err:
            if err.errno != errno.ENOENT:
                raise
            if warn:
                print("warning: no such %s file: %s" % (listtype, filename))
            continue

        for line in f.readlines():
            line = line.split('#', 1)[0].strip()
            if line:

    Arguments are strings to print.
    """
    iolock.acquire()
    if verbose:
        print(verbose, end=' ')
    for m in msg:
        print(m, end=' ')
    print()
    sys.stdout.flush()
    iolock.release()

def terminate(proc):
    """Terminate subprocess (with fallback for Python versions < 2.6)"""

        self._out = None
        self._skipped = None

        try:
            os.mkdir(self._threadtmp)
        except OSError as e:
            if e.errno != errno.EEXIST:
                raise

        self._testtmp = os.path.join(self._threadtmp,
                                     os.path.basename(self.path))

        # Remove any previous output files.
        if os.path.exists(self.errpath):
            try:
                os.remove(self.errpath)
            except OSError as e:
                # We might have raced another test to clean up a .err
                # file, so ignore ENOENT when removing a previous .err
                # file.
                if e.errno != errno.ENOENT:
                    raise

            try:
                self.runTest()
            except KeyboardInterrupt:
                self._aborted = True
                raise
            except SkipTest as e:
                result.addSkip(self, str(e))
                # The base class will have already counted this as a
                # test we "ran", but we want to exclude skipped tests
                # from those we count towards those run.
                result.testsRun -= 1
            except IgnoreTest as e:
                result.addIgnore(self, str(e))
                # As with skips, ignores also should be excluded from
                # the number of tests executed.
                result.testsRun -= 1
            except WarnTest as e:
                result.addWarn(self, str(e))
            except self.failureException as e:
                # This differs from unittest in that we don't capture
                # the stack trace. This is for historical reasons and
                # this decision could be revisited in the future,
                # especially for PythonTest instances.
                if result.addFailure(self, str(e)):

        stdout, stderr = proc.communicate()
        ret = proc.wait()
        if wifexited(ret):
            ret = os.WEXITSTATUS(ret)
        if ret == 2:
            print(stdout)
            sys.exit(1)

        return ret == 0

    def _parsetest(self, lines):

        self._portoffset = 0
        self._ports = {}

    def run(self, args, parser=None):
        """Run the test suite."""
        oldmask = os.umask(0o22)
        try:
            parser = parser or getparser()
            options, args = parseargs(args, parser)
            self.options = options

            slow = 'svn gendoc check-code-hg'.split()
            def sortkey(f):
                # run largest tests first, as they tend to take the longest
                try:
                    val = -os.stat(f).st_size
                except OSError as e:
                    if e.errno != errno.ENOENT:
                        raise
                    return -1e9 # file does not exist, tell early
                for kw in slow:
                    if kw in f:

            tmpdir = self.options.tmpdir
            if os.path.exists(tmpdir):
                # Meaning of tmpdir has changed since 1.3: we used to create
                # HGTMP inside tmpdir; now HGTMP is tmpdir.  So fail if
                # tmpdir already exists.
                print("error: temp dir %r already exists" % tmpdir)
                return 1

                # Automatically removing tmpdir sounds convenient, but could
                # really annoy anyone in the habit of using "--tmpdir=/tmp"
                # or "--tmpdir=$HOME".

                while tests:
                    if os.path.exists(tests[0] + ".err"):
                        break
                    tests.pop(0)
                if not tests:
                    print("running all tests")
                    tests = orig

            tests = [self._gettest(t, i) for i, t in enumerate(tests)]

            failed = False

            if self.options.anycoverage:
                self._outputcoverage()
        except KeyboardInterrupt:
            failed = True
            print("\ninterrupted!")

        if failed:
            return 1
        if warned:
            return 80

            mypython = os.path.join(self._tmpbindir, pyexename)
            try:
                if os.readlink(mypython) == sys.executable:
                    return
                os.unlink(mypython)
            except OSError as err:
                if err.errno != errno.ENOENT:
                    raise
            if self._findprogram(pyexename) != sys.executable:
                try:
                    os.symlink(sys.executable, mypython)
                    self._createdfiles.append(mypython)
                except OSError as err:
                    # child processes may race, which is harmless
                    if err.errno != errno.EEXIST:
                        raise
        else:
            exedir, exename = os.path.split(sys.executable)

            path = os.environ['PATH'].split(os.pathsep)
            while exedir in path:
                path.remove(exedir)
            os.environ['PATH'] = os.pathsep.join([exedir] + path)
            if not self._findprogram(pyexename):
                print("WARNING: Cannot find %s in search path" % pyexename)

    def _installhg(self):
        """Install hg into the test environment.

        This will also configure hg with the appropriate testing settings.

        # setuptools requires install directories to exist.
        def makedirs(p):
            try:
                os.makedirs(p)
            except OSError as e:
                if e.errno != errno.EEXIST:
                    raise
        makedirs(self._pythondir)
        makedirs(self._bindir)

                                    '"%~dp0python" "%~dp0hg" %*')
                f = open(hgbat, 'wb')
                f.write(data)
                f.close()
            else:
                print('WARNING: cannot fix hg.bat reference to python.exe')

        if self.options.anycoverage:
            custom = os.path.join(self._testdir, 'sitecustomize.py')
            target = os.path.join(self._pythondir, 'sitecustomize.py')
            vlog('# Installing coverage trigger to %s' % target)

            vlog('# Installing coverage rc to %s' % rc)
            os.environ['COVERAGE_PROCESS_START'] = rc
            covdir = os.path.join(self._installdir, '..', 'coverage')
            try:
                os.mkdir(covdir)
            except OSError as e:
                if e.errno != errno.EEXIST:
                    raise

            os.environ['COVERAGE_DIR'] = covdir

                p += '.exe'
            found = self._findprogram(p)
            if found:
                vlog("# Found prerequisite", p, "at", found)
            else:
                print("WARNING: Did not find prerequisite tool: %s " % p)

if __name__ == '__main__':
    runner = TestRunner()

    try: