tests/utils/python/tap/tracker.py

   1 # Copyright (c) 2016, Matt Layman
   2
   3 from __future__ import print_function
   4 import os
   5 import string
   6 import sys
   7
   8 from tap.directive import Directive
   9 from tap.i18n import _
  10 from tap.line import Result
  11
  12
  13 class Tracker(object):
  14
  15     def __init__(
  16             self, outdir=None, combined=False, streaming=False, stream=None,
  17             header=True):
  18         self.outdir = outdir
  19
  20         # Combine all the test results into one file.
  21         self.combined = combined
  22         self.combined_line_number = 0
  23         # Test case ordering is important for the combined results
  24         # because of how numbers are assigned. The test cases
  25         # must be tracked in order so that reporting can sequence
  26         # the line numbers properly.
  27         self.combined_test_cases_seen = []
  28
  29         # Stream output directly to a stream instead of file output.
  30         self.streaming = streaming
  31         self.stream = stream
  32
  33         # Display the test case header unless told not to.
  34         self.header = header
  35
  36         # Internal state for tracking each test case.
  37         self._test_cases = {}
  38
  39         # Python versions 2 and 3 keep maketrans in different locations.
  40         if sys.version_info[0] < 3:
  41             self._sanitized_table = string.maketrans(' \\/\n', '----')
  42         else:  # pragma: no cover
  43             self._sanitized_table = str.maketrans(' \\/\n', '----')
  44
  45     def _get_outdir(self):
  46         return self._outdir
  47
  48     def _set_outdir(self, outdir):
  49         self._outdir = outdir
  50         if outdir and not os.path.exists(outdir):
  51             os.makedirs(outdir)
  52
  53     outdir = property(_get_outdir, _set_outdir)
  54
  55     def _track(self, class_name):
  56         """Keep track of which test cases have executed."""
  57         if self._test_cases.get(class_name) is None:
  58             if self.streaming and self.header:
  59                 self._write_test_case_header(class_name, self.stream)
  60
  61             self._test_cases[class_name] = []
  62             if self.combined:
  63                 self.combined_test_cases_seen.append(class_name)
  64
  65     def add_ok(self, class_name, description, directive=''):
  66         result = Result(
  67             ok=True, number=self._get_next_line_number(class_name),
  68             description=description, directive=Directive(directive))
  69         self._add_line(class_name, result)
  70
  71     def add_not_ok(
  72             self, class_name, description, directive='', diagnostics=None):
  73         result = Result(
  74             ok=False, number=self._get_next_line_number(class_name),
  75             description=description, diagnostics=diagnostics,
  76             directive=Directive(directive))
  77         self._add_line(class_name, result)
  78
  79     def add_skip(self, class_name, description, reason):
  80         directive = 'SKIP {0}'.format(reason)
  81         result = Result(
  82             ok=True, number=self._get_next_line_number(class_name),
  83             description=description, directive=Directive(directive))
  84         self._add_line(class_name, result)
  85
  86     def _add_line(self, class_name, result):
  87         self._track(class_name)
  88         if self.streaming:
  89             print(result, file=self.stream)
  90         self._test_cases[class_name].append(result)
  91
  92     def _get_next_line_number(self, class_name):
  93         if self.combined or self.streaming:
  94             # This has an obvious side effect. Oh well.
  95             self.combined_line_number += 1
  96             return self.combined_line_number
  97         else:
  98             try:
  99                 return len(self._test_cases[class_name]) + 1
 100             except KeyError:
 101                 # A result is created before the call to _track so the test
 102                 # case may not be tracked yet. In that case, the line is 1.
 103                 return 1
 104
 105     def generate_tap_reports(self):
 106         """Generate TAP reports.
 107
 108         The results are either combined into a single output file or
 109         the output file name is generated from the test case.
 110         """
 111         if self.streaming:
 112             # The results already went to the stream, record the plan.
 113             print('1..{0}'.format(self.combined_line_number), file=self.stream)
 114             return
 115
 116         if self.combined:
 117             combined_file = 'testresults.tap'
 118             if self.outdir:
 119                 combined_file = os.path.join(self.outdir, combined_file)
 120             with open(combined_file, 'w') as out_file:
 121                 for test_case in self.combined_test_cases_seen:
 122                     self.generate_tap_report(
 123                         test_case, self._test_cases[test_case], out_file)
 124                 print(
 125                     '1..{0}'.format(self.combined_line_number), file=out_file)
 126         else:
 127             for test_case, tap_lines in self._test_cases.items():
 128                 with open(self._get_tap_file_path(test_case), 'w') as out_file:
 129                     self.generate_tap_report(test_case, tap_lines, out_file)
 130
 131     def generate_tap_report(self, test_case, tap_lines, out_file):
 132         self._write_test_case_header(test_case, out_file)
 133
 134         for tap_line in tap_lines:
 135             print(tap_line, file=out_file)
 136
 137         # For combined results, the plan is only output once after
 138         # all the test cases complete.
 139         if not self.combined:
 140             print('1..{0}'.format(len(tap_lines)), file=out_file)
 141
 142     def _write_test_case_header(self, test_case, stream):
 143         print(_('# TAP results for {test_case}').format(
 144             test_case=test_case), file=stream)
 145
 146     def _get_tap_file_path(self, test_case):
 147         """Get the TAP output file path for the test case."""
 148         sanitized_test_case = test_case.translate(self._sanitized_table)
 149         tap_file = sanitized_test_case + '.tap'
 150         if self.outdir:
 151             return os.path.join(self.outdir, tap_file)
 152         return tap_file