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

   1 #  subunit: extensions to python unittest to get test results from subprocesses.
   2 #  Copyright (C) 2013 Subunit Contributors
   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 import datetime
  17 from functools import partial
  18 from optparse import (
  19     OptionGroup,
  20     OptionParser,
  21     OptionValueError,
  22 )
  23 import sys
  24
  25 from subunit import make_stream_binary
  26 from subunit.iso8601 import UTC
  27 from subunit.v2 import StreamResultToBytes
  28
  29
  30 _FINAL_ACTIONS = frozenset([
  31     'exists',
  32     'fail',
  33     'skip',
  34     'success',
  35     'uxsuccess',
  36     'xfail',
  37 ])
  38 _ALL_ACTIONS = _FINAL_ACTIONS.union(['inprogress'])
  39 _CHUNK_SIZE=3670016 # 3.5 MiB
  40
  41
  42 def output_main():
  43     args = parse_arguments()
  44     output = StreamResultToBytes(sys.stdout)
  45     generate_stream_results(args, output)
  46     return 0
  47
  48
  49 def parse_arguments(args=None, ParserClass=OptionParser):
  50     """Parse arguments from the command line.
  51
  52     If specified, args must be a list of strings, similar to sys.argv[1:].
  53
  54     ParserClass may be specified to override the class we use to parse the
  55     command-line arguments. This is useful for testing.
  56     """
  57     parser = ParserClass(
  58         prog="subunit-output",
  59         description="A tool to generate a subunit v2 result byte-stream",
  60         usage="subunit-output [-h] [status TEST_ID] [options]",
  61     )
  62     parser.set_default('tags', None)
  63     parser.set_default('test_id', None)
  64
  65     status_commands = OptionGroup(
  66         parser,
  67         "Status Commands",
  68         "These options report the status of a test. TEST_ID must be a string "
  69             "that uniquely identifies the test."
  70     )
  71     for action_name in _ALL_ACTIONS:
  72         status_commands.add_option(
  73             "--%s" % action_name,
  74             nargs=1,
  75             action="callback",
  76             callback=set_status_cb,
  77             callback_args=(action_name,),
  78             dest="action",
  79             metavar="TEST_ID",
  80             help="Report a test status."
  81         )
  82     parser.add_option_group(status_commands)
  83
  84     file_commands = OptionGroup(
  85         parser,
  86         "File Options",
  87         "These options control attaching data to a result stream. They can "
  88             "either be specified with a status command, in which case the file "
  89             "is attached to the test status, or by themselves, in which case "
  90             "the file is attached to the stream (and not associated with any "
  91             "test id)."
  92     )
  93     file_commands.add_option(
  94         "--attach-file",
  95         help="Attach a file to the result stream for this test. If '-' is "
  96             "specified, stdin will be read instead. In this case, the file "
  97             "name will be set to 'stdin' (but can still be overridden with "
  98             "the --file-name option)."
  99     )
 100     file_commands.add_option(
 101         "--file-name",
 102         help="The name to give this file attachment. If not specified, the "
 103             "name of the file on disk will be used, or 'stdin' in the case "
 104             "where '-' was passed to the '--attach-file' argument. This option"
 105             " may only be specified when '--attach-file' is specified.",
 106         )
 107     file_commands.add_option(
 108         "--mimetype",
 109         help="The mime type to send with this file. This is only used if the "
 110             "--attach-file argument is used. This argument is optional. If it "
 111             "is not specified, the file will be sent without a mime type. This "
 112             "option may only be specified when '--attach-file' is specified.",
 113         default=None
 114     )
 115     parser.add_option_group(file_commands)
 116
 117     parser.add_option(
 118         "--tag",
 119         help="Specifies a tag. May be used multiple times",
 120         action="append",
 121         dest="tags",
 122         default=[]
 123     )
 124
 125     (options, args) = parser.parse_args(args)
 126     if options.mimetype and not options.attach_file:
 127         parser.error("Cannot specify --mimetype without --attach-file")
 128     if options.file_name and not options.attach_file:
 129         parser.error("Cannot specify --file-name without --attach-file")
 130     if options.attach_file:
 131         if options.attach_file == '-':
 132             if not options.file_name:
 133                 options.file_name = 'stdin'
 134                 options.attach_file = make_stream_binary(sys.stdin)
 135         else:
 136             try:
 137                 options.attach_file = open(options.attach_file, 'rb')
 138             except IOError as e:
 139                 parser.error("Cannot open %s (%s)" % (options.attach_file, e.strerror))
 140     if options.tags and not options.action:
 141         parser.error("Cannot specify --tag without a status command")
 142     if not (options.attach_file or options.action):
 143         parser.error("Must specify either --attach-file or a status command")
 144
 145     return options
 146
 147
 148 def set_status_cb(option, opt_str, value, parser, status_name):
 149     if getattr(parser.values, "action", None) is not None:
 150         raise OptionValueError("argument %s: Only one status may be specified at once." % opt_str)
 151
 152     if len(parser.rargs) == 0:
 153         raise OptionValueError("argument %s: must specify a single TEST_ID." % opt_str)
 154     parser.values.action = status_name
 155     parser.values.test_id = parser.rargs.pop(0)
 156
 157
 158 def generate_stream_results(args, output_writer):
 159     output_writer.startTestRun()
 160
 161     if args.attach_file:
 162         reader = partial(args.attach_file.read, _CHUNK_SIZE)
 163         this_file_hunk = reader()
 164         next_file_hunk = reader()
 165
 166     is_first_packet = True
 167     is_last_packet = False
 168     while not is_last_packet:
 169         write_status = output_writer.status
 170
 171         if is_first_packet:
 172             if args.attach_file:
 173                 if args.mimetype:
 174                     write_status = partial(write_status, mime_type=args.mimetype)
 175             if args.tags:
 176                 write_status = partial(write_status, test_tags=args.tags)
 177             write_status = partial(write_status, timestamp=create_timestamp())
 178             if args.action not in _FINAL_ACTIONS:
 179                 write_status = partial(write_status, test_status=args.action)
 180             is_first_packet = False
 181
 182         if args.attach_file:
 183             filename = args.file_name or args.attach_file.name
 184             write_status = partial(write_status, file_name=filename, file_bytes=this_file_hunk)
 185             if next_file_hunk == b'':
 186                 write_status = partial(write_status, eof=True)
 187                 is_last_packet = True
 188             else:
 189                 this_file_hunk = next_file_hunk
 190                 next_file_hunk = reader()
 191         else:
 192             is_last_packet = True
 193
 194         if args.test_id:
 195             write_status = partial(write_status, test_id=args.test_id)
 196
 197         if is_last_packet:
 198             if args.action in _FINAL_ACTIONS:
 199                 write_status = partial(write_status, test_status=args.action)
 200
 201         write_status()
 202
 203     output_writer.stopTestRun()
 204
 205
 206 def create_timestamp():
 207     return datetime.datetime.now(UTC)