comparison tests/run-tests.py @ 21521:855f092c96e5

run-tests: move diff generation into TestResult

TestResult is the thing that captures all our test results. It's logical
for diff viewing to be handled there and not inside Test.

While writing this patch, it was discovered that Test.fail() was
printing redundant test result output. This has been removed.

Arguments controlling diffs have been removed from Test.__init__.

21520:fa6ba4372678

    for existing destination support.
    """
    shutil.copy(src, dst)
    os.remove(src)

def showdiff(expected, output, ref, err):
    print
    servefail = False
    for line in difflib.unified_diff(expected, output, ref, err):
        sys.stdout.write(line)
        if not servefail and line.startswith(
                             '+  abort: child process failed to start'):
            servefail = True
    return {'servefail': servefail}


verbose = False
def vlog(*msg):
    if verbose is not False:
        iolock.acquire()

    # Status code reserved for skipped tests (used by hghave).
    SKIPPED_STATUS = 80

    def __init__(self, path, tmpdir, keeptmpdir=False,
                 debug=False, nodiff=False, diffviewer=None,
                 interactive=False, timeout=defaults['timeout'],
                 startport=defaults['port'], extraconfigopts=None,
                 py3kwarnings=False, shell=None):
        """Create a test from parameters.

        after execution. It defaults to removal (False).

        debug mode will make the test execute verbosely, with unfiltered
        output.

        nodiff will suppress the printing of a diff when output changes.

        diffviewer is the program that should be used to display diffs. Only
        used when output changes.

        interactive controls whether the test will run interactively.

        timeout controls the maximum run time of the test. It is ignored when
        debug is True.

        self.errpath = os.path.join(self._testdir, '%s.err' % self.name)

        self._threadtmp = tmpdir
        self._keeptmpdir = keeptmpdir
        self._debug = debug
        self._nodiff = nodiff
        self._diffviewer = diffviewer
        self._interactive = interactive
        self._timeout = timeout
        self._startport = startport
        self._extraconfigopts = extraconfigopts or []
        self._py3kwarnings = py3kwarnings

        # If we're not in --debug mode and reference output file exists,
        # check test output against it.
        if debug:
            self._refout = None # to match "out is None"
        elif os.path.exists(self._refpath):
            f = open(self._refpath, 'r')
            self._refout = f.read().splitlines(True)
            f.close()
        else:
            self._refout = []

        # Remove any previous output files.
        if os.path.exists(self.errpath):
            os.remove(self.errpath)

    def run(self, result):
        result.startTest(self)
        try:
            try:
                self.setUp()
            except (KeyboardInterrupt, SystemExit):

                self._skipped = True
                raise SkipTest(missing[-1])
        elif ret == 'timeout':
            self.fail('timed out', ret)
        elif out != self._refout:
            info = {}
            if not self._nodiff:
                iolock.acquire()
                if self._diffviewer:
                    os.system("%s %s %s" % (self._diffviewer, self._refpath,
                                            self.errpath))
                else:
                    info = showdiff(self._refout, out, self._refpath,
                                    self.errpath)
                iolock.release()
            msg = ''
            if info.get('servefail'):
                msg += 'serve failed and '
            if ret:
                msg += 'output changed and ' + describe(ret)
            else:
                msg += 'output changed'

            if (ret != 0 or out != self._refout) and not self._skipped \
                and not self._debug:
                f = open(self.errpath, 'wb')
                for line in out:

            hgrc.write('[%s]\n%s\n' % (section, key))
        hgrc.close()

    def fail(self, msg, ret):
        warned = ret is False
        if not self._nodiff:
            log("\n%s: %s %s" % (warned and 'Warning' or 'ERROR', self.name,
                                 msg))
        if (not ret and self._interactive and
            os.path.exists(self.errpath)):
            iolock.acquire()
            print 'Accept this change? [n] ',
            answer = sys.stdin.readline().strip()

class PythonTest(Test):
    """A Python-based test."""

    @property
    def _refpath(self):
        return os.path.join(self._testdir, '%s.out' % self.name)

    def _run(self, replacements, env):
        py3kswitch = self._py3kwarnings and ' -3' or ''
        cmd = '%s%s "%s"' % (PYTHON, py3kswitch, self.path)

    ESCAPESUB = re.compile(r'[\x00-\x08\x0b-\x1f\\\x7f-\xff]').sub
    ESCAPEMAP = dict((chr(i), r'\x%02x' % i) for i in range(256)).update(
                     {'\\': '\\\\', '\r': r'\r'})

    @property
    def _refpath(self):
        return os.path.join(self._testdir, self.name)

    def _run(self, replacements, env):
        f = open(self.path)
        lines = f.readlines()

            self.stream.writeln('warned %s' % reason)
        else:
            self.stream.write('~')
            self.stream.flush()

    def startTest(self, test):
        super(TestResult, self).startTest(test)

        self._started[test.name] = time.time()

        tmpdir = os.path.join(self.hgtmp, 'child%d' % count)

        return testcls(refpath, tmpdir,
                       keeptmpdir=self.options.keep_tmpdir,
                       debug=self.options.debug,
                       nodiff = self.options.nodiff,
                       diffviewer=self.options.view,
                       interactive=self.options.interactive,
                       timeout=self.options.timeout,
                       startport=self.options.port + count * 3,
                       extraconfigopts=self.options.extra_config_opt,
                       py3kwarnings=self.options.py3k_warnings,

21521:855f092c96e5

    for existing destination support.
    """
    shutil.copy(src, dst)
    os.remove(src)

def getdiff(expected, output, ref, err):
    servefail = False
    lines = []
    for line in difflib.unified_diff(expected, output, ref, err):
        lines.append(line)
        if not servefail and line.startswith(
                             '+  abort: child process failed to start'):
            servefail = True

    return servefail, lines

verbose = False
def vlog(*msg):
    if verbose is not False:
        iolock.acquire()

    # Status code reserved for skipped tests (used by hghave).
    SKIPPED_STATUS = 80

    def __init__(self, path, tmpdir, keeptmpdir=False,
                 debug=False,
                 interactive=False, timeout=defaults['timeout'],
                 startport=defaults['port'], extraconfigopts=None,
                 py3kwarnings=False, shell=None):
        """Create a test from parameters.

        after execution. It defaults to removal (False).

        debug mode will make the test execute verbosely, with unfiltered
        output.

        interactive controls whether the test will run interactively.

        timeout controls the maximum run time of the test. It is ignored when
        debug is True.

        self.errpath = os.path.join(self._testdir, '%s.err' % self.name)

        self._threadtmp = tmpdir
        self._keeptmpdir = keeptmpdir
        self._debug = debug
        self._interactive = interactive
        self._timeout = timeout
        self._startport = startport
        self._extraconfigopts = extraconfigopts or []
        self._py3kwarnings = py3kwarnings

        # If we're not in --debug mode and reference output file exists,
        # check test output against it.
        if debug:
            self._refout = None # to match "out is None"
        elif os.path.exists(self.refpath):
            f = open(self.refpath, 'r')
            self._refout = f.read().splitlines(True)
            f.close()
        else:
            self._refout = []

        # Remove any previous output files.
        if os.path.exists(self.errpath):
            os.remove(self.errpath)

    def run(self, result):
        self._result = result
        result.startTest(self)
        try:
            try:
                self.setUp()
            except (KeyboardInterrupt, SystemExit):

                self._skipped = True
                raise SkipTest(missing[-1])
        elif ret == 'timeout':
            self.fail('timed out', ret)
        elif out != self._refout:
            # The result object handles diff calculation for us.
            self._result.addOutputMismatch(self, out, self._refout)

            if ret:
                msg = 'output changed and ' + describe(ret)
            else:
                msg = 'output changed'

            if (ret != 0 or out != self._refout) and not self._skipped \
                and not self._debug:
                f = open(self.errpath, 'wb')
                for line in out:

            hgrc.write('[%s]\n%s\n' % (section, key))
        hgrc.close()

    def fail(self, msg, ret):
        warned = ret is False
        if (not ret and self._interactive and
            os.path.exists(self.errpath)):
            iolock.acquire()
            print 'Accept this change? [n] ',
            answer = sys.stdin.readline().strip()

class PythonTest(Test):
    """A Python-based test."""

    @property
    def refpath(self):
        return os.path.join(self._testdir, '%s.out' % self.name)

    def _run(self, replacements, env):
        py3kswitch = self._py3kwarnings and ' -3' or ''
        cmd = '%s%s "%s"' % (PYTHON, py3kswitch, self.path)

    ESCAPESUB = re.compile(r'[\x00-\x08\x0b-\x1f\\\x7f-\xff]').sub
    ESCAPEMAP = dict((chr(i), r'\x%02x' % i) for i in range(256)).update(
                     {'\\': '\\\\', '\r': r'\r'})

    @property
    def refpath(self):
        return os.path.join(self._testdir, self.name)

    def _run(self, replacements, env):
        f = open(self.path)
        lines = f.readlines()

            self.stream.writeln('warned %s' % reason)
        else:
            self.stream.write('~')
            self.stream.flush()

    def addOutputMismatch(self, test, got, expected):
        """Record a mismatch in test output for a particular test."""

        if self._options.nodiff:
            return

        if self._options.view:
            os.system("%s %s %s" % (self._view, test.refpath, test.errpath))
        else:
            failed, lines = getdiff(expected, got,
                                    test.refpath, test.errpath)
            if failed:
                self.addFailure(test, 'diff generation failed')
            else:
                self.stream.write('\n')
                for line in lines:
                    self.stream.write(line)
                self.stream.flush()

    def startTest(self, test):
        super(TestResult, self).startTest(test)

        self._started[test.name] = time.time()

        tmpdir = os.path.join(self.hgtmp, 'child%d' % count)

        return testcls(refpath, tmpdir,
                       keeptmpdir=self.options.keep_tmpdir,
                       debug=self.options.debug,
                       interactive=self.options.interactive,
                       timeout=self.options.timeout,
                       startport=self.options.port + count * 3,
                       extraconfigopts=self.options.extra_config_opt,
                       py3kwarnings=self.options.py3k_warnings,