hg-stable: comparison tests/run-tests.py

equal deleted inserted replaced

-:25d5a9ecbb85
+:fa6ba4372678
 """
 # Status code reserved for skipped tests (used by hghave).
 SKIPPED_STATUS = 80
-def __init__(self, path, tmpdir, abort, keeptmpdir=False,
+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.
 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.
 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.
 self.name = os.path.basename(path)
 self._testdir = os.path.dirname(path)
 self.errpath = os.path.join(self._testdir, '%s.err' % self.name)
 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._extraconfigopts = extraconfigopts or []
 self._py3kwarnings = py3kwarnings
 self._shell = shell
+self._aborted = False
 self._daemonpids = []
 self._finished = None
 self._ret = None
 self._out = None
 self._skipped = None
 self._testtmp = None
 if os.path.exists(self.errpath):
 os.remove(self.errpath)
 def run(self, result):
 result.startTest(self)
-interrupted = False
 try:
 try:
 self.setUp()
 except (KeyboardInterrupt, SystemExit):
-interrupted = True
+self._aborted = True
 raise
 except Exception:
 result.addError(self, sys.exc_info())
 return
 success = False
 try:
 self.runTest()
 except KeyboardInterrupt:
-interrupted = True
+self._aborted = True
 raise
 except SkipTest, e:
 result.addSkip(self, str(e))
 except IgnoreTest, e:
 result.addIgnore(self, str(e))
 success = True
 try:
 self.tearDown()
 except (KeyboardInterrupt, SystemExit):
-interrupted = True
+self._aborted = True
 raise
 except Exception:
 result.addError(self, sys.exc_info())
 success = False
 if success:
 result.addSuccess(self)
 finally:
-result.stopTest(self, interrupted=interrupted)
+result.stopTest(self, interrupted=self._aborted)
 def runTest(self):
 """Run this test instance.
 This will return a tuple describing the result of the test.
 def _run(self, replacements, env):
 # This should be implemented in child classes to run tests.
 raise SkipTest('unknown test type')
+def abort(self):
+"""Terminate execution of this test."""
+self._aborted = True
 def _getreplacements(self):
 r = [
 (r':%s\b' % self._startport, ':$HGPORT'),
 (r':%s\b' % (self._startport + 1), ':$HGPORT1'),
 (r':%s\b' % (self._startport + 2), ':$HGPORT2'),
 py3kswitch = self._py3kwarnings and ' -3' or ''
 cmd = '%s%s "%s"' % (PYTHON, py3kswitch, self.path)
 vlog("# Running", cmd)
 if os.name == 'nt':
 replacements.append((r'\r\n', '\n'))
-return run(cmd, self._testtmp, replacements, env, self._abort,
+result = run(cmd, self._testtmp, replacements, env,
 debug=self._debug, timeout=self._timeout)
+if self._aborted:
+raise KeyboardInterrupt()
+return result
 class TTest(Test):
 """A "t test" is a test backed by a .t file."""
 SKIPPED_PREFIX = 'skipped: '
 cmd = '%s "%s"' % (self._shell, fname)
 vlog("# Running", cmd)
 exitcode, output = run(cmd, self._testtmp, replacements, env,
-self._abort, debug=self._debug,
+debug=self._debug, timeout=self._timeout)
-timeout=self._timeout)
+if self._aborted:
+raise KeyboardInterrupt()
 # Do not merge output if skipped. Return hghave message instead.
 # Similarly, with --debug, output is None.
 if exitcode == self.SKIPPED_STATUS or output is None:
 return exitcode, output
 def _stringescape(s):
 return TTest.ESCAPESUB(TTest._escapef, s)
 wifexited = getattr(os, "WIFEXITED", lambda x: False)
-def run(cmd, wd, replacements, env, abort, debug=False, timeout=None):
+def run(cmd, wd, replacements, env, debug=False, timeout=None):
 """Run command in a sub-process, capturing the output (stdout and stderr).
 Return a tuple (exitcode, output).  output is None in debug mode."""
 if debug:
 proc = subprocess.Popen(cmd, shell=True, cwd=wd, env=env)
 ret = proc.wait()
 if proc.timeout:
 ret = 'timeout'
 if ret:
 killdaemons(env['DAEMON_PIDS'])
-if abort[0]:
-raise KeyboardInterrupt()
 for s, r in replacements:
 output = re.sub(s, r, output)
 return ret, output.splitlines(True)
 if ignored:
 continue
 tests.append(test)
+runtests = list(tests)
 jobs = self._runner.options.jobs
 done = queue.Queue()
 running = 0
 def job(test, result):
 t = threading.Thread(target=job, name=test.name,
 args=(test, result))
 t.start()
 running += 1
 except KeyboardInterrupt:
-self._runner.abort[0] = True
+for test in runtests:
+test.abort()
 return result
 class TextTestRunner(unittest.TextTestRunner):
 """Custom unittest test runner that uses appropriate settings."""
 self.inst = None
 self.bindir = None
 self.tmpbinddir = None
 self.pythondir = None
 self.coveragefile = None
-self.abort = [False]
 self._createdfiles = []
 self._hgpath = None
 def run(self, args, parser=None):
 """Run the test suite."""
 break
 refpath = os.path.join(self.testdir, test)
 tmpdir = os.path.join(self.hgtmp, 'child%d' % count)
-return testcls(refpath, tmpdir, self.abort,
+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,

changeset 21520	fa6ba4372678
parent 21518	8e8049b9bda4
child 21521	855f092c96e5