694
685
super(HtmlOutput, self).startTestRun()
697
class FileAccumulator(testtools.StreamResult):
700
super(FileAccumulator, self).__init__()
701
self.route_codes = collections.defaultdict(io.BytesIO)
703
def status(self, **kwargs):
704
if kwargs.get('file_name') != 'stdout':
706
file_bytes = kwargs.get('file_bytes')
709
route_code = kwargs.get('route_code')
710
stream = self.route_codes[route_code]
711
stream.write(file_bytes)
688
class FilesResultOutput(HtmlOutput):
689
"""Output test results in html."""
691
def __init__(self, html_file='results.html'):
692
super(FilesResultOutput, self).__init__(html_file)
694
def stopTest(self, test):
695
super(FilesResultOutput, self).stopTest(test)
696
self.save_file_attachments(test)
698
def save_file_attachments(self, test):
699
for key, detail in test._details.items():
700
slugged_text = re.sub(r'(?u)[^-\w.]', '', key)
701
file_name = '%s-%s' % (test.id(), slugged_text)
702
if detail.iter_bytes() != [b'']:
703
with open(file_name, 'wb') as f:
704
for chunk in detail.iter_bytes():
708
return re.sub(r'(?u)[^-\w.]', '', text)
715
if len(sys.argv) < 2:
716
print("Need at least one argument: path to subunit log.")
718
subunit_file = sys.argv[1]
719
if len(sys.argv) > 2:
720
html_file = sys.argv[2]
722
html_file = 'results.html'
712
html_file = 'results.html'
713
subunit_file = 'results.sub'
724
html_result = HtmlOutput(html_file)
715
html_result = FilesResultOutput(html_file)
725
716
stream = open(subunit_file, 'rb')
727
# Feed the subunit stream through both a V1 and V2 parser.
728
# Depends on having the v2 capable libraries installed.
730
# Non-v2 content and captured non-test output will be presented as file
731
# segments called stdout.
732
suite = subunit.ByteStreamToStreamResult(stream, non_subunit_name='stdout')
733
# The HTML output code is in legacy mode.
718
suite = subunit.ByteStreamToStreamResult(stream)
734
719
result = testtools.StreamToExtendedDecorator(html_result)
735
# Divert non-test output
736
accumulator = FileAccumulator()
737
result = testtools.StreamResultRouter(result)
738
result.add_rule(accumulator, 'test_id', test_id=None)
739
720
result.startTestRun()
740
721
suite.run(result)
741
# Now reprocess any found stdout content as V1 subunit
742
for bytes_io in accumulator.route_codes.values():
744
suite = subunit.ProtocolTestCase(bytes_io)
745
suite.run(html_result)
746
722
result.stopTestRun()