1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
|
###################################################################################
# LAVA QA tool
# Copyright (C) 2015 Collabora Ltd.
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 US
###################################################################################
from __future__ import print_function
from datetime import datetime
from lqa_tool.commands import Command
from lqa_api.analyse import Analyse
from lqa_tool.settings import lqa_logger
from lqa_tool.version import __version__
from lqa_api.exit_codes import SUCCESS, OPERATION_FAILED
class ReportCmd(Command):
"""Sub-command to generate a LAVA lqa report"""
def __init__(self, args):
Command.__init__(self, args)
self._exit_code = SUCCESS
def run(self):
lqa_logger.info("Generating lqa report for jobs ID: {}" \
.format(" ".join(map(lambda x: str(x), self.job_ids))))
self._analyse = Analyse(self.job_ids, self.server)
self._jobs_by_status = self._analyse.sort_jobs_by_status()
self._print_report()
exit(self._exit_code)
def _print_report(self):
print("/========= LQA Report v{} ({}) =========/\n" \
.format(__version__, _set_datetime()))
if self.args.all_results:
# If --all-results passed then show all results both for failed and
# successful jobs.
show_failed = show_success = ['fail', 'skip', 'unknown', 'pass']
else:
# Otherwise only show failed results for faileds jobs by default.
show_failed, show_success = ['fail'], []
# Failed jobs
jobs_count = len(self._jobs_by_status['complete']['failed'] +
self._jobs_by_status['incomplete']['bundles'] +
self._jobs_by_status['incomplete']['no_bundles'])
if jobs_count:
# Exit with failed code if there is any failed job.
self._exit_code = OPERATION_FAILED
print("========== Failed jobs [{}] ==========\n".format(jobs_count))
self._print_jobs(self._jobs_by_status['complete']['failed'],
"Complete jobs with failures", show_failed)
self._print_jobs(self._jobs_by_status['incomplete']['bundles'],
"Incomplete jobs with bundles", show_failed)
self._print_jobs(self._jobs_by_status['incomplete']['no_bundles'],
"Incomplete jobs without bundles")
# Pending Jobs
jobs_count = len(self._jobs_by_status['pending']['submitted'] +
self._jobs_by_status['pending']['running'])
if jobs_count:
print("========== Pending jobs [{}] ==========\n".format(jobs_count))
self._print_jobs(self._jobs_by_status['pending']['submitted'],
"Submitted jobs")
self._print_jobs(self._jobs_by_status['pending']['running'],
"Running jobs")
# Canceled Jobs
jobs_count = len(self._jobs_by_status['canceled']['canceled'] +
self._jobs_by_status['canceled']['canceling'])
if jobs_count:
print("========== Canceled jobs [{}] ==========\n".format(jobs_count))
self._print_jobs(self._jobs_by_status['canceled']['canceling'],
"Canceling jobs")
self._print_jobs(self._jobs_by_status['canceled']['canceled'],
"Canceled jobs")
# Successful jobs
jobs_count = len(self._jobs_by_status['complete']['successful'])
if jobs_count:
print("========== Successful jobs [{}] ==========\n".format(jobs_count))
self._print_jobs(self._jobs_by_status['complete']['successful'],
"Successful jobs (no failures)", show_success)
def _print_jobs(self, jobs, section_header, results_status=[]):
if jobs:
print("[ --- {} --- ]\n".format(section_header))
for job in jobs:
_print_job_header(job)
if results_status:
_print_tests(job.tests, results_status, self.args.limit)
def _print_job_header(job):
space = " "
print(space, job)
print(space, len(str(job)) * "=")
if job.failure_comment:
print(space, "Failure:", job.failure_comment.strip())
print(space, "Bundle Link:", job.bundle_link)
print()
def _print_tests(tests, results_status, limit):
for test in tests:
is_test_printed = False
for status in results_status:
results = { "fail": test.results_failed,
"pass": test.results_passed,
"skip": test.results_skipped,
"unknown": test.results_unknown }.get(status, [])
# Continue to next status if no results for this one.
if not results:
continue
# Print test name only once for all its results.
if not is_test_printed:
print("\t", test)
is_test_printed = True
for i, test_case in enumerate(results, 1):
if i > limit:
print("\t[* NOTE: {} test cases with status {}." \
" Visit bundle link for complete list of" \
" results or set a new limit with the --limit option *]"\
.format(limit, status.upper()))
break
print("\t {} - {}".format(test_case['result'].upper(),
test_case['test_case_id']))
print()
def _set_datetime():
return datetime.today().strftime("%a %b %d %H:%M:%S %Y")
|
