-
Notifications
You must be signed in to change notification settings - Fork 7
/
notifications.py
executable file
·180 lines (150 loc) · 5.58 KB
/
notifications.py
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
#!/usr/bin/env python3
# -*- coding:utf-8 -*
#
# Florent Jacquet <florent.jacquet@free-electrons.com>
#
import json
import os
from xmlrpc import client
from pprint import pprint
from datetime import datetime, timedelta
from collections import OrderedDict
import configparser
import smtplib
from email.mime.text import MIMEText
JOB_STATUS = {
0: "Submitted",
1: "Running",
2: "Complete",
3: "Incomplete",
4: "Canceled",
5: "Canceling",
}
JOB_REPORT_STATUS = OrderedDict()
JOB_REPORT_STATUS['failed_tests'] = "The following jobs did not pass the custom tests:"
JOB_REPORT_STATUS['failed_boot'] = "The following jobs did not boot:"
JOB_REPORT_STATUS['lava_error'] = "The following jobs did not reach power up:"
JOB_REPORT_STATUS['passed_tests'] = "The following jobs have ended up without problem:"
# Get config
config = configparser.ConfigParser()
config.read("notifications.conf")
# Get board config
ctt_root_location = os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
with open(os.path.join(ctt_root_location, "ci_tests.json")) as f:
boards_config = json.load(f)
# Get LAVA API handler
lava_api = client.ServerProxy("http://%s:%s@%s/RPC2" % (
config.get("lava", "user"),
config.get("lava", "token"),
config.get("lava", "hostname")), allow_none=True)
# Define a Job class to ease processing
class Job(object):
TABLE_TITLE = "".join([
'{:<12}'.format("tree"),
'{:<14}'.format("branch"),
'{:<25}'.format("device"),
'{:<30}'.format("defconfig"),
'{:<12}'.format("test"),
"link",
])
def __str__(self):
return "".join([
'{:<12}'.format(self.tree[:9] + '..' if len(self.tree) > 9 else self.tree),
'{:<14}'.format(self.branch[:11] + '..' if len(self.branch) > 11 else self.branch),
'{:<25}'.format(self.device[:22] + '..' if len(self.device) > 22 else self.device),
'{:<30}'.format(self.defconfig[:27] + '..' if len(self.defconfig) > 27 else self.defconfig),
'{:<12}'.format(self.test[:9] + '..' if len(self.test) > 9 else self.test),
"http://lava.bootlin.com/scheduler/job/%s" % self.id,
])
def __init__(self, job):
"""
`job` is a dictionary returned by the LAVA API
"""
self.id = job['id']
d = job['description'].split('--')
self.device = d[0]
self.tree = d[1]
self.branch = d[2]
self.defconfig = d[3]
self.test = d[4]
def has_powered_up(self):
power_up = lava_api.results.make_custom_query("testjob",
"testjob__id__exact__%s,"
"testcase__name__exact__pdu-reboot"
% (self.id))
return bool(power_up)
def has_booted(self):
boot = lava_api.results.make_custom_query("testjob",
"testjob__id__exact__%s,"
"testcase__name__exact__auto-login-action"
% (self.id))
return bool(boot)
def has_passed_test(self):
tests = lava_api.results.make_custom_query("testcase",
"testjob__id__exact__%s,"
"testsuite__name__endswith__%s"
% (self.id, "custom-tests"))
if not tests:
return False
return not bool(tests[0]['result'])
def main():
end_date = datetime.now()
start_date = end_date - timedelta(hours=24)
# mail_list is a data structure built like this:
# mail_list = {
# 'email.address@domain.tld': {
# 'passed_tests': [str(Job object), str(other Job object)],
# 'failed_tests': [str(Job object), ...],
# }
# }
mail_list = {}
# Get all the last jobs
all_jobs = lava_api.results.make_custom_query("testjob",
"testjob__end_time__gt__%s,"
"testjob__end_time__lt__%s,"
"testjob__submitter__exact__custom-tests"
% (start_date, end_date), None)
# Build mail_list
for job in all_jobs:
if job['requested_device_type_id'] == "dummy-ssh":
continue
j = Job(job)
if j.has_passed_test():
status = "passed_tests"
elif j.has_booted():
status = "failed_tests"
elif j.has_powered_up():
status = "failed_boot"
else:
status = "lava_error"
board = boards_config.get(j.device, None)
if board:
for email in board.get("notify", []):
if email not in mail_list:
mail_list[email] = {}
if status not in mail_list[email]:
mail_list[email][status] = [Job.TABLE_TITLE]
mail_list[email][status].append(str(j))
# Now send the emails
server = smtplib.SMTP(config.get("mail", "server"), config.get("mail", "port"))
server.ehlo()
server.starttls()
server.login(config.get("mail", "login"), config.get("mail", "password"))
for user,job_list in mail_list.items():
msg_list = []
for status,msg in JOB_REPORT_STATUS.items():
if status in job_list:
msg_list += [msg]
msg_list += job_list[status]
msg_list += [""]
print("\nMessage to %s:" % user)
print('\n'.join(msg_list))
msg = MIMEText('\n'.join(msg_list))
msg['Subject'] = "CI summary"
msg['From'] = config.get("mail", "from")
msg['To'] = user
server.sendmail(config.get("mail", "from"), [user], msg.as_string())
server.quit()
return
if __name__ == "__main__":
main()