e1f0cf2c527f49a0e53ed02c459e174e0b597c50
[babeltrace.git] / 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
This page took 0.032713 seconds and 3 git commands to generate.