# Ubuntu Testing Automation Harness
# Copyright 2012 Canonical Ltd.
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License version 3, as published
# by the Free Software Foundation.
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU General Public License for more details.
# You should have received a copy of the GNU General Public License along
# with this program. If not, see <http://www.gnu.org/licenses/>.
"""Provide functionality for test result handling."""
import json
import pprint
import sys
import yaml
from contextlib import contextmanager
from utah.client.common import (
get_host_info,
get_build_number,
get_release,
get_arch,
)
[docs]class Result(object):
"""Result collection class.
:params filename:
Filename where the report should be written to or None
to print it to `sys.stdout`
:type filename: str | None
:param append_to_file:
Whether results should be appended to the results from a previous
execution or not. If that's the case, then `filename` is read to get
the old results.
:type append_to_file: bool
"""
def __init__(self, filename, append_to_file,
name=None, testsuite=None, testcase=None,
runlist=None, install_type=None):
self.results = []
self.status = 'PASS'
self.filename = filename
self.append_to_file = append_to_file
self.name = name
self.testsuite = testsuite
self.testcase = testcase
self.runlist = runlist
self.install_type = install_type
self.probe_data = {}
self.packages = None
self.delta_packages = None
self.errors = 0
self.fetch_errors = 0
self.failures = 0
self.passes = 0
[docs] def load(self, old_results):
"""Load results from a previous run in which a reboot happened.
:param old_results: Old results as loaded from the report file
:type old_results: str
.. seealso:: :meth:`_payload`
"""
raise NotImplementedError
[docs] def add_result(self, result, extra_info=None):
"""Add a result to the object.
Note: 'result' is expected to be a dictionary like this::
{
'command': '',
'returncode': 0,
'stdout': '',
'stderr': '',
'start_time': '',
'time_delta': '',
}
"""
if result is None:
return
if extra_info is None:
extra_info = {}
# Add items from extra_info into the result
if len(extra_info) > 0:
result['extra_info'] = extra_info
if self.testsuite is not None:
result['testsuite'] = self.testsuite
if self.testcase is not None:
result['testcase'] = self.testcase
self.results.append(result)
if result['returncode'] != 0 and self.status != 'ERROR':
self.status = 'FAIL'
@contextmanager
[docs] def get_result_file(self):
"""Return file object to use to write the report."""
if self.filename:
if self.append_to_file:
open_flags = 'a'
else:
open_flags = 'w'
with open(self.filename, open_flags) as fp:
yield fp
else:
fp = sys.stdout
yield fp
[docs] def result(self, verbose=False):
"""Output a text based result.
:param verbose: Enable verbose mode
:type verbose: bool
:returns: Status 'PASS' 'FAIL' or 'ERROR'
:rtype: str
"""
status = self.status
sep = '-' * 70
with self.get_result_file() as output_file:
output_file.write('{}\n'.format(sep))
if self.name:
output_file.write('{}\n{}\n'.format(self.name, sep))
for result in self.results:
output_file.write(
'command: {}\n'
'returned: {}\n'
'started: {}\n'
'runtime: {}\n'
.format(result['command'],
result['returncode'],
result['start_time'],
result['time_delta']))
if self.testsuite is not None:
output_file.write('testsuite: {}\n'.format(self.testsuite))
if self.testcase is not None:
output_file.write('testcase: {}\n'.format(self.testcase))
if self.runlist is not None:
output_file.write('runlist: {}\n'.format(self.runlist))
if self.status != 'PASS' or verbose and result['stdout'] != '':
output_file.write('stdout:\n{}\n'.format(result['stdout']))
if self.status != 'PASS' or verbose and result['stderr'] != '':
output_file.write('stderr:\n{}\n'.format(result['stderr']))
output_file.write('\n')
data = self._payload()
for key, value in data.iteritems():
output_file.write('{}: {}\n'
.format(key, pprint.pformat(value)))
self._clear_results()
return status
def _clear_results(self):
"""Reset the list of results so this object can be reused."""
self.results = []
# reset the status, this should be safe since the only places that
# should be calling _clear_results() is testcase.run(),
# testsuite.run(), and runner after attempting to fetch the testsuite.
self.status = 'PASS'
def _count_results(self):
return len(self.results)
def _payload(self):
"""Construct the result payload.
:returns: Test result
:rtype: dict
"""
host_info = get_host_info()
data = {
'runlist': self.runlist,
'commands': self.results,
'errors': self.errors,
'failures': self.failures,
'fetch_errors': self.fetch_errors,
'passes': self.passes,
'uname': list(host_info['uname']),
'media-info': host_info['media-info'],
'install_type': self.install_type,
'build_number': get_build_number(),
'release': get_release(),
'ran_at': self.results[0]['start_time'],
'arch': get_arch(),
'name': self.name,
'packages': self.packages,
'delta_packages': self.delta_packages,
'probes': self.probe_data,
}
return data
# Trick to get strings printed as literal blocks
# inspired by: http://stackoverflow.com/a/7445560/183066
class _LiteralString(object):
def __init__(self, str_data):
self.str_data = str_data
def _literal_block(dumper, data):
return dumper.represent_scalar('tag:yaml.org,2002:str',
data.str_data, style='|')
yaml.add_representer(_LiteralString, _literal_block)
[docs]class ResultYAML(Result):
"""Return results in a YAML format."""
def _literalize(self, data):
"""Transform long strings into literal blocks.
:param data: Data to literalize
:type data: string, dict, or list
:returns: literalized form of data
:rtype: str, dict, or list
"""
if isinstance(data, basestring) and '\n' in data:
# Remove trailing whitespace to serialize in yaml
# as a literal string
lines = [line.rstrip() for line in data.splitlines()]
return _LiteralString('\n'.join(lines))
if isinstance(data, dict):
for key, value in data.iteritems():
data[key] = self._literalize(value)
elif isinstance(data, list):
data = [self._literalize(element)
for element in data]
return data
[docs] def result(self, verbose=False):
"""Output a YAML result.
:returns: Status 'PASS' 'FAIL' or 'ERROR'
:rtype: str
"""
if self.results:
data = self._literalize(self._payload())
with self.get_result_file() as output_file:
yaml.dump(data, output_file,
explicit_start='---',
default_flow_style=False,
allow_unicode=True)
status = self.status
self._clear_results()
return status
[docs] def load(self, old_results):
"""Load results from a previous run in which a reboot happened.
:param old_results: Old results as loaded from the report file
:type old_results: str
.. seealso:: :meth:`_payload`
"""
data = yaml.load(old_results)
self.runlist = data['runlist']
self.results = data['commands']
self.errors = data['errors']
self.failures = data['failures']
self.fetch_errors = data['fetch_errors']
self.passes = data['passes']
self.install_type = data['install_type']
self.name = data['name']
[docs]class ResultJSON(Result):
"""Return results in a JSON format."""
[docs] def result(self, _verbose=False):
"""Output a JSON result.
:returns: Status 'PASS' 'FAIL' or 'ERROR'
:rtype: str
"""
if self.results:
data = self._payload()
with self.get_result_file() as output_file:
json.dump(data, output_file, indent=4)
status = self.status
self._clear_results()
return status
# Map output format to the class that implements such a format
classes = {'text': Result,
'yaml': ResultYAML,
'json': ResultJSON,
}