Remove shebang from subunit._output module.
[third_party/subunit] / python / subunit / _output.py
1 #  subunit: extensions to python unittest to get test results from subprocesses.
2 #  Copyright (C) 2013  Thomi Richards <thomi.richards@canonical.com>
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 argparse import ArgumentParser
16 import datetime
17 from functools import partial
18 from sys import stdout
19 from testtools.compat import _b
20
21 from subunit.v2 import StreamResultToBytes
22
23
24 def output_main():
25     args = parse_arguments()
26     output = get_output_stream_writer()
27     generate_bytestream(args, output)
28
29     return 0
30
31
32 def parse_arguments(args=None, ParserClass=ArgumentParser):
33     """Parse arguments from the command line.
34
35     If specified, args must be a list of strings, similar to sys.argv[1:].
36
37     ParserClass can be specified to override the class we use to parse the
38     command-line arguments. This is useful for testing.
39
40     """
41     parser = ParserClass(
42         prog='subunit-output',
43         description="A tool to generate a subunit result byte-stream",
44         usage="""%(prog)s [-h] action [-h] test [--attach-file ATTACH_FILE]
45                       [--mimetype MIMETYPE] [--tags TAGS]""",
46         epilog="""Additional help can be printed by passing -h to an action
47         (e.g.- '%(prog)s pass -h' will show help for the 'pass' action)."""
48     )
49
50     common_args = ParserClass(add_help=False)
51     common_args.add_argument(
52         "test_id",
53         help="A string that uniquely identifies this test."
54     )
55     common_args.add_argument(
56         "--attach-file",
57         type=open,
58         help="Attach a file to the result stream for this test."
59     )
60     common_args.add_argument(
61         "--mimetype",
62         help="The mime type to send with this file. This is only used if the "
63         "--attach-file argument is used. This argument is optional. If it is "
64         "not specified, the file will be sent wihtout a mime type.",
65         default=None
66     )
67     common_args.add_argument(
68         "--tags",
69         help="A comma-separated list of tags to associate with this test.",
70         type=lambda s: s.split(','),
71         default=None
72     )
73     sub_parsers = parser.add_subparsers(
74         dest="action",
75         title="actions",
76         description="These actions are supported by this tool",
77     )
78
79     final_state = "This is a final action: No more actions may be generated "\
80         "for this test id after this one."
81
82     sub_parsers.add_parser(
83         "start",
84         help="Start a test.",
85         parents=[common_args]
86     )
87
88     sub_parsers.add_parser(
89         "pass",
90         help="Pass a test. " + final_state,
91         parents=[common_args],
92     )
93
94     sub_parsers.add_parser(
95         "fail",
96         help="Fail a test. " + final_state,
97         parents=[common_args]
98     )
99
100     sub_parsers.add_parser(
101         "skip",
102         help="Skip a test. " + final_state,
103         parents=[common_args]
104     )
105
106     sub_parsers.add_parser(
107         "exists",
108         help="Marks a test as existing. " + final_state,
109         parents=[common_args]
110     )
111
112     sub_parsers.add_parser(
113         "expected-fail",
114         help="Marks a test as failing expectedly (this is not counted as a "
115         "failure). " + final_state,
116         parents=[common_args],
117     )
118
119     sub_parsers.add_parser(
120         "unexpected-success",
121         help="Marks a test as succeeding unexpectedly (this is counted as a "
122         "failure). " + final_state,
123         parents=[common_args],
124     )
125
126     return parser.parse_args(args)
127
128
129 def translate_command_name(command_name):
130     """Turn the friendly command names we show users on the command line into
131     something subunit understands.
132
133     """
134     return {
135         'start': 'inprogress',
136         'pass': 'success',
137         'expected-fail': 'xfail',
138         'unexpected-success': 'uxsuccess',
139     }.get(command_name, command_name)
140
141
142 def get_output_stream_writer():
143     return StreamResultToBytes(stdout)
144
145
146 def generate_bytestream(args, output_writer):
147     output_writer.startTestRun()
148     if args.attach_file:
149         write_chunked_file(
150             args.attach_file,
151             args.test_id,
152             output_writer,
153             args.mimetype,
154         )
155     output_writer.status(
156         test_id=args.test_id,
157         test_status=translate_command_name(args.action),
158         timestamp=create_timestamp(),
159         test_tags=args.tags,
160         )
161     output_writer.stopTestRun()
162
163
164 def write_chunked_file(file_obj, test_id, output_writer, chunk_size=1024,
165                        mime_type=None):
166     reader = partial(file_obj.read, chunk_size)
167     for chunk in iter(reader, _b('')):
168         output_writer.status(
169             test_id=test_id,
170             file_name=file_obj.name,
171             file_bytes=chunk,
172             mime_type=mime_type,
173             eof=False,
174         )
175     output_writer.status(
176         test_id=test_id,
177         file_name=file_obj.name,
178         file_bytes=_b(''),
179         mime_type=mime_type,
180         eof=True,
181     )
182
183
184 _ZERO = datetime.timedelta(0)
185
186
187 class UTC(datetime.tzinfo):
188
189     def utcoffset(self, dt):
190         return _ZERO
191
192     def tzname(self, dt):
193         return "UTC"
194
195     def dst(self, dt):
196         return _ZERO
197
198
199 utc = UTC()
200
201
202 def create_timestamp():
203     return datetime.datetime.now(utc)