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