| # Copyright 2015, Google Inc. |
| # All rights reserved. |
| # |
| # Redistribution and use in source and binary forms, with or without |
| # modification, are permitted provided that the following conditions are |
| # met: |
| # |
| # * Redistributions of source code must retain the above copyright |
| # notice, this list of conditions and the following disclaimer. |
| # * Redistributions in binary form must reproduce the above |
| # copyright notice, this list of conditions and the following disclaimer |
| # in the documentation and/or other materials provided with the |
| # distribution. |
| # * Neither the name of Google Inc. nor the names of its |
| # contributors may be used to endorse or promote products derived from |
| # this software without specific prior written permission. |
| # |
| # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
| # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
| # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
| # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
| # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
| # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
| # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
| # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
| # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
| # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
| # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
| |
| from __future__ import absolute_import |
| |
| import collections |
| import itertools |
| import traceback |
| import unittest |
| from xml.etree import ElementTree |
| |
| import coverage |
| from six import moves |
| |
| from tests import _loader |
| |
| |
| class CaseResult( |
| collections.namedtuple('CaseResult', [ |
| 'id', 'name', 'kind', 'stdout', 'stderr', 'skip_reason', 'traceback' |
| ])): |
| """A serializable result of a single test case. |
| |
| Attributes: |
| id (object): Any serializable object used to denote the identity of this |
| test case. |
| name (str or None): A human-readable name of the test case. |
| kind (CaseResult.Kind): The kind of test result. |
| stdout (object or None): Output on stdout, or None if nothing was captured. |
| stderr (object or None): Output on stderr, or None if nothing was captured. |
| skip_reason (object or None): The reason the test was skipped. Must be |
| something if self.kind is CaseResult.Kind.SKIP, else None. |
| traceback (object or None): The traceback of the test. Must be something if |
| self.kind is CaseResult.Kind.{ERROR, FAILURE, EXPECTED_FAILURE}, else |
| None. |
| """ |
| |
| class Kind: |
| UNTESTED = 'untested' |
| RUNNING = 'running' |
| ERROR = 'error' |
| FAILURE = 'failure' |
| SUCCESS = 'success' |
| SKIP = 'skip' |
| EXPECTED_FAILURE = 'expected failure' |
| UNEXPECTED_SUCCESS = 'unexpected success' |
| |
| def __new__(cls, |
| id=None, |
| name=None, |
| kind=None, |
| stdout=None, |
| stderr=None, |
| skip_reason=None, |
| traceback=None): |
| """Helper keyword constructor for the namedtuple. |
| |
| See this class' attributes for information on the arguments.""" |
| assert id is not None |
| assert name is None or isinstance(name, str) |
| if kind is CaseResult.Kind.UNTESTED: |
| pass |
| elif kind is CaseResult.Kind.RUNNING: |
| pass |
| elif kind is CaseResult.Kind.ERROR: |
| assert traceback is not None |
| elif kind is CaseResult.Kind.FAILURE: |
| assert traceback is not None |
| elif kind is CaseResult.Kind.SUCCESS: |
| pass |
| elif kind is CaseResult.Kind.SKIP: |
| assert skip_reason is not None |
| elif kind is CaseResult.Kind.EXPECTED_FAILURE: |
| assert traceback is not None |
| elif kind is CaseResult.Kind.UNEXPECTED_SUCCESS: |
| pass |
| else: |
| assert False |
| return super(cls, CaseResult).__new__(cls, id, name, kind, stdout, |
| stderr, skip_reason, traceback) |
| |
| def updated(self, |
| name=None, |
| kind=None, |
| stdout=None, |
| stderr=None, |
| skip_reason=None, |
| traceback=None): |
| """Get a new validated CaseResult with the fields updated. |
| |
| See this class' attributes for information on the arguments.""" |
| name = self.name if name is None else name |
| kind = self.kind if kind is None else kind |
| stdout = self.stdout if stdout is None else stdout |
| stderr = self.stderr if stderr is None else stderr |
| skip_reason = self.skip_reason if skip_reason is None else skip_reason |
| traceback = self.traceback if traceback is None else traceback |
| return CaseResult( |
| id=self.id, |
| name=name, |
| kind=kind, |
| stdout=stdout, |
| stderr=stderr, |
| skip_reason=skip_reason, |
| traceback=traceback) |
| |
| |
| class AugmentedResult(unittest.TestResult): |
| """unittest.Result that keeps track of additional information. |
| |
| Uses CaseResult objects to store test-case results, providing additional |
| information beyond that of the standard Python unittest library, such as |
| standard output. |
| |
| Attributes: |
| id_map (callable): A unary callable mapping unittest.TestCase objects to |
| unique identifiers. |
| cases (dict): A dictionary mapping from the identifiers returned by id_map |
| to CaseResult objects corresponding to those IDs. |
| """ |
| |
| def __init__(self, id_map): |
| """Initialize the object with an identifier mapping. |
| |
| Arguments: |
| id_map (callable): Corresponds to the attribute `id_map`.""" |
| super(AugmentedResult, self).__init__() |
| self.id_map = id_map |
| self.cases = None |
| |
| def startTestRun(self): |
| """See unittest.TestResult.startTestRun.""" |
| super(AugmentedResult, self).startTestRun() |
| self.cases = dict() |
| |
| def stopTestRun(self): |
| """See unittest.TestResult.stopTestRun.""" |
| super(AugmentedResult, self).stopTestRun() |
| |
| def startTest(self, test): |
| """See unittest.TestResult.startTest.""" |
| super(AugmentedResult, self).startTest(test) |
| case_id = self.id_map(test) |
| self.cases[case_id] = CaseResult( |
| id=case_id, name=test.id(), kind=CaseResult.Kind.RUNNING) |
| |
| def addError(self, test, error): |
| """See unittest.TestResult.addError.""" |
| super(AugmentedResult, self).addError(test, error) |
| case_id = self.id_map(test) |
| self.cases[case_id] = self.cases[case_id].updated( |
| kind=CaseResult.Kind.ERROR, traceback=error) |
| |
| def addFailure(self, test, error): |
| """See unittest.TestResult.addFailure.""" |
| super(AugmentedResult, self).addFailure(test, error) |
| case_id = self.id_map(test) |
| self.cases[case_id] = self.cases[case_id].updated( |
| kind=CaseResult.Kind.FAILURE, traceback=error) |
| |
| def addSuccess(self, test): |
| """See unittest.TestResult.addSuccess.""" |
| super(AugmentedResult, self).addSuccess(test) |
| case_id = self.id_map(test) |
| self.cases[case_id] = self.cases[case_id].updated( |
| kind=CaseResult.Kind.SUCCESS) |
| |
| def addSkip(self, test, reason): |
| """See unittest.TestResult.addSkip.""" |
| super(AugmentedResult, self).addSkip(test, reason) |
| case_id = self.id_map(test) |
| self.cases[case_id] = self.cases[case_id].updated( |
| kind=CaseResult.Kind.SKIP, skip_reason=reason) |
| |
| def addExpectedFailure(self, test, error): |
| """See unittest.TestResult.addExpectedFailure.""" |
| super(AugmentedResult, self).addExpectedFailure(test, error) |
| case_id = self.id_map(test) |
| self.cases[case_id] = self.cases[case_id].updated( |
| kind=CaseResult.Kind.EXPECTED_FAILURE, traceback=error) |
| |
| def addUnexpectedSuccess(self, test): |
| """See unittest.TestResult.addUnexpectedSuccess.""" |
| super(AugmentedResult, self).addUnexpectedSuccess(test) |
| case_id = self.id_map(test) |
| self.cases[case_id] = self.cases[case_id].updated( |
| kind=CaseResult.Kind.UNEXPECTED_SUCCESS) |
| |
| def set_output(self, test, stdout, stderr): |
| """Set the output attributes for the CaseResult corresponding to a test. |
| |
| Args: |
| test (unittest.TestCase): The TestCase to set the outputs of. |
| stdout (str): Output from stdout to assign to self.id_map(test). |
| stderr (str): Output from stderr to assign to self.id_map(test). |
| """ |
| case_id = self.id_map(test) |
| self.cases[case_id] = self.cases[case_id].updated( |
| stdout=stdout.decode(), stderr=stderr.decode()) |
| |
| def augmented_results(self, filter): |
| """Convenience method to retrieve filtered case results. |
| |
| Args: |
| filter (callable): A unary predicate to filter over CaseResult objects. |
| """ |
| return (self.cases[case_id] for case_id in self.cases |
| if filter(self.cases[case_id])) |
| |
| |
| class CoverageResult(AugmentedResult): |
| """Extension to AugmentedResult adding coverage.py support per test.\ |
| |
| Attributes: |
| coverage_context (coverage.Coverage): coverage.py management object. |
| """ |
| |
| def __init__(self, id_map): |
| """See AugmentedResult.__init__.""" |
| super(CoverageResult, self).__init__(id_map=id_map) |
| self.coverage_context = None |
| |
| def startTest(self, test): |
| """See unittest.TestResult.startTest. |
| |
| Additionally initializes and begins code coverage tracking.""" |
| super(CoverageResult, self).startTest(test) |
| self.coverage_context = coverage.Coverage(data_suffix=True) |
| self.coverage_context.start() |
| |
| def stopTest(self, test): |
| """See unittest.TestResult.stopTest. |
| |
| Additionally stops and deinitializes code coverage tracking.""" |
| super(CoverageResult, self).stopTest(test) |
| self.coverage_context.stop() |
| self.coverage_context.save() |
| self.coverage_context = None |
| |
| def stopTestRun(self): |
| """See unittest.TestResult.stopTestRun.""" |
| super(CoverageResult, self).stopTestRun() |
| # TODO(atash): Dig deeper into why the following line fails to properly |
| # combine coverage data from the Cython plugin. |
| #coverage.Coverage().combine() |
| |
| |
| class _Colors: |
| """Namespaced constants for terminal color magic numbers.""" |
| HEADER = '\033[95m' |
| INFO = '\033[94m' |
| OK = '\033[92m' |
| WARN = '\033[93m' |
| FAIL = '\033[91m' |
| BOLD = '\033[1m' |
| UNDERLINE = '\033[4m' |
| END = '\033[0m' |
| |
| |
| class TerminalResult(CoverageResult): |
| """Extension to CoverageResult adding basic terminal reporting.""" |
| |
| def __init__(self, out, id_map): |
| """Initialize the result object. |
| |
| Args: |
| out (file-like): Output file to which terminal-colored live results will |
| be written. |
| id_map (callable): See AugmentedResult.__init__. |
| """ |
| super(TerminalResult, self).__init__(id_map=id_map) |
| self.out = out |
| |
| def startTestRun(self): |
| """See unittest.TestResult.startTestRun.""" |
| super(TerminalResult, self).startTestRun() |
| self.out.write(_Colors.HEADER + 'Testing gRPC Python...\n' + |
| _Colors.END) |
| |
| def stopTestRun(self): |
| """See unittest.TestResult.stopTestRun.""" |
| super(TerminalResult, self).stopTestRun() |
| self.out.write(summary(self)) |
| self.out.flush() |
| |
| def addError(self, test, error): |
| """See unittest.TestResult.addError.""" |
| super(TerminalResult, self).addError(test, error) |
| self.out.write(_Colors.FAIL + 'ERROR {}\n'.format(test.id()) + |
| _Colors.END) |
| self.out.flush() |
| |
| def addFailure(self, test, error): |
| """See unittest.TestResult.addFailure.""" |
| super(TerminalResult, self).addFailure(test, error) |
| self.out.write(_Colors.FAIL + 'FAILURE {}\n'.format(test.id()) + |
| _Colors.END) |
| self.out.flush() |
| |
| def addSuccess(self, test): |
| """See unittest.TestResult.addSuccess.""" |
| super(TerminalResult, self).addSuccess(test) |
| self.out.write(_Colors.OK + 'SUCCESS {}\n'.format(test.id()) + |
| _Colors.END) |
| self.out.flush() |
| |
| def addSkip(self, test, reason): |
| """See unittest.TestResult.addSkip.""" |
| super(TerminalResult, self).addSkip(test, reason) |
| self.out.write(_Colors.INFO + 'SKIP {}\n'.format(test.id()) + |
| _Colors.END) |
| self.out.flush() |
| |
| def addExpectedFailure(self, test, error): |
| """See unittest.TestResult.addExpectedFailure.""" |
| super(TerminalResult, self).addExpectedFailure(test, error) |
| self.out.write(_Colors.INFO + 'FAILURE_OK {}\n'.format(test.id()) + |
| _Colors.END) |
| self.out.flush() |
| |
| def addUnexpectedSuccess(self, test): |
| """See unittest.TestResult.addUnexpectedSuccess.""" |
| super(TerminalResult, self).addUnexpectedSuccess(test) |
| self.out.write(_Colors.INFO + 'UNEXPECTED_OK {}\n'.format(test.id()) + |
| _Colors.END) |
| self.out.flush() |
| |
| |
| def _traceback_string(type, value, trace): |
| """Generate a descriptive string of a Python exception traceback. |
| |
| Args: |
| type (class): The type of the exception. |
| value (Exception): The value of the exception. |
| trace (traceback): Traceback of the exception. |
| |
| Returns: |
| str: Formatted exception descriptive string. |
| """ |
| buffer = moves.cStringIO() |
| traceback.print_exception(type, value, trace, file=buffer) |
| return buffer.getvalue() |
| |
| |
| def summary(result): |
| """A summary string of a result object. |
| |
| Args: |
| result (AugmentedResult): The result object to get the summary of. |
| |
| Returns: |
| str: The summary string. |
| """ |
| assert isinstance(result, AugmentedResult) |
| untested = list( |
| result.augmented_results( |
| lambda case_result: case_result.kind is CaseResult.Kind.UNTESTED)) |
| running = list( |
| result.augmented_results( |
| lambda case_result: case_result.kind is CaseResult.Kind.RUNNING)) |
| failures = list( |
| result.augmented_results( |
| lambda case_result: case_result.kind is CaseResult.Kind.FAILURE)) |
| errors = list( |
| result.augmented_results( |
| lambda case_result: case_result.kind is CaseResult.Kind.ERROR)) |
| successes = list( |
| result.augmented_results( |
| lambda case_result: case_result.kind is CaseResult.Kind.SUCCESS)) |
| skips = list( |
| result.augmented_results( |
| lambda case_result: case_result.kind is CaseResult.Kind.SKIP)) |
| expected_failures = list( |
| result.augmented_results( |
| lambda case_result: case_result.kind is CaseResult.Kind.EXPECTED_FAILURE |
| )) |
| unexpected_successes = list( |
| result.augmented_results( |
| lambda case_result: case_result.kind is CaseResult.Kind.UNEXPECTED_SUCCESS |
| )) |
| running_names = [case.name for case in running] |
| finished_count = (len(failures) + len(errors) + len(successes) + |
| len(expected_failures) + len(unexpected_successes)) |
| statistics = ('{finished} tests finished:\n' |
| '\t{successful} successful\n' |
| '\t{unsuccessful} unsuccessful\n' |
| '\t{skipped} skipped\n' |
| '\t{expected_fail} expected failures\n' |
| '\t{unexpected_successful} unexpected successes\n' |
| 'Interrupted Tests:\n' |
| '\t{interrupted}\n'.format( |
| finished=finished_count, |
| successful=len(successes), |
| unsuccessful=(len(failures) + len(errors)), |
| skipped=len(skips), |
| expected_fail=len(expected_failures), |
| unexpected_successful=len(unexpected_successes), |
| interrupted=str(running_names))) |
| tracebacks = '\n\n'.join( |
| [(_Colors.FAIL + '{test_name}' + _Colors.END + '\n' + _Colors.BOLD + |
| 'traceback:' + _Colors.END + '\n' + '{traceback}\n' + _Colors.BOLD + |
| 'stdout:' + _Colors.END + '\n' + '{stdout}\n' + _Colors.BOLD + |
| 'stderr:' + _Colors.END + '\n' + '{stderr}\n').format( |
| test_name=result.name, |
| traceback=_traceback_string(*result.traceback), |
| stdout=result.stdout, |
| stderr=result.stderr) |
| for result in itertools.chain(failures, errors)]) |
| notes = 'Unexpected successes: {}\n'.format( |
| [result.name for result in unexpected_successes]) |
| return statistics + '\nErrors/Failures: \n' + tracebacks + '\n' + notes |
| |
| |
| def jenkins_junit_xml(result): |
| """An XML tree object that when written is recognizable by Jenkins. |
| |
| Args: |
| result (AugmentedResult): The result object to get the junit xml output of. |
| |
| Returns: |
| ElementTree.ElementTree: The XML tree. |
| """ |
| assert isinstance(result, AugmentedResult) |
| root = ElementTree.Element('testsuites') |
| suite = ElementTree.SubElement(root, 'testsuite', { |
| 'name': 'Python gRPC tests', |
| }) |
| for case in result.cases.values(): |
| if case.kind is CaseResult.Kind.SUCCESS: |
| ElementTree.SubElement(suite, 'testcase', { |
| 'name': case.name, |
| }) |
| elif case.kind in (CaseResult.Kind.ERROR, CaseResult.Kind.FAILURE): |
| case_xml = ElementTree.SubElement(suite, 'testcase', { |
| 'name': case.name, |
| }) |
| error_xml = ElementTree.SubElement(case_xml, 'error', {}) |
| error_xml.text = ''.format(case.stderr, case.traceback) |
| return ElementTree.ElementTree(element=root) |
