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
|
# subunit: extensions to python unittest to get test results from subprocesses.
# Copyright (C) 2009 Robert Collins <robertc@robertcollins.net>
#
# Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
# license at the users choice. A copy of both licenses are available in the
# project source as Apache-2.0 and BSD. You may not use this file except in
# compliance with one of these two licences.
#
# Unless required by applicable law or agreed to in writing, software
# distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# license you chose for the specific language governing permissions and
# limitations under that license.
#
from optparse import OptionParser
import sys
from subunit import DiscardStream, ProtocolTestCase
def make_options(description):
parser = OptionParser(description=description)
parser.add_option(
"--no-passthrough", action="store_true",
help="Hide all non subunit input.", default=False,
dest="no_passthrough")
parser.add_option(
"-o", "--output-to",
help="Send the output to this path rather than stdout.")
parser.add_option(
"-f", "--forward", action="store_true", default=False,
help="Forward subunit stream on stdout.")
return parser
def run_tests_from_stream(input_stream, result, passthrough_stream=None,
forward_stream=None):
"""Run tests from a subunit input stream through 'result'.
:param input_stream: A stream containing subunit input.
:param result: A TestResult that will receive the test events.
:param passthrough_stream: All non-subunit input received will be
sent to this stream. If not provided, uses the ``TestProtocolServer``
default, which is ``sys.stdout``.
:param forward_stream: All subunit input received will be forwarded
to this stream. If not provided, uses the ``TestProtocolServer``
default, which is to not forward any input.
"""
test = ProtocolTestCase(
input_stream, passthrough=passthrough_stream,
forward=forward_stream)
result.startTestRun()
test.run(result)
result.stopTestRun()
def filter_by_result(result_factory, output_path, passthrough, forward,
input_stream=sys.stdin):
"""Filter an input stream using a test result.
:param result_factory: A callable that when passed an output stream
returns a TestResult. It is expected that this result will output
to the given stream.
:param output_path: A path send output to. If None, output will be go
to ``sys.stdout``.
:param passthrough: If True, all non-subunit input will be sent to
``sys.stdout``. If False, that input will be discarded.
:param forward: If True, all subunit input will be forwarded directly to
``sys.stdout`` as well as to the ``TestResult``.
:param input_stream: The source of subunit input. Defaults to
``sys.stdin``.
:return: A test result with the resultts of the run.
"""
if passthrough:
passthrough_stream = sys.stdout
else:
passthrough_stream = DiscardStream()
if forward:
forward_stream = sys.stdout
else:
forward_stream = DiscardStream()
if output_path is None:
output_to = sys.stdout
else:
output_to = file(output_path, 'wb')
try:
result = result_factory(output_to)
run_tests_from_stream(
input_stream, result, passthrough_stream, forward_stream)
finally:
if output_path:
output_to.close()
return result
def run_filter_script(result_factory, description, post_run_hook=None):
"""Main function for simple subunit filter scripts.
Many subunit filter scripts take a stream of subunit input and use a
TestResult to handle the events generated by that stream. This function
wraps a lot of the boiler-plate around that by making a script with
options for handling passthrough information and stream forwarding, and
that will exit with a successful return code (i.e. 0) if the input stream
represents a successful test run.
:param result_factory: A callable that takes an output stream and returns
a test result that outputs to that stream.
:param description: A description of the filter script.
"""
parser = make_options(description)
(options, args) = parser.parse_args()
result = filter_by_result(
result_factory, options.output_to, not options.no_passthrough,
options.forward)
if post_run_hook:
post_run_hook(result)
if result.wasSuccessful():
sys.exit(0)
else:
sys.exit(1)
|
