diff options
Diffstat (limited to 'python/subunit/_output.py')
| -rw-r--r-- | python/subunit/_output.py | 203 |
1 files changed, 203 insertions, 0 deletions
diff --git a/python/subunit/_output.py b/python/subunit/_output.py new file mode 100644 index 0000000..aa92646 --- /dev/null +++ b/python/subunit/_output.py @@ -0,0 +1,203 @@ +# subunit: extensions to python unittest to get test results from subprocesses. +# Copyright (C) 2013 Subunit Contributors +# +# 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. +# + +import datetime +from functools import partial +from optparse import ( + OptionGroup, + OptionParser, + OptionValueError, +) +import sys + +from subunit import make_stream_binary +from subunit.iso8601 import UTC +from subunit.v2 import StreamResultToBytes + + +_FINAL_ACTIONS = frozenset([ + 'exists', + 'fail', + 'skip', + 'success', + 'uxsuccess', + 'xfail', +]) +_ALL_ACTIONS = _FINAL_ACTIONS.union(['inprogress']) +_CHUNK_SIZE=3670016 # 3.5 MiB + + +def output_main(): + args = parse_arguments() + output = StreamResultToBytes(sys.stdout) + generate_stream_results(args, output) + return 0 + + +def parse_arguments(args=None, ParserClass=OptionParser): + """Parse arguments from the command line. + + If specified, args must be a list of strings, similar to sys.argv[1:]. + + ParserClass may be specified to override the class we use to parse the + command-line arguments. This is useful for testing. + """ + parser = ParserClass( + prog="subunit-output", + description="A tool to generate a subunit v2 result byte-stream", + usage="subunit-output [-h] [status TEST_ID] [options]", + ) + parser.set_default('tags', None) + parser.set_default('test_id', None) + + status_commands = OptionGroup( + parser, + "Status Commands", + "These options report the status of a test. TEST_ID must be a string " + "that uniquely identifies the test." + ) + for action_name in _ALL_ACTIONS: + status_commands.add_option( + "--%s" % action_name, + nargs=1, + action="callback", + callback=set_status_cb, + callback_args=(action_name,), + dest="action", + metavar="TEST_ID", + help="Report a test status." + ) + parser.add_option_group(status_commands) + + file_commands = OptionGroup( + parser, + "File Options", + "These options control attaching data to a result stream. They can " + "either be specified with a status command, in which case the file " + "is attached to the test status, or by themselves, in which case " + "the file is attached to the stream (and not associated with any " + "test id)." + ) + file_commands.add_option( + "--attach-file", + help="Attach a file to the result stream for this test. If '-' is " + "specified, stdin will be read instead. In this case, the file " + "name will be set to 'stdin' (but can still be overridden with " + "the --file-name option)." + ) + file_commands.add_option( + "--file-name", + help="The name to give this file attachment. If not specified, the " + "name of the file on disk will be used, or 'stdin' in the case " + "where '-' was passed to the '--attach-file' argument. This option" + " may only be specified when '--attach-file' is specified.", + ) + file_commands.add_option( + "--mimetype", + help="The mime type to send with this file. This is only used if the " + "--attach-file argument is used. This argument is optional. If it " + "is not specified, the file will be sent without a mime type. This " + "option may only be specified when '--attach-file' is specified.", + default=None + ) + parser.add_option_group(file_commands) + + parser.add_option( + "--tag", + help="Specifies a tag. May be used multiple times", + action="append", + dest="tags", + default=[] + ) + + (options, args) = parser.parse_args(args) + if options.mimetype and not options.attach_file: + parser.error("Cannot specify --mimetype without --attach-file") + if options.file_name and not options.attach_file: + parser.error("Cannot specify --file-name without --attach-file") + if options.attach_file: + if options.attach_file == '-': + if not options.file_name: + options.file_name = 'stdin' + options.attach_file = make_stream_binary(sys.stdin) + else: + try: + options.attach_file = open(options.attach_file, 'rb') + except IOError as e: + parser.error("Cannot open %s (%s)" % (options.attach_file, e.strerror)) + + return options + + +def set_status_cb(option, opt_str, value, parser, status_name): + if getattr(parser.values, "action", None) is not None: + raise OptionValueError("argument %s: Only one status may be specified at once." % opt_str) + + if len(parser.rargs) == 0: + raise OptionValueError("argument %s: must specify a single TEST_ID." % opt_str) + parser.values.action = status_name + parser.values.test_id = parser.rargs.pop(0) + + +def generate_stream_results(args, output_writer): + output_writer.startTestRun() + + if args.attach_file: + reader = partial(args.attach_file.read, _CHUNK_SIZE) + this_file_hunk = reader() + next_file_hunk = reader() + + is_first_packet = True + is_last_packet = False + while not is_last_packet: + write_status = output_writer.status + + if is_first_packet: + if args.attach_file: + if args.mimetype: + write_status = partial(write_status, mime_type=args.mimetype) + if args.tags: + write_status = partial(write_status, test_tags=set(args.tags)) + write_status = partial(write_status, timestamp=create_timestamp()) + if args.action not in _FINAL_ACTIONS: + write_status = partial(write_status, test_status=args.action) + is_first_packet = False + + if args.attach_file: + filename = args.file_name or args.attach_file.name + write_status = partial(write_status, file_name=filename, file_bytes=this_file_hunk) + if next_file_hunk == b'': + write_status = partial(write_status, eof=True) + is_last_packet = True + else: + this_file_hunk = next_file_hunk + next_file_hunk = reader() + else: + is_last_packet = True + + if args.test_id: + write_status = partial(write_status, test_id=args.test_id) + + if is_last_packet: + if args.action in _FINAL_ACTIONS: + write_status = partial(write_status, test_status=args.action) + + write_status() + + output_writer.stopTestRun() + + +def create_timestamp(): + return datetime.datetime.now(UTC) |