comparison tests/run-tests.py @ 21523:9fb6f328576a

run-tests: move interactive test acceptance into TestResult

We shouldn't be performing interactive prompting during test execution.
The first step of this is to move the functionality into TestResult.

21522:eeaec308ad5f

    # 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.

        path is the full path to the file defining the test.

        keeptmpdir determines whether to keep the test's temporary directory
        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.

        startport controls the starting port number to use for this test. Each

        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
        self._shell = shell

            if not missing:
                missing = ['irrelevant']

            if failed:
                self.fail('hg have failed checking for %s' % failed[-1], ret)
            else:
                self._skipped = True
                raise SkipTest(missing[-1])
        elif ret == 'timeout':
            self.fail('timed out', ret)
        elif ret is False:
            raise WarnTest('no result code from test')
        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'

                f = open(self.errpath, 'wb')
                for line in out:
                    f.write(line)
            f.close()

            self.fail(msg, ret)
        elif ret:
            self.fail(describe(ret), ret)

    def tearDown(self):
        """Tasks to perform after run()."""
        for entry in self._daemonpids:
            killdaemons(entry)

            assert '=' in key, ('extra config opt %s must '
                                'have an = for assignment' % opt)
            hgrc.write('[%s]\n%s\n' % (section, key))
        hgrc.close()

    def fail(self, msg, ret):
        if (not ret and self._interactive and
            os.path.exists(self.errpath)):
            iolock.acquire()
            print 'Accept this change? [n] ',
            answer = sys.stdin.readline().strip()
            iolock.release()
            if answer.lower() in ('y', 'yes'):
                if self.name.endswith('.t'):
                    rename(self.errpath, self.path)
                else:
                    rename(self.errpath, '%s.out' % self.path)

                return '.', self.name, ''

        # unittest differentiates between errored and failed.
        # Failed is denoted by AssertionError (by default at least).
        raise AssertionError(msg)

class PythonTest(Test):

            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

            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,
                       shell=self.options.shell)

21523:9fb6f328576a

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

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

        path is the full path to the file defining the test.

        keeptmpdir determines whether to keep the test's temporary directory
        after execution. It defaults to removal (False).

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

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

        startport controls the starting port number to use for this test. Each

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

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

            if not missing:
                missing = ['irrelevant']

            if failed:
                self.fail('hg have failed checking for %s' % failed[-1])
            else:
                self._skipped = True
                raise SkipTest(missing[-1])
        elif ret == 'timeout':
            self.fail('timed out')
        elif ret is False:
            raise WarnTest('no result code from test')
        elif out != self._refout:
            # The result object handles diff calculation for us.
            self._result.addOutputMismatch(self, ret, out, self._refout)

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

                f = open(self.errpath, 'wb')
                for line in out:
                    f.write(line)
            f.close()

            self.fail(msg)
        elif ret:
            self.fail(describe(ret))

    def tearDown(self):
        """Tasks to perform after run()."""
        for entry in self._daemonpids:
            killdaemons(entry)

            assert '=' in key, ('extra config opt %s must '
                                'have an = for assignment' % opt)
            hgrc.write('[%s]\n%s\n' % (section, key))
        hgrc.close()

    def fail(self, msg):
        # unittest differentiates between errored and failed.
        # Failed is denoted by AssertionError (by default at least).
        raise AssertionError(msg)

class PythonTest(Test):

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

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

        if self._options.nodiff:
            return

            else:
                self.stream.write('\n')
                for line in lines:
                    self.stream.write(line)
                self.stream.flush()

        if ret or not self._options.interactive or \
            not os.path.exists(test.errpath):
            return

        iolock.acquire()
        print 'Accept this change? [n] ',
        answer = sys.stdin.readline().strip()
        iolock.release()
        if answer.lower() in ('y', 'yes'):
            if test.name.endswith('.t'):
                rename(test.errpath, test.path)
            else:
                rename(test.errpath, '%s.out' % test.path)

    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,
                       timeout=self.options.timeout,
                       startport=self.options.port + count * 3,
                       extraconfigopts=self.options.extra_config_opt,
                       py3kwarnings=self.options.py3k_warnings,
                       shell=self.options.shell)