@@ -538,41 +538,44 @@ def outputcoverage(options):
covrun('-i', '-b', '"--directory=%s"' % htmldir, '"--omit=%s"' % omit)
if options.annotate:
adir = os.path.join(TESTDIR, 'annotated')
if not os.path.isdir(adir):
os.mkdir(adir)
covrun('-i', '-a', '"--directory=%s"' % adir, '"--omit=%s"' % omit)
class Test(object):
- """Encapsulates a single, runnable test."""
+ """Encapsulates a single, runnable test.
+
+ Test instances can be run multiple times via run(). However, multiple
+ runs cannot be run concurrently.
+ """
def __init__(self, path, options, count):
self._path = path
self._options = options
+ self._count = count
self.threadtmp = os.path.join(HGTMP, 'child%d' % count)
os.mkdir(self.threadtmp)
- self._testtmp = os.path.join(self.threadtmp, os.path.basename(path))
- os.mkdir(self._testtmp)
-
- self._setreplacements(count)
-
def run(self, result, refpath):
- env = self._getenv()
+ testtmp = os.path.join(self.threadtmp, os.path.basename(self._path))
+ os.mkdir(testtmp)
+ replacements, port = self._getreplacements(testtmp)
+ env = self._getenv(testtmp, port)
createhgrc(env['HGRCPATH'], self._options)
starttime = time.time()
def updateduration():
result.duration = time.time() - starttime
try:
- ret, out = self._run(self._replacements, env)
+ ret, out = self._run(testtmp, replacements, env)
updateduration()
result.ret = ret
result.out = out
except KeyboardInterrupt:
updateduration()
result.interrupted = True
except Exception, e:
updateduration()
@@ -586,45 +589,47 @@ class Test(object):
result.refout = None # to match "out is None"
elif os.path.exists(refpath):
f = open(refpath, 'r')
result.refout = f.read().splitlines(True)
f.close()
else:
result.refout = []
- def _run(self, replacements, env):
+ if not self._options.keep_tmpdir:
+ shutil.rmtree(testtmp)
+
+ def _run(self, testtmp, replacements, env):
raise NotImplemented('Subclasses must implement Test.run()')
- def _setreplacements(self, count):
- port = self._options.port + count * 3
+ def _getreplacements(self, testtmp):
+ 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'))
+ for c in testtmp), '$TESTTMP'))
else:
- r.append((re.escape(self._testtmp), '$TESTTMP'))
+ r.append((re.escape(testtmp), '$TESTTMP'))
- self._replacements = r
- self._port = port
+ return r, port
- def _getenv(self):
+ def _getenv(self, testtmp, port):
env = os.environ.copy()
- env['TESTTMP'] = self._testtmp
- env['HOME'] = self._testtmp
- env["HGPORT"] = str(self._port)
- env["HGPORT1"] = str(self._port + 1)
- env["HGPORT2"] = str(self._port + 2)
+ env['TESTTMP'] = testtmp
+ env['HOME'] = 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"
env["HGENCODING"] = "ascii"
env["HGENCODINGMODE"] = "strict"
@@ -669,18 +674,18 @@ def pytest(test, wd, options, replacemen
cmd = '%s%s "%s"' % (PYTHON, py3kswitch, test)
vlog("# Running", cmd)
if os.name == 'nt':
replacements.append((r'\r\n', '\n'))
return run(cmd, wd, options, replacements, env)
class PythonTest(Test):
"""A Python-based test."""
- def _run(self, replacements, env):
- return pytest(self._path, self._testtmp, self._options, replacements,
+ def _run(self, testtmp, replacements, env):
+ return pytest(self._path, testtmp, self._options, replacements,
env)
needescape = re.compile(r'[\x00-\x08\x0b-\x1f\x7f-\xff]').search
escapesub = re.compile(r'[\x00-\x08\x0b-\x1f\\\x7f-\xff]').sub
escapemap = dict((chr(i), r'\x%02x' % i) for i in range(256))
escapemap.update({'\\': '\\\\', '\r': r'\r'})
def escapef(m):
return escapemap[m.group(0)]
@@ -932,18 +937,18 @@ def tsttest(test, wd, options, replaceme
if warnonly == 2:
exitcode = False # set exitcode to warned
return exitcode, postout
class TTest(Test):
"""A "t test" is a test backed by a .t file."""
- def _run(self, replacements, env):
- return tsttest(self._path, self._testtmp, self._options, replacements,
+ def _run(self, testtmp, replacements, env):
+ return tsttest(self._path, testtmp, self._options, replacements,
env)
wifexited = getattr(os, "WIFEXITED", lambda x: False)
def run(cmd, wd, options, replacements, env):
"""Run command in a sub-process, capturing the output (stdout and stderr).
Return a tuple (exitcode, output). output is None in debug mode."""
# TODO: Use subprocess.Popen if we're running on Python 2.4
if options.debug: