comparison tests/run-tests.py @ 39253:c496e8c14b9e

tests: add support for emitting trace events to run-tests

Right now this is pretty basic, but it's a start.

Differential Revision: https://phab.mercurial-scm.org/D4343

39252:9a81f126f9fa

import sysconfig
import tempfile
import threading
import time
import unittest
import xml.dom.minidom as minidom

try:
    import Queue as queue
except ImportError:

        env["HGUSER"]   = "test"
        env["HGENCODING"] = "ascii"
        env["HGENCODINGMODE"] = "strict"
        env["HGHOSTNAME"] = "test-hostname"
        env['HGIPV6'] = str(int(self._useipv6))

        extraextensions = []
        for opt in self._extraconfigopts:
            section, key = opt.encode('utf-8').split(b'.', 1)
            if section != 'extensions':

        def addsalt(line, inpython):
            if inpython:
                script.append(b'%s %d 0\n' % (salt, line))
            else:
                script.append(b'echo %s %d $?\n' % (salt, line))

        script = []

        # After we run the shell script, we re-unify the script output
        # with non-active parts of the source, with synchronization by our

            script.append(b'set -x\n')
        if self._hgcommand != b'hg':
            script.append(b'alias hg="%s"\n' % self._hgcommand)
        if os.getenv('MSYSTEM'):
            script.append(b'alias pwd="pwd -W"\n')
        if self._case:
            casestr = b'#'.join(self._case)
            if isinstance(self._case, str):
                quoted = shellquote(casestr)
            else:

                    inpython = False
                after.setdefault(pos, []).append(l)
                prepos = pos
                pos = n
                addsalt(n, False)
                cmd = l[4:].split()
                if len(cmd) == 2 and cmd[0] == b'cd':
                    l = b'  $ cd %s || exit 1\n' % cmd[1]
                script.append(l[4:])
            elif l.startswith(b'  > '): # continuations
                after.setdefault(prepos, []).append(l)
                script.append(l[4:])
            elif l.startswith(b'  '): # results
                # Queue up a list of expected results.

        if inpython:
            script.append(b'EOF\n')
        if skipping is not None:
            after.setdefault(pos, []).append('  !!! missing #endif\n')
        addsalt(n + 1, False)

        return salt, script, after, expected

    def _processoutput(self, exitcode, output, salt, after, expected):
        # Merge the script output back into a unified test.
        warnonly = 1 # 1: not yet; 2: yes; 3: for sure not

39253:c496e8c14b9e

import sysconfig
import tempfile
import threading
import time
import unittest
import uuid
import xml.dom.minidom as minidom

try:
    import Queue as queue
except ImportError:

        env["HGUSER"]   = "test"
        env["HGENCODING"] = "ascii"
        env["HGENCODINGMODE"] = "strict"
        env["HGHOSTNAME"] = "test-hostname"
        env['HGIPV6'] = str(int(self._useipv6))
        if 'HGCATAPULTSERVERPIPE' not in env:
            env['HGCATAPULTSERVERPIPE'] = '/dev/null'

        extraextensions = []
        for opt in self._extraconfigopts:
            section, key = opt.encode('utf-8').split(b'.', 1)
            if section != 'extensions':

        def addsalt(line, inpython):
            if inpython:
                script.append(b'%s %d 0\n' % (salt, line))
            else:
                script.append(b'echo %s %d $?\n' % (salt, line))
        active = []
        session = str(uuid.uuid4())
        if PYTHON3:
            session = session.encode('ascii')
        def toggletrace(cmd):
            quoted = shellquote(cmd.strip()).replace(b'\\', b'\\\\')
            if active:
                script.append(
                    b'echo END %s %s >> "$HGCATAPULTSERVERPIPE"\n' % (
                        session, active[0]))
            script.append(
                b'echo START %s %s >> "$HGCATAPULTSERVERPIPE"\n' % (
                    session, quoted))
            active[0:] = [quoted]

        script = []

        # After we run the shell script, we re-unify the script output
        # with non-active parts of the source, with synchronization by our

            script.append(b'set -x\n')
        if self._hgcommand != b'hg':
            script.append(b'alias hg="%s"\n' % self._hgcommand)
        if os.getenv('MSYSTEM'):
            script.append(b'alias pwd="pwd -W"\n')

        if os.getenv('HGCATAPULTSERVERPIPE'):
            # Kludge: use a while loop to keep the pipe from getting
            # closed by our echo commands. The still-running file gets
            # reaped at the end of the script, which causes the while
            # loop to exit and closes the pipe. Sigh.
            script.append(
                b'rtendtracing() {\n'
                b'  echo END %(session)s %(name)s >> $HGCATAPULTSERVERPIPE\n'
                b'  rm -f "$TESTTMP/.still-running"\n'
                b'}\n'
                b'trap "rtendtracing" 0\n'
                b'touch "$TESTTMP/.still-running"\n'
                b'while [ -f "$TESTTMP/.still-running" ]; do sleep 1; done '
                b'> $HGCATAPULTSERVERPIPE &\n'
                b'HGCATAPULTSESSION=%(session)s ; export HGCATAPULTSESSION\n'
                b'echo START %(session)s %(name)s >> $HGCATAPULTSERVERPIPE\n'
                % {
                    'name': self.name,
                    'session': session,
                }
            )

        if self._case:
            casestr = b'#'.join(self._case)
            if isinstance(self._case, str):
                quoted = shellquote(casestr)
            else:

                    inpython = False
                after.setdefault(pos, []).append(l)
                prepos = pos
                pos = n
                addsalt(n, False)
                rawcmd = l[4:]
                cmd = rawcmd.split()
                toggletrace(rawcmd)
                if len(cmd) == 2 and cmd[0] == b'cd':
                    l = b'  $ cd %s || exit 1\n' % cmd[1]
                script.append(rawcmd)
            elif l.startswith(b'  > '): # continuations
                after.setdefault(prepos, []).append(l)
                script.append(l[4:])
            elif l.startswith(b'  '): # results
                # Queue up a list of expected results.

        if inpython:
            script.append(b'EOF\n')
        if skipping is not None:
            after.setdefault(pos, []).append('  !!! missing #endif\n')
        addsalt(n + 1, False)
        return salt, script, after, expected

    def _processoutput(self, exitcode, output, salt, after, expected):
        # Merge the script output back into a unified test.
        warnonly = 1 # 1: not yet; 2: yes; 3: for sure not