1#  subunit: extensions to python unittest to get test results from subprocesses.
2#  Copyright (C) 2009  Robert Collins <robertc@robertcollins.net>
3#
4#  Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
5#  license at the users choice. A copy of both licenses are available in the
6#  project source as Apache-2.0 and BSD. You may not use this file except in
7#  compliance with one of these two licences.
8#
9#  Unless required by applicable law or agreed to in writing, software
10#  distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT
11#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
12#  license you chose for the specific language governing permissions and
13#  limitations under that license.
14#
15
16
17from optparse import OptionParser
18import sys
19
20from extras import safe_hasattr
21from testtools import CopyStreamResult, StreamResult, StreamResultRouter
22
23from subunit import (
24    DiscardStream, ProtocolTestCase, ByteStreamToStreamResult,
25    StreamResultToBytes,
26    )
27from subunit.test_results import CatFiles
28
29
30def make_options(description):
31    parser = OptionParser(description=description)
32    parser.add_option(
33        "--no-passthrough", action="store_true",
34        help="Hide all non subunit input.", default=False,
35        dest="no_passthrough")
36    parser.add_option(
37        "-o", "--output-to",
38        help="Send the output to this path rather than stdout.")
39    parser.add_option(
40        "-f", "--forward", action="store_true", default=False,
41        help="Forward subunit stream on stdout. When set, received "
42            "non-subunit output will be encapsulated in subunit.")
43    return parser
44
45
46def run_tests_from_stream(input_stream, result, passthrough_stream=None,
47    forward_stream=None, protocol_version=1, passthrough_subunit=True):
48    """Run tests from a subunit input stream through 'result'.
49
50    Non-test events - top level file attachments - are expected to be
51    dropped by v2 StreamResults at the present time (as all the analysis code
52    is in ExtendedTestResult API's), so to implement passthrough_stream they
53    are diverted and copied directly when that is set.
54
55    :param input_stream: A stream containing subunit input.
56    :param result: A TestResult that will receive the test events.
57        NB: This should be an ExtendedTestResult for v1 and a StreamResult for
58        v2.
59    :param passthrough_stream: All non-subunit input received will be
60        sent to this stream.  If not provided, uses the ``TestProtocolServer``
61        default, which is ``sys.stdout``.
62    :param forward_stream: All subunit input received will be forwarded
63        to this stream. If not provided, uses the ``TestProtocolServer``
64        default, which is to not forward any input. Do not set this when
65        transforming the stream - items would be double-reported.
66    :param protocol_version: What version of the subunit protocol to expect.
67    :param passthrough_subunit: If True, passthrough should be as subunit
68        otherwise unwrap it. Only has effect when forward_stream is None.
69        (when forwarding as subunit non-subunit input is always turned into
70        subunit)
71    """
72    if 1==protocol_version:
73        test = ProtocolTestCase(
74            input_stream, passthrough=passthrough_stream,
75            forward=forward_stream)
76    elif 2==protocol_version:
77        # In all cases we encapsulate unknown inputs.
78        if forward_stream is not None:
79            # Send events to forward_stream as subunit.
80            forward_result = StreamResultToBytes(forward_stream)
81            # If we're passing non-subunit through, copy:
82            if passthrough_stream is None:
83                # Not passing non-test events - split them off to nothing.
84                router = StreamResultRouter(forward_result)
85                router.add_rule(StreamResult(), 'test_id', test_id=None)
86                result = CopyStreamResult([router, result])
87            else:
88                # otherwise, copy all events to forward_result
89                result = CopyStreamResult([forward_result, result])
90        elif passthrough_stream is not None:
91            if not passthrough_subunit:
92                # Route non-test events to passthrough_stream, unwrapping them for
93                # display.
94                passthrough_result = CatFiles(passthrough_stream)
95            else:
96                passthrough_result = StreamResultToBytes(passthrough_stream)
97            result = StreamResultRouter(result)
98            result.add_rule(passthrough_result, 'test_id', test_id=None)
99        test = ByteStreamToStreamResult(input_stream,
100            non_subunit_name='stdout')
101    else:
102        raise Exception("Unknown protocol version.")
103    result.startTestRun()
104    test.run(result)
105    result.stopTestRun()
106
107
108def filter_by_result(result_factory, output_path, passthrough, forward,
109                     input_stream=sys.stdin, protocol_version=1,
110                     passthrough_subunit=True):
111    """Filter an input stream using a test result.
112
113    :param result_factory: A callable that when passed an output stream
114        returns a TestResult.  It is expected that this result will output
115        to the given stream.
116    :param output_path: A path send output to.  If None, output will be go
117        to ``sys.stdout``.
118    :param passthrough: If True, all non-subunit input will be sent to
119        ``sys.stdout``.  If False, that input will be discarded.
120    :param forward: If True, all subunit input will be forwarded directly to
121        ``sys.stdout`` as well as to the ``TestResult``.
122    :param input_stream: The source of subunit input.  Defaults to
123        ``sys.stdin``.
124    :param protocol_version: The subunit protocol version to expect.
125    :param passthrough_subunit: If True, passthrough should be as subunit.
126    :return: A test result with the results of the run.
127    """
128    if passthrough:
129        passthrough_stream = sys.stdout
130    else:
131        if 1==protocol_version:
132            passthrough_stream = DiscardStream()
133        else:
134            passthrough_stream = None
135
136    if forward:
137        forward_stream = sys.stdout
138    elif 1==protocol_version:
139        forward_stream = DiscardStream()
140    else:
141        forward_stream = None
142
143    if output_path is None:
144        output_to = sys.stdout
145    else:
146        output_to = file(output_path, 'wb')
147
148    try:
149        result = result_factory(output_to)
150        run_tests_from_stream(
151            input_stream, result, passthrough_stream, forward_stream,
152            protocol_version=protocol_version,
153            passthrough_subunit=passthrough_subunit)
154    finally:
155        if output_path:
156            output_to.close()
157    return result
158
159
160def run_filter_script(result_factory, description, post_run_hook=None,
161    protocol_version=1, passthrough_subunit=True):
162    """Main function for simple subunit filter scripts.
163
164    Many subunit filter scripts take a stream of subunit input and use a
165    TestResult to handle the events generated by that stream.  This function
166    wraps a lot of the boiler-plate around that by making a script with
167    options for handling passthrough information and stream forwarding, and
168    that will exit with a successful return code (i.e. 0) if the input stream
169    represents a successful test run.
170
171    :param result_factory: A callable that takes an output stream and returns
172        a test result that outputs to that stream.
173    :param description: A description of the filter script.
174    :param protocol_version: What protocol version to consume/emit.
175    :param passthrough_subunit: If True, passthrough should be as subunit.
176    """
177    parser = make_options(description)
178    (options, args) = parser.parse_args()
179    result = filter_by_result(
180        result_factory, options.output_to, not options.no_passthrough,
181        options.forward, protocol_version=protocol_version,
182        passthrough_subunit=passthrough_subunit,
183        input_stream=find_stream(sys.stdin, args))
184    if post_run_hook:
185        post_run_hook(result)
186    if not safe_hasattr(result, 'wasSuccessful'):
187        result = result.decorated
188    if result.wasSuccessful():
189        sys.exit(0)
190    else:
191        sys.exit(1)
192
193
194def find_stream(stdin, argv):
195    """Find a stream to use as input for filters.
196
197    :param stdin: Standard in - used if no files are named in argv.
198    :param argv: Command line arguments after option parsing. If one file
199        is named, that is opened in read only binary mode and returned.
200        A missing file will raise an exception, as will multiple file names.
201    """
202    assert len(argv) < 2, "Too many filenames."
203    if argv:
204        return open(argv[0], 'rb')
205    else:
206        return stdin
207