Source code for stestr.output

# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import io
import sys

import subunit
import testtools

[docs]def output_table(table, output=sys.stdout): """Display a table of information. :param table: A list of sets representing each row in the table. Each element in the set represents a column in the table. :param output: The output file object to write the table to. By default this is sys.stdout """ # stringify contents = [] for row in table: new_row = [] for column in row: new_row.append(str(column)) contents.append(new_row) if not contents: return widths = [0] * len(contents[0]) for row in contents: for idx, column in enumerate(row): if widths[idx] < len(column): widths[idx] = len(column) # Show a row outputs = [] def show_row(row): for idx, column in enumerate(row): outputs.append(column) if idx == len(row) - 1: outputs.append('\n') return # spacers for the next column outputs.append(' ' * (widths[idx] - len(column))) outputs.append(' ') show_row(contents[0]) # title spacer for idx, width in enumerate(widths): outputs.append('-' * width) if idx == len(widths) - 1: outputs.append('\n') continue outputs.append(' ') for row in contents[1:]: show_row(row) output.write(''.join(outputs))
[docs]def output_tests(tests, output=sys.stdout): """Display a list of tests. :param tests: A list of test objects to output :param output: The output file object to write the list to. By default this is sys.stdout """ for test in tests: id_str = output.write(str(id_str)) output.write('\n')
def make_result(get_id, output=sys.stdout): serializer = subunit.StreamResultToBytes(output) # By pass user transforms - just forward it all, result = serializer # and interpret everything as success. summary = testtools.StreamSummary() summary.startTestRun() summary.stopTestRun() return result, summary
[docs]def output_summary(successful, tests, tests_delta, time, time_delta, values, output=sys.stdout): """Display a summary view for the test run. :param bool successful: Was the test run successful :param int tests: The number of tests that ran :param int tests_delta: The change in the number of tests that ran since the last run :param float time: The number of seconds that it took for the run to execute :param float time_delta: The change in run time since the last run :param values: A list of sets that are used for a breakdown of statuses other than success. Each set is in the format: (status, number of tests, change in number of tests). :param output: The output file object to use. This defaults to stdout """ summary = [] a = summary.append if tests: a("Ran {}".format(tests)) if tests_delta: a(" (%+d)" % (tests_delta,)) a(" tests") if time: if not summary: a("Ran tests") a(" in {:0.3f}s".format(time)) if time_delta: a(" ({:+0.3f}s)".format(time_delta)) if summary: a("\n") if successful: a('PASSED') else: a('FAILED') if values: a(' (') values_strings = [] for name, value, delta in values: value_str = '{}={}'.format(name, value) if delta: value_str += ' (%+d)' % (delta,) values_strings.append(value_str) a(', '.join(values_strings)) a(')') output.write(''.join(summary) + '\n')
def output_stream(stream, output=sys.stdout): _binary_stdout = subunit.make_stream_binary(output) contents = assert type(contents) is bytes, \ "Bad stream contents %r" % type(contents) # If there are unflushed bytes in the text wrapper, we need to sync.. output.flush() while contents: _binary_stdout.write(contents) contents = _binary_stdout.flush()
[docs]class ReturnCodeToSubunit: """Converts a process return code to a subunit error on the process stdout. The ReturnCodeToSubunit object behaves as a read-only stream, supplying the read, readline and readlines methods. If the process exits non-zero a synthetic test is added to the output, making the error accessible to subunit stream consumers. If the process closes its stdout and then does not terminate, reading from the ReturnCodeToSubunit stream will hang. :param process: A subprocess.Popen object that is generating subunit. """ def __init__(self, process): """Adapt a process to a readable stream.""" self.proc = process self.done = False self.source = self.proc.stdout self.lastoutput = bytes((b'\n')[0]) def _append_return_code_as_test(self): if self.done is True: return self.source = io.BytesIO() returncode = self.proc.wait() if returncode != 0: if self.lastoutput != bytes((b'\n')[0]): # Subunit V1 is line orientated, it has to start on a fresh # line. V2 needs to start on any fresh utf8 character border # - which is not guaranteed in an arbitrary stream endpoint, so # injecting a \n gives us such a guarantee. self.source.write(bytes('\n')) stream = subunit.StreamResultToBytes(self.source) stream.status(test_id='process-returncode', test_status='fail', file_name='traceback', mime_type='text/plain;charset=utf8', file_bytes=( 'returncode %d' % returncode).encode('utf8')) self.done = True def read(self, count=-1): if count == 0: return '' result = if result: self.lastoutput = result[-1] return result self._append_return_code_as_test() return def readline(self): result = self.source.readline() if result: self.lastoutput = result[-1] return result self._append_return_code_as_test() return self.source.readline() def readlines(self): result = self.source.readlines() if result: self.lastoutput = result[-1][-1] self._append_return_code_as_test() result.extend(self.source.readlines()) return result