git.samba.org - third_party/subunit/blob - python/subunit/filters.py

   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
  17 from optparse import OptionParser
  18 import sys
  19
  20 from subunit import DiscardStream, ProtocolTestCase
  21
  22
  23 def make_options(description):
  24     parser = OptionParser(description=description)
  25     parser.add_option(
  26         "--no-passthrough", action="store_true",
  27         help="Hide all non subunit input.", default=False,
  28         dest="no_passthrough")
  29     parser.add_option(
  30         "-o", "--output-to",
  31         help="Send the output to this path rather than stdout.")
  32     parser.add_option(
  33         "-f", "--forward", action="store_true", default=False,
  34         help="Forward subunit stream on stdout.")
  35     return parser
  36
  37
  38 def run_tests_from_stream(input_stream, result, passthrough_stream=None,
  39                           forward_stream=None):
  40     """Run tests from a subunit input stream through 'result'.
  41
  42     :param input_stream: A stream containing subunit input.
  43     :param result: A TestResult that will receive the test events.
  44     :param passthrough_stream: All non-subunit input received will be
  45         sent to this stream.  If not provided, uses the ``TestProtocolServer``
  46         default, which is ``sys.stdout``.
  47     :param forward_stream: All subunit input received will be forwarded
  48         to this stream.  If not provided, uses the ``TestProtocolServer``
  49         default, which is to not forward any input.
  50     :return: True if the test run described by ``input_stream`` was
  51         successful.  False otherwise.
  52     """
  53     test = ProtocolTestCase(
  54         input_stream, passthrough=passthrough_stream,
  55         forward=forward_stream)
  56     result.startTestRun()
  57     test.run(result)
  58     result.stopTestRun()
  59     return result.wasSuccessful()
  60
  61
  62 def filter_by_result(result_factory, output_path, no_passthrough, forward,
  63                      input_stream=sys.stdin):
  64     """Filter an input stream using a test result.
  65
  66     :param result_factory: A callable that when passed an output stream
  67         returns a TestResult.  It is expected that this result will output
  68         to the given stream.
  69     :param output_path: A path send output to.  If None, output will be go
  70         to ``sys.stdout``.
  71     :param no_passthrough: If True, all non-subunit input will be discarded.
  72         If False, that input will be sent to ``sys.stdout``.
  73     :param forward: If True, all subunit input will be forwarded directly to
  74         ``sys.stdout`` as well as to the ``TestResult``.
  75     :param input_stream: The source of subunit input.  Defaults to
  76         ``sys.stdin``.
  77     :return: 0 if the input represents a successful test run, 1 if a failed
  78         test run.
  79     """
  80     if no_passthrough:
  81         passthrough_stream = DiscardStream()
  82     else:
  83         passthrough_stream = None
  84
  85     if forward:
  86         forward_stream = sys.stdout
  87     else:
  88         forward_stream = None
  89
  90     if output_path is None:
  91         output_to = sys.stdout
  92     else:
  93         output_to = file(output_path, 'wb')
  94
  95     try:
  96         result = result_factory(output_to)
  97         was_successful = run_tests_from_stream(
  98             input_stream, result, passthrough_stream, forward_stream)
  99     finally:
 100         if output_path:
 101             output_to.close()
 102     if was_successful:
 103         return 0
 104     else:
 105         return 1
 106
 107
 108 def run_filter_script(result_factory, description):
 109     """Main function for simple subunit filter scripts.
 110
 111     Many subunit filter scripts take a stream of subunit input and use a
 112     TestResult to handle the events generated by that stream.  This function
 113     wraps a lot of the boiler-plate around that by making a script with
 114     options for handling passthrough information and stream forwarding, and
 115     that will exit with a successful return code (i.e. 0) if the input stream
 116     represents a successful test run.
 117
 118     :param result_factory: A callable that takes an output stream and returns
 119         a test result that outputs to that stream.
 120     :param description: A description of the filter script.
 121     """
 122     parser = make_options(description)
 123     (options, args) = parser.parse_args()
 124     sys.exit(
 125         filter_by_result(
 126             result_factory, options.output_to, options.no_passthrough,
 127             options.forward))