| # Copyright 2015 gRPC authors. |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| from __future__ import absolute_import |
| |
| import collections |
| import io |
| import itertools |
| import traceback |
| import unittest |
| from xml.etree import ElementTree |
| |
| import coverage |
| |
| 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(object): |
| 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 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, err): |
| """See unittest.TestResult.addError.""" |
| super(AugmentedResult, self).addError(test, err) |
| case_id = self.id_map(test) |
| self.cases[case_id] = self.cases[case_id].updated( |
| kind=CaseResult.Kind.ERROR, traceback=err |
| ) |
| |
| def addFailure(self, test, err): |
| """See unittest.TestResult.addFailure.""" |
| super(AugmentedResult, self).addFailure(test, err) |
| case_id = self.id_map(test) |
| self.cases[case_id] = self.cases[case_id].updated( |
| kind=CaseResult.Kind.FAILURE, traceback=err |
| ) |
| |
| 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, err): |
| """See unittest.TestResult.addExpectedFailure.""" |
| super(AugmentedResult, self).addExpectedFailure(test, err) |
| case_id = self.id_map(test) |
| self.cases[case_id] = self.cases[case_id].updated( |
| kind=CaseResult.Kind.EXPECTED_FAILURE, traceback=err |
| ) |
| |
| 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 |
| |
| |
| class _Colors(object): |
| """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, err): |
| """See unittest.TestResult.addError.""" |
| super(TerminalResult, self).addError(test, err) |
| self.out.write( |
| _Colors.FAIL + "ERROR {}\n".format(test.id()) + _Colors.END |
| ) |
| self.out.flush() |
| |
| def addFailure(self, test, err): |
| """See unittest.TestResult.addFailure.""" |
| super(TerminalResult, self).addFailure(test, err) |
| 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, err): |
| """See unittest.TestResult.addExpectedFailure.""" |
| super(TerminalResult, self).addExpectedFailure(test, err) |
| 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 = io.StringIO() |
| 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) |
