summaryrefslogtreecommitdiffstats
path: root/build-aux/tap-driver.py
blob: c231caecf838148744e03db7bcbb417863f78b68 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
#!/usr/bin/env python3
# Adapted from tappy copyright (c) 2016, Matt Layman
# MIT license
# https://github.com/python-tap/tappy

import io
import re
import subprocess
import sys


class Directive(object):
    """A representation of a result line directive."""

    skip_pattern = re.compile(
        r"""^SKIP\S*
            (?P<whitespace>\s*) # Optional whitespace.
            (?P<reason>.*)      # Slurp up the rest.""",
        re.IGNORECASE | re.VERBOSE)
    todo_pattern = re.compile(
        r"""^TODO\b             # The directive name
            (?P<whitespace>\s*) # Immediately following must be whitespace.
            (?P<reason>.*)      # Slurp up the rest.""",
        re.IGNORECASE | re.VERBOSE)

    def __init__(self, text):
        """Initialize the directive by parsing the text.
        The text is assumed to be everything after a '#\s*' on a result line.
        """
        self._text = text
        self._skip = False
        self._todo = False
        self._reason = None

        match = self.skip_pattern.match(text)
        if match:
            self._skip = True
            self._reason = match.group('reason')

        match = self.todo_pattern.match(text)
        if match:
            if match.group('whitespace'):
                self._todo = True
            else:
                # Catch the case where the directive has no descriptive text.
                if match.group('reason') == '':
                    self._todo = True
            self._reason = match.group('reason')

    @property
    def text(self):
        """Get the entire text."""
        return self._text

    @property
    def skip(self):
        """Check if the directive is a SKIP type."""
        return self._skip

    @property
    def todo(self):
        """Check if the directive is a TODO type."""
        return self._todo

    @property
    def reason(self):
        """Get the reason for the directive."""
        return self._reason


class Parser(object):
    """A parser for TAP files and lines."""

    # ok and not ok share most of the same characteristics.
    result_base = r"""
        \s*                    # Optional whitespace.
        (?P<number>\d*)        # Optional test number.
        \s*                    # Optional whitespace.
        (?P<description>[^#]*) # Optional description before #.
        \#?                    # Optional directive marker.
        \s*                    # Optional whitespace.
        (?P<directive>.*)      # Optional directive text.
    """
    ok = re.compile(r'^ok' + result_base, re.VERBOSE)
    not_ok = re.compile(r'^not\ ok' + result_base, re.VERBOSE)
    plan = re.compile(r"""
        ^1..(?P<expected>\d+) # Match the plan details.
        [^#]*                 # Consume any non-hash character to confirm only
                              # directives appear with the plan details.
        \#?                   # Optional directive marker.
        \s*                   # Optional whitespace.
        (?P<directive>.*)     # Optional directive text.
    """, re.VERBOSE)
    diagnostic = re.compile(r'^#')
    bail = re.compile(r"""
        ^Bail\ out!
        \s*            # Optional whitespace.
        (?P<reason>.*) # Optional reason.
    """, re.VERBOSE)
    version = re.compile(r'^TAP version (?P<version>\d+)$')

    TAP_MINIMUM_DECLARED_VERSION = 13

    def parse(self, fh):
        """Generate tap.line.Line objects, given a file-like object `fh`.
        `fh` may be any object that implements both the iterator and
        context management protocol (i.e. it can be used in both a
        "with" statement and a "for...in" statement.)
        Trailing whitespace and newline characters will be automatically
        stripped from the input lines.
        """
        with fh:
            for line in fh:
                yield self.parse_line(line.rstrip())

    def parse_line(self, text):
        """Parse a line into whatever TAP category it belongs."""
        match = self.ok.match(text)
        if match:
            return self._parse_result(True, match)

        match = self.not_ok.match(text)
        if match:
            return self._parse_result(False, match)

        if self.diagnostic.match(text):
            return ('diagnostic', text)

        match = self.plan.match(text)
        if match:
            return self._parse_plan(match)

        match = self.bail.match(text)
        if match:
            return ('bail', match.group('reason'))

        match = self.version.match(text)
        if match:
            return self._parse_version(match)

        return ('unknown',)

    def _parse_plan(self, match):
        """Parse a matching plan line."""
        expected_tests = int(match.group('expected'))
        directive = Directive(match.group('directive'))

        # Only SKIP directives are allowed in the plan.
        if directive.text and not directive.skip:
            return ('unknown',)

        return ('plan', expected_tests, directive)

    def _parse_result(self, ok, match):
        """Parse a matching result line into a result instance."""
        return ('result', ok, match.group('number'),
            match.group('description').strip(),
            Directive(match.group('directive')))

    def _parse_version(self, match):
        version = int(match.group('version'))
        if version < self.TAP_MINIMUM_DECLARED_VERSION:
            raise ValueError('It is an error to explicitly specify '
                             'any version lower than 13.')
        return ('version', version)


class Rules(object):

    def __init__(self):
        self._lines_seen = {'plan': [], 'test': 0, 'failed': 0, 'version': []}
        self._errors = []

    def check(self, final_line_count):
        """Check the status of all provided data and update the suite."""
        if self._lines_seen['version']:
            self._process_version_lines()
        self._process_plan_lines(final_line_count)

    def check_errors(self):
        if self._lines_seen['failed'] > 0:
            self._add_error('Tests failed.')
        if self._errors:
            for error in self._errors:
                print(error)
            return 1
        return 0

    def _process_version_lines(self):
        """Process version line rules."""
        if len(self._lines_seen['version']) > 1:
            self._add_error('Multiple version lines appeared.')
        elif self._lines_seen['version'][0] != 1:
            self._add_error('The version must be on the first line.')

    def _process_plan_lines(self, final_line_count):
        """Process plan line rules."""
        if not self._lines_seen['plan']:
            self._add_error('Missing a plan.')
            return

        if len(self._lines_seen['plan']) > 1:
            self._add_error('Only one plan line is permitted per file.')
            return

        expected_tests, at_line = self._lines_seen['plan'][0]
        if not self._plan_on_valid_line(at_line, final_line_count):
            self._add_error(
                'A plan must appear at the beginning or end of the file.')
            return

        if expected_tests != self._lines_seen['test']:
            self._add_error(
                'Expected {expected_count} tests '
                'but only {seen_count} ran.'.format(
                    expected_count=expected_tests,
                    seen_count=self._lines_seen['test']))

    def _plan_on_valid_line(self, at_line, final_line_count):
        """Check if a plan is on a valid line."""
        # Put the common cases first.
        if at_line == 1 or at_line == final_line_count:
            return True

        # The plan may only appear on line 2 if the version is at line 1.
        after_version = (
            self._lines_seen['version'] and
            self._lines_seen['version'][0] == 1 and
            at_line == 2)
        if after_version:
            return True

        return False

    def handle_bail(self, reason):
        """Handle a bail line."""
        self._add_error('Bailed: {reason}').format(reason=reason)

    def handle_skipping_plan(self):
        """Handle a plan that contains a SKIP directive."""
        sys.exit(77)

    def saw_plan(self, expected_tests, at_line):
        """Record when a plan line was seen."""
        self._lines_seen['plan'].append((expected_tests, at_line))

    def saw_test(self, ok):
        """Record when a test line was seen."""
        self._lines_seen['test'] += 1
        if not ok:
            self._lines_seen['failed'] += 1

    def saw_version_at(self, line_counter):
        """Record when a version line was seen."""
        self._lines_seen['version'].append(line_counter)

    def _add_error(self, message):
        self._errors += [message]


if __name__ == '__main__':
    parser = Parser()
    rules = Rules()

    try:
        out = subprocess.check_output(sys.argv[1:], universal_newlines=True)
    except subprocess.CalledProcessError as e:
        sys.stdout.write(e.output)
        raise e

    line_generator = parser.parse(io.StringIO(out))
    line_counter = 0
    for line in line_generator:
        line_counter += 1

        if line[0] == 'unknown':
            continue

        if line[0] == 'result':
            rules.saw_test(line[1])
            print('{okay} {num} {description} {directive}'.format(
                okay=('' if line[1] else 'not ') + 'ok', num=line[2],
                description=line[3], directive=line[4].text))
        elif line[0] == 'plan':
            if line[2].skip:
                rules.handle_skipping_plan()
            rules.saw_plan(line[1], line_counter)
        elif line[0] == 'bail':
            rules.handle_bail(line[1])
        elif line[0] == 'version':
            rules.saw_version_at(line_counter)
        elif line[0] == 'diagnostic':
            print(line[1])

    rules.check(line_counter)
    sys.exit(rules.check_errors())