comparison tests/run-tests.py @ 21514:59fe123dbb00

run-tests: refactor port number declaration

Instead of making port numbers derived from count and a global initial
port, we now pass a start port to Test.__init__ and do the calculation
at a higher level.

21513:acfd19f3e79c

    """

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

    def __init__(self, options, path, count, tmpdir, abort, keeptmpdir=False,
                 debug=False, nodiff=False, diffviewer=None,
                 interactive=False, timeout=defaults['timeout']):
        """Create a test from parameters.

        options are parsed command line options that control test execution.

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

        count is an identifier used to denote this test instance.

        tmpdir is the main temporary directory to use for this test.

        abort is a flag that turns to True if test execution should be aborted.
        It is consulted periodically during the execution of tests.

        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.path = path
        self.name = os.path.basename(path)
        self._testdir = os.path.dirname(path)
        self.errpath = os.path.join(self._testdir, '%s.err' % self.name)

        self._options = options
        self._count = count
        self._threadtmp = tmpdir
        self._abort = abort
        self._keeptmpdir = keeptmpdir
        self._debug = debug
        self._nodiff = nodiff
        self._diffviewer = diffviewer
        self._interactive = interactive
        self._timeout = timeout
        self._daemonpids = []

        self._finished = None
        self._ret = None
        self._out = None

    def runTest(self):
        """Run this test instance.

        This will return a tuple describing the result of the test.
        """
        replacements, port = self._getreplacements()
        env = self._getenv(port)
        self._daemonpids.append(env['DAEMON_PIDS'])
        self._createhgrc(env['HGRCPATH'])

        vlog('# Test', self.name)

    def _run(self, replacements, env):
        # This should be implemented in child classes to run tests.
        raise SkipTest('unknown test type')

    def _getreplacements(self):
        port = self._options.port + self._count * 3
        r = [
            (r':%s\b' % port, ':$HGPORT'),
            (r':%s\b' % (port + 1), ':$HGPORT1'),
            (r':%s\b' % (port + 2), ':$HGPORT2'),
            ]

        if os.name == 'nt':
            r.append(
                (''.join(c.isalpha() and '[%s%s]' % (c.lower(), c.upper()) or
                    c in '/\\' and r'[/\\]' or c.isdigit() and c or '\\' + c
                    for c in self._testtmp), '$TESTTMP'))
        else:
            r.append((re.escape(self._testtmp), '$TESTTMP'))

        return r, port

    def _getenv(self, port):
        env = os.environ.copy()
        env['TESTTMP'] = self._testtmp
        env['HOME'] = self._testtmp
        env["HGPORT"] = str(port)
        env["HGPORT1"] = str(port + 1)
        env["HGPORT2"] = str(port + 2)
        env["HGRCPATH"] = os.path.join(self._threadtmp, '.hgrc')
        env["DAEMON_PIDS"] = os.path.join(self._threadtmp, 'daemon.pids')
        env["HGEDITOR"] = sys.executable + ' -c "import sys; sys.exit(0)"'
        env["HGMERGE"] = "internal:merge"
        env["HGUSER"]   = "test"

                break

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

        return testcls(self.options, refpath, count, tmpdir, self.abort,
                       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)

    def _cleanup(self):
        """Clean up state from this test invocation."""

        if self.options.keep_tmpdir:

21514:59fe123dbb00

    """

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

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

        options are parsed command line options that control test execution.

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

        tmpdir is the main temporary directory to use for this test.

        abort is a flag that turns to True if test execution should be aborted.
        It is consulted periodically during the execution of tests.

        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
        test will reserve 3 port numbers for execution. It is the caller's
        responsibility to allocate a non-overlapping port range to Test
        instances.
        """

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

        self._options = options
        self._threadtmp = tmpdir
        self._abort = abort
        self._keeptmpdir = keeptmpdir
        self._debug = debug
        self._nodiff = nodiff
        self._diffviewer = diffviewer
        self._interactive = interactive
        self._timeout = timeout
        self._startport = startport
        self._daemonpids = []

        self._finished = None
        self._ret = None
        self._out = None

    def runTest(self):
        """Run this test instance.

        This will return a tuple describing the result of the test.
        """
        replacements = self._getreplacements()
        env = self._getenv()
        self._daemonpids.append(env['DAEMON_PIDS'])
        self._createhgrc(env['HGRCPATH'])

        vlog('# Test', self.name)

    def _run(self, replacements, env):
        # This should be implemented in child classes to run tests.
        raise SkipTest('unknown test type')

    def _getreplacements(self):
        r = [
            (r':%s\b' % self._startport, ':$HGPORT'),
            (r':%s\b' % (self._startport + 1), ':$HGPORT1'),
            (r':%s\b' % (self._startport + 2), ':$HGPORT2'),
            ]

        if os.name == 'nt':
            r.append(
                (''.join(c.isalpha() and '[%s%s]' % (c.lower(), c.upper()) or
                    c in '/\\' and r'[/\\]' or c.isdigit() and c or '\\' + c
                    for c in self._testtmp), '$TESTTMP'))
        else:
            r.append((re.escape(self._testtmp), '$TESTTMP'))

        return r

    def _getenv(self):
        env = os.environ.copy()
        env['TESTTMP'] = self._testtmp
        env['HOME'] = self._testtmp
        env["HGPORT"] = str(self._startport)
        env["HGPORT1"] = str(self._startport + 1)
        env["HGPORT2"] = str(self._startport + 2)
        env["HGRCPATH"] = os.path.join(self._threadtmp, '.hgrc')
        env["DAEMON_PIDS"] = os.path.join(self._threadtmp, 'daemon.pids')
        env["HGEDITOR"] = sys.executable + ' -c "import sys; sys.exit(0)"'
        env["HGMERGE"] = "internal:merge"
        env["HGUSER"]   = "test"

                break

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

        return testcls(self.options, refpath, tmpdir, self.abort,
                       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)

    def _cleanup(self):
        """Clean up state from this test invocation."""

        if self.options.keep_tmpdir: