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 wihtout 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         "--tags",
 119         help="A comma-separated list of tags to associate with a test. This "
 120             "option may only be used with a status command.",
 121         action="callback",
 122         callback=set_tags_cb,
 123         default=[]
 124     )
 125
 126     (options, args) = parser.parse_args(args)
 127     if options.mimetype and not options.attach_file:
 128         parser.error("Cannot specify --mimetype without --attach-file")
 129     if options.file_name and not options.attach_file:
 130         parser.error("Cannot specify --file-name without --attach-file")
 131     if options.attach_file:
 132         if options.attach_file == '-':
 133             if not options.file_name:
 134                 options.file_name = 'stdin'
 135                 options.attach_file = make_stream_binary(sys.stdin)
 136         else:
 137             try:
 138                 options.attach_file = open(options.attach_file, 'rb')
 139             except IOError as e:
 140                 parser.error("Cannot open %s (%s)" % (options.attach_file, e.strerror))
 141     if options.tags and not options.action:
 142         parser.error("Cannot specify --tags without a status command")
 143     if not (options.attach_file or options.action):
 144         parser.error("Must specify either --attach-file or a status command")
 145
 146     return options
 147
 148
 149 def set_status_cb(option, opt_str, value, parser, status_name):
 150     if getattr(parser.values, "action", None) is not None:
 151         raise OptionValueError("argument %s: Only one status may be specified at once." % opt_str)
 152
 153     if len(parser.rargs) == 0:
 154         raise OptionValueError("argument %s: must specify a single TEST_ID." % opt_str)
 155     parser.values.action = status_name
 156     parser.values.test_id = parser.rargs.pop(0)
 157
 158
 159 def set_tags_cb(option, opt_str, value, parser):
 160     if not parser.rargs:
 161         raise OptionValueError("Must specify at least one tag with --tags")
 162     parser.values.tags = parser.rargs.pop(0).split(',')
 163
 164
 165 def generate_stream_results(args, output_writer):
 166     output_writer.startTestRun()
 167
 168     if args.attach_file:
 169         reader = partial(args.attach_file.read, _CHUNK_SIZE)
 170         this_file_hunk = reader()
 171         next_file_hunk = reader()
 172
 173     is_first_packet = True
 174     is_last_packet = False
 175     while not is_last_packet:
 176         write_status = output_writer.status
 177
 178         if is_first_packet:
 179             if args.attach_file:
 180                 if args.mimetype:
 181                     write_status = partial(write_status, mime_type=args.mimetype)
 182             if args.tags:
 183                 write_status = partial(write_status, test_tags=args.tags)
 184             write_status = partial(write_status, timestamp=create_timestamp())
 185             if args.action not in _FINAL_ACTIONS:
 186                 write_status = partial(write_status, test_status=args.action)
 187             is_first_packet = False
 188
 189         if args.attach_file:
 190             filename = args.file_name or args.attach_file.name
 191             write_status = partial(write_status, file_name=filename, file_bytes=this_file_hunk)
 192             if next_file_hunk == b'':
 193                 write_status = partial(write_status, eof=True)
 194                 is_last_packet = True
 195             else:
 196                 this_file_hunk = next_file_hunk
 197                 next_file_hunk = reader()
 198         else:
 199             is_last_packet = True
 200
 201         if args.test_id:
 202             write_status = partial(write_status, test_id=args.test_id)
 203
 204         if is_last_packet:
 205             write_status = partial(write_status, eof=True)
 206             if args.action in _FINAL_ACTIONS:
 207                 write_status = partial(write_status, test_status=args.action)
 208
 209         write_status()
 210
 211     output_writer.stopTestRun()
 212
 213
 214 def create_timestamp():
 215     return datetime.datetime.now(UTC)