This is an archive of the discontinued Mercurial Phabricator instance.

Differential D3701

run-tests: extract onStart and onEnd into the test result
ClosedPublic

Authored by lothiraldan on Jun 7 2018, 3:19 PM.

Details

Reviewers
durin42
Group Reviewers
hg-reviewers
Commits
rHG948691ea92a9: run-tests: extract onStart and onEnd into the test result
Summary

It would allow custom test result to display custom messages.

Diff Detail

Repository
rHG Mercurial
Lint
Lint Skipped
Unit
Unit Tests Skipped

Event Timeline

lothiraldan created this revision.Jun 7 2018, 3:19 PM
lothiraldan updated this revision to Diff 9032.Jun 12 2018, 5:05 PM
durin42 added a subscriber: durin42.Jul 3 2018, 2:00 PM
durin42 added inline comments.
tests/run-tests.py
1717

omit superfluous pass

lothiraldan updated this revision to Diff 9417.Jul 3 2018, 2:28 PM
lothiraldan marked an inline comment as done.Jul 3 2018, 2:28 PM
durin42 accepted this revision as: durin42.Jul 9 2018, 10:23 AM
durin42 accepted this revision.Jul 9 2018, 10:44 AM
This revision is now accepted and ready to land.Jul 9 2018, 10:44 AM
Closed by commit rHG948691ea92a9: run-tests: extract onStart and onEnd into the test result (authored by lothiraldan). · Explain WhyJul 9 2018, 10:45 AM
This revision was automatically updated to reflect the committed changes.

Revision Contents
Changeset List

PathPackages
Mtests/run-tests.py (61 lines)

Diff 8993

tests/run-tests.py

if options.color == 'auto': if options.color == 'auto':
self.color = pygmentspresent and self.stream.isatty() self.color = pygmentspresent and self.stream.isatty()
elif options.color == 'never': elif options.color == 'never':
self.color = False self.color = False
else: # 'always', for testing purposes else: # 'always', for testing purposes
self.color = pygmentspresent self.color = pygmentspresent
def onStart(self, test):
""" Can be overriden by custom TestResult
"""
pass
durin42Unsubmitted
Done

omit superfluous pass

durin42: omit superfluous pass
def addFailure(self, test, reason): def addFailure(self, test, reason):
self.failures.append((test, reason)) self.failures.append((test, reason))
if self._options.first: if self._options.first:
self.stop() self.stop()
else: else:
with iolock: with iolock:
if reason == "timed out": if reason == "timed out":
class TextTestRunner(unittest.TextTestRunner): class TextTestRunner(unittest.TextTestRunner):
"""Custom unittest test runner that uses appropriate settings.""" """Custom unittest test runner that uses appropriate settings."""
def __init__(self, runner, *args, **kwargs): def __init__(self, runner, *args, **kwargs):
super(TextTestRunner, self).__init__(*args, **kwargs) super(TextTestRunner, self).__init__(*args, **kwargs)
self._runner = runner self._runner = runner
self._result = getTestResult()(self._runner.options, self.stream,
self.descriptions, 0)
def listtests(self, test): def listtests(self, test):
result = getTestResult()(self._runner.options, self.stream,
self.descriptions, 0)
test = sorted(test, key=lambda t: t.name) test = sorted(test, key=lambda t: t.name)
self._result.onStart(test)
for t in test: for t in test:
print(t.name) print(t.name)
result.addSuccess(t) self._result.addSuccess(t)
if self._runner.options.xunit: if self._runner.options.xunit:
with open(self._runner.options.xunit, "wb") as xuf: with open(self._runner.options.xunit, "wb") as xuf:
self._writexunit(result, xuf) self._writexunit(self._result, xuf)
if self._runner.options.json: if self._runner.options.json:
jsonpath = os.path.join(self._runner._outputdir, b'report.json') jsonpath = os.path.join(self._runner._outputdir, b'report.json')
with open(jsonpath, 'w') as fp: with open(jsonpath, 'w') as fp:
self._writejson(result, fp) self._writejson(self._result, fp)
return result return self._result
def run(self, test): def run(self, test):
result = getTestResult()(self._runner.options, self.stream, self._result.onStart(test)
self.descriptions, self.verbosity) test(self._result)
test(result)
failed = len(result.failures) failed = len(self._result.failures)
skipped = len(result.skipped) skipped = len(self._result.skipped)
ignored = len(result.ignored) ignored = len(self._result.ignored)
with iolock: with iolock:
self.stream.writeln('') self.stream.writeln('')
if not self._runner.options.noskips: if not self._runner.options.noskips:
for test, msg in result.skipped: for test, msg in self._result.skipped:
formatted = 'Skipped %s: %s\n' % (test.name, msg) formatted = 'Skipped %s: %s\n' % (test.name, msg)
self.stream.write(highlightmsg(formatted, result.color)) self.stream.write(highlightmsg(formatted, self._result.color))
for test, msg in result.failures: for test, msg in self._result.failures:
formatted = 'Failed %s: %s\n' % (test.name, msg) formatted = 'Failed %s: %s\n' % (test.name, msg)
self.stream.write(highlightmsg(formatted, result.color)) self.stream.write(highlightmsg(formatted, self._result.color))
for test, msg in result.errors: for test, msg in self._result.errors:
self.stream.writeln('Errored %s: %s' % (test.name, msg)) self.stream.writeln('Errored %s: %s' % (test.name, msg))
if self._runner.options.xunit: if self._runner.options.xunit:
with open(self._runner.options.xunit, "wb") as xuf: with open(self._runner.options.xunit, "wb") as xuf:
self._writexunit(result, xuf) self._writexunit(self._result, xuf)
if self._runner.options.json: if self._runner.options.json:
jsonpath = os.path.join(self._runner._outputdir, b'report.json') jsonpath = os.path.join(self._runner._outputdir, b'report.json')
with open(jsonpath, 'w') as fp: with open(jsonpath, 'w') as fp:
self._writejson(result, fp) self._writejson(self._result, fp)
self._runner._checkhglib('Tested') self._runner._checkhglib('Tested')
savetimes(self._runner._outputdir, result) savetimes(self._runner._outputdir, self._result)
if failed and self._runner.options.known_good_rev: if failed and self._runner.options.known_good_rev:
self._bisecttests(t for t, m in result.failures) self._bisecttests(t for t, m in self._result.failures)
self.stream.writeln( self.stream.writeln(
'# Ran %d tests, %d skipped, %d failed.' '# Ran %d tests, %d skipped, %d failed.'
% (result.testsRun, skipped + ignored, failed)) % (self._result.testsRun, skipped + ignored, failed))
if failed: if failed:
self.stream.writeln('python hash seed: %s' % self.stream.writeln('python hash seed: %s' %
os.environ['PYTHONHASHSEED']) os.environ['PYTHONHASHSEED'])
if self._runner.options.time: if self._runner.options.time:
self.printtimes(result.times) self.printtimes(self._result.times)
if self._runner.options.exceptions: if self._runner.options.exceptions:
exceptions = aggregateexceptions( exceptions = aggregateexceptions(
os.path.join(self._runner._outputdir, b'exceptions')) os.path.join(self._runner._outputdir, b'exceptions'))
self.stream.writeln('Exceptions Report:') self.stream.writeln('Exceptions Report:')
self.stream.writeln('%d total from %d frames' % self.stream.writeln('%d total from %d frames' %
(exceptions['total'], (exceptions['total'],
len(exceptions['exceptioncounts']))) len(exceptions['exceptioncounts'])))
combined = exceptions['combined'] combined = exceptions['combined']
for key in sorted(combined, key=combined.get, reverse=True): for key in sorted(combined, key=combined.get, reverse=True):
frame, line, exc = key frame, line, exc = key
totalcount, testcount, leastcount, leasttest = combined[key] totalcount, testcount, leastcount, leasttest = combined[key]
self.stream.writeln('%d (%d tests)\t%s: %s (%s - %d total)' self.stream.writeln('%d (%d tests)\t%s: %s (%s - %d total)'
% (totalcount, % (totalcount,
testcount, testcount,
frame, exc, frame, exc,
leasttest, leastcount)) leasttest, leastcount))
self.stream.flush() self.stream.flush()
return result return self._result
def _bisecttests(self, tests): def _bisecttests(self, tests):
bisectcmd = ['hg', 'bisect'] bisectcmd = ['hg', 'bisect']
bisectrepo = self._runner.options.bisect_repo bisectrepo = self._runner.options.bisect_repo
if bisectrepo: if bisectrepo:
bisectcmd.extend(['-R', os.path.abspath(bisectrepo)]) bisectcmd.extend(['-R', os.path.abspath(bisectrepo)])
def pread(args): def pread(args):
env = os.environ.copy() env = os.environ.copy()
assert self._installdir assert self._installdir
self._installchg() self._installchg()
result = runner.run(suite) result = runner.run(suite)
if result.failures: if result.failures:
failed = True failed = True
result.onEnd()
if self.options.anycoverage: if self.options.anycoverage:
self._outputcoverage() self._outputcoverage()
except KeyboardInterrupt: except KeyboardInterrupt:
failed = True failed = True
print("\ninterrupted!") print("\ninterrupted!")
if failed: if failed:
return 1 return 1