#4403

pending completion

Build # #4403

Build Type

push

coveralls-python

Committed by

suhaibmujahid

Commit Message

Replace the word "tool" with "rule"

Run Details

639 of 3205 branches covered (19.94%)

57 of 57 new or added lines in 6 files covered. (100.0%)

1816 of 8004 relevant lines covered (22.69%)

0.23 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

13.04

/bugbot/nag_me.py

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.

import copy

from jinja2 import Environment, FileSystemLoader

from bugbot import db, logger, mail, utils
from bugbot.escalation import Escalation
from bugbot.people import People


class Nag(object):
    def __init__(self):
        super(Nag, self).__init__()
        self.people = People.get_instance()
        self.send_nag_mail = True
        self.data = {}
        self.nag_date = None
        self.white_list = []
        self.black_list = []
        self.escalation = Escalation(self.people)
        self.triage_owners_components = {}
        self.all_owners = None
        self.query_params = {}
        self.round_robin = None

    @staticmethod
    def get_from():
        return utils.get_config("bugbot", "from", "release-mgmt@mozilla.com")

    def get_cc(self):
        cc = self.get_config("cc", None)
        if cc is None:
            cc = utils.get_config("bugbot", "cc", [])

        return set(cc)

    def get_priority(self, bug):
        tracking = bug[self.tracking]
        if tracking == "blocking":
            return "high"
        return "normal"

    def filter_bug(self, priority):
        days = (utils.get_next_release_date() - self.nag_date).days
        weekday = self.nag_date.weekday()
        return self.escalation.filter(priority, days, weekday)

    def get_people(self):
        return self.people

    def set_people_to_nag(self, bug, buginfo):
        return bug

    def escalate(self, person, priority, **kwargs):
        days = (utils.get_next_release_date() - self.nag_date).days
        return self.escalation.get_supervisor(priority, days, person, **kwargs)

    def add(self, persons, bug_data, priority="default", **kwargs):
        if not isinstance(persons, list):
            persons = [persons]

        persons = [p for p in persons if self.people.is_mozilla(p)]
        if not persons:
            return False

        managers = {p: self.escalate(p, priority, **kwargs) for p in persons}
        return self.add_couples(managers, bug_data)

    def add_couples(self, managers, bug_data):
        for person, manager in managers.items():
            person = self.people.get_moz_mail(person)

            if manager in self.data:
                data = self.data[manager]
            else:
                self.data[manager] = data = {}

            if person in data:
                data[person].append(bug_data)
            else:
                data[person] = [bug_data]

        return True

    def nag_template(self):
        return self.name() + "_nag.html"

    def nag_preamble(self):
        return None

    def get_extra_for_nag_template(self):
        return {}

    def columns_nag(self):
        return None

    def sort_columns_nag(self):
        return None

    def _is_in_list(self, mail, _list):
        for manager in _list:
            if self.people.is_under(mail, manager):
                return True
        return False

    def is_under(self, mail):
        if not self.white_list:
            if not self.black_list:
                return True
            return not self._is_in_list(mail, self.black_list)
        if not self.black_list:
            return self._is_in_list(mail, self.white_list)
        return self._is_in_list(mail, self.white_list) and not self._is_in_list(
            mail, self.black_list
        )

    def add_triage_owner(self, owners, real_owner):
        if self.round_robin is None:
            return

        if not isinstance(owners, list):
            owners = [owners]
        for owner in owners:
            person = self.people.get_moz_mail(owner)
            if person not in self.triage_owners_components:
                self.triage_owners_components[person] = set(
                    self.round_robin.get_components_for_triager(owner)
                )
            else:
                self.triage_owners_components[
                    person
                ] |= self.round_robin.get_components_for_triager(owner)

    def get_query_url_for_components(self, components):
        params = copy.deepcopy(self.query_params)
        for field in ["include_fields", "product", "component", "bug_id"]:
            if field in params:
                del params[field]

        utils.add_prod_comp_to_query(params, components)
        url = utils.get_bz_search_url(params)

        return url

    def organize_nag(self, bugs):
        columns = self.columns_nag()
        if columns is None:
            columns = self.columns()
        key = self.sort_columns_nag()
        if key is None:
            key = self.sort_columns()

        return utils.organize(bugs, columns, key=key)

    def send_mails(self, title, dryrun=False):
        if not self.send_nag_mail:
            return

        env = Environment(loader=FileSystemLoader("templates"))
        common = env.get_template("common.html")
        login_info = utils.get_login_info()
        From = Nag.get_from()
        Default_Cc = self.get_cc()
        mails = self.prepare_mails()

        for m in mails:
            Cc = Default_Cc | m["management_chain"]
            if m["manager"]:
                Cc.add(m["manager"])
            body = common.render(message=m["body"], query_url=None)
            receivers = set(m["to"]) | set(Cc)
            status = "Success"
            try:
                mail.send(
                    From,
                    sorted(m["to"]),
                    title,
                    body,
                    Cc=sorted(Cc),
                    html=True,
                    login=login_info,
                    dryrun=dryrun,
                )
            except Exception:
                logger.exception("Rule {}".format(self.name()))
                status = "Failure"

            db.Email.add(self.name(), receivers, "individual", status)

    def prepare_mails(self):
        if not self.data:
            return []

        template = self.nag_template()
        if not template:
            return []

        # If we escalating only to hierarchical mangers, we should always have a
        # management chain.
        fail_on_missing_mgmt_chain = self.escalation.is_hierarchical_escalation_only()

        extra = self.get_extra_for_nag_template()
        env = Environment(loader=FileSystemLoader("templates"))
        template = env.get_template(template)
        mails = []
        for manager, info in self.data.items():
            # The same bug can be several times in the list
            # because we send an email to a team.
            added_bug_ids = set()

            data = []
            To = sorted(info.keys())
            components = set()
            management_chain = set()
            for person in To:
                data += [
                    bug_data
                    for bug_data in info[person]
                    if bug_data["id"] not in added_bug_ids
                ]
                added_bug_ids.update(bug_data["id"] for bug_data in info[person])
                if person in self.triage_owners_components:
                    components |= self.triage_owners_components[person]

                management_chain |= self.people.get_management_chain_mails(
                    person, manager, fail_on_missing_mgmt_chain
                )

            if components:
                query_url = self.get_query_url_for_components(sorted(components))
            else:
                query_url = None

            body = template.render(
                date=self.nag_date,
                extra=extra,
                plural=utils.plural,
                enumerate=enumerate,
                data=self.organize_nag(data),
                nag=True,
                query_url_nag=utils.shorten_long_bz_url(query_url),
                table_attrs=self.get_config("table_attrs"),
                nag_preamble=self.nag_preamble(),
            )

            m = {
                "manager": manager,
                "management_chain": management_chain,
                "to": set(To),
                "body": body,
            }
            mails.append(m)

        return mails

    def reorganize_to_bag(self, data):
        return data

1	# This Source Code Form is subject to the terms of the Mozilla Public
2	# License, v. 2.0. If a copy of the MPL was not distributed with this file,
3	# You can obtain one at http://mozilla.org/MPL/2.0/.
4
5	import copy	1✔
6
7	from jinja2 import Environment, FileSystemLoader	1✔
8
9	from bugbot import db, logger, mail, utils	1✔
10	from bugbot.escalation import Escalation	1✔
11	from bugbot.people import People	1✔
12
13
14	class Nag(object):	1✔
15	def __init__(self):	1✔
16	super(Nag, self).__init__()	×
17	self.people = People.get_instance()	×
18	self.send_nag_mail = True	×
19	self.data = {}	×
20	self.nag_date = None	×
21	self.white_list = []	×
22	self.black_list = []	×
23	self.escalation = Escalation(self.people)	×
24	self.triage_owners_components = {}	×
25	self.all_owners = None	×
26	self.query_params = {}	×
27	self.round_robin = None	×
28
29	@staticmethod	1✔
30	def get_from():	1✔
31	return utils.get_config("bugbot", "from", "release-mgmt@mozilla.com")	×
32
33	def get_cc(self):	1✔
34	cc = self.get_config("cc", None)	×
35	if cc is None:	×
36	cc = utils.get_config("bugbot", "cc", [])	×
37
38	return set(cc)	×
39
40	def get_priority(self, bug):	1✔
41	tracking = bug[self.tracking]	×
42	if tracking == "blocking":	×
43	return "high"	×
44	return "normal"	×
45
46	def filter_bug(self, priority):	1✔
47	days = (utils.get_next_release_date() - self.nag_date).days	×
48	weekday = self.nag_date.weekday()	×
49	return self.escalation.filter(priority, days, weekday)	×
50
51	def get_people(self):	1✔
52	return self.people	×
53
54	def set_people_to_nag(self, bug, buginfo):	1✔
55	return bug	×
56
57	def escalate(self, person, priority, **kwargs):	1✔
58	days = (utils.get_next_release_date() - self.nag_date).days	×
59	return self.escalation.get_supervisor(priority, days, person, **kwargs)	×
60
61	def add(self, persons, bug_data, priority="default", **kwargs):	1✔
62	if not isinstance(persons, list):	×
63	persons = [persons]	×
64
65	persons = [p for p in persons if self.people.is_mozilla(p)]	×
66	if not persons:	×
67	return False	×
68
69	managers = {p: self.escalate(p, priority, **kwargs) for p in persons}	×
70	return self.add_couples(managers, bug_data)	×
71
72	def add_couples(self, managers, bug_data):	1✔
73	for person, manager in managers.items():	×
74	person = self.people.get_moz_mail(person)	×
75
76	if manager in self.data:	×
77	data = self.data[manager]	×
78	else:
79	self.data[manager] = data = {}	×
80
81	if person in data:	×
82	data[person].append(bug_data)	×
83	else:
84	data[person] = [bug_data]	×
85
86	return True	×
87
88	def nag_template(self):	1✔
89	return self.name() + "_nag.html"	×
90
91	def nag_preamble(self):	1✔
92	return None	×
93
94	def get_extra_for_nag_template(self):	1✔
95	return {}	×
96
97	def columns_nag(self):	1✔
98	return None	×
99
100	def sort_columns_nag(self):	1✔
101	return None	×
102
103	def _is_in_list(self, mail, _list):	1✔
104	for manager in _list:	×
105	if self.people.is_under(mail, manager):	×
106	return True	×
107	return False	×
108
109	def is_under(self, mail):	1✔
110	if not self.white_list:	×
111	if not self.black_list:	×
112	return True	×
113	return not self._is_in_list(mail, self.black_list)	×
114	if not self.black_list:	×
115	return self._is_in_list(mail, self.white_list)	×
116	return self._is_in_list(mail, self.white_list) and not self._is_in_list(	×
117	mail, self.black_list
118	)
119
120	def add_triage_owner(self, owners, real_owner):	1✔
121	if self.round_robin is None:	×
122	return	×
123
124	if not isinstance(owners, list):	×
125	owners = [owners]	×
126	for owner in owners:	×
127	person = self.people.get_moz_mail(owner)	×
128	if person not in self.triage_owners_components:	×
129	self.triage_owners_components[person] = set(	×
130	self.round_robin.get_components_for_triager(owner)
131	)
132	else:
133	self.triage_owners_components[	×
134	person
135	] \|= self.round_robin.get_components_for_triager(owner)
136
137	def get_query_url_for_components(self, components):	1✔
138	params = copy.deepcopy(self.query_params)	×
139	for field in ["include_fields", "product", "component", "bug_id"]:	×
140	if field in params:	×
141	del params[field]	×
142
143	utils.add_prod_comp_to_query(params, components)	×
144	url = utils.get_bz_search_url(params)	×
145
146	return url	×
147
148	def organize_nag(self, bugs):	1✔
149	columns = self.columns_nag()	×
150	if columns is None:	×
151	columns = self.columns()	×
152	key = self.sort_columns_nag()	×
153	if key is None:	×
154	key = self.sort_columns()	×
155
156	return utils.organize(bugs, columns, key=key)	×
157
158	def send_mails(self, title, dryrun=False):	1✔
159	if not self.send_nag_mail:	×
160	return	×
161
162	env = Environment(loader=FileSystemLoader("templates"))	×
163	common = env.get_template("common.html")	×
164	login_info = utils.get_login_info()	×
165	From = Nag.get_from()	×
166	Default_Cc = self.get_cc()	×
167	mails = self.prepare_mails()	×
168
169	for m in mails:	×
170	Cc = Default_Cc \| m["management_chain"]	×
171	if m["manager"]:	×
172	Cc.add(m["manager"])	×
173	body = common.render(message=m["body"], query_url=None)	×
174	receivers = set(m["to"]) \| set(Cc)	×
175	status = "Success"	×
176	try:	×
177	mail.send(	×
178	From,
179	sorted(m["to"]),
180	title,
181	body,
182	Cc=sorted(Cc),
183	html=True,
184	login=login_info,
185	dryrun=dryrun,
186	)
187	except Exception:	×
188	logger.exception("Rule {}".format(self.name()))	×
189	status = "Failure"	×
190
191	db.Email.add(self.name(), receivers, "individual", status)	×
192
193	def prepare_mails(self):	1✔
194	if not self.data:	×
195	return []	×
196
197	template = self.nag_template()	×
198	if not template:	×
199	return []	×
200
201	# If we escalating only to hierarchical mangers, we should always have a
202	# management chain.
203	fail_on_missing_mgmt_chain = self.escalation.is_hierarchical_escalation_only()	×
204
205	extra = self.get_extra_for_nag_template()	×
206	env = Environment(loader=FileSystemLoader("templates"))	×
207	template = env.get_template(template)	×
208	mails = []	×
209	for manager, info in self.data.items():	×
210	# The same bug can be several times in the list
211	# because we send an email to a team.
212	added_bug_ids = set()	×
213
214	data = []	×
215	To = sorted(info.keys())	×
216	components = set()	×
217	management_chain = set()	×
218	for person in To:	×
219	data += [	×
220	bug_data
221	for bug_data in info[person]
222	if bug_data["id"] not in added_bug_ids
223	]
224	added_bug_ids.update(bug_data["id"] for bug_data in info[person])	×
225	if person in self.triage_owners_components:	×
226	components \|= self.triage_owners_components[person]	×
227
228	management_chain \|= self.people.get_management_chain_mails(	×
229	person, manager, fail_on_missing_mgmt_chain
230	)
231
232	if components:	×
233	query_url = self.get_query_url_for_components(sorted(components))	×
234	else:
235	query_url = None	×
236
237	body = template.render(	×
238	date=self.nag_date,
239	extra=extra,
240	plural=utils.plural,
241	enumerate=enumerate,
242	data=self.organize_nag(data),
243	nag=True,
244	query_url_nag=utils.shorten_long_bz_url(query_url),
245	table_attrs=self.get_config("table_attrs"),
246	nag_preamble=self.nag_preamble(),
247	)
248
249	m = {	×
250	"manager": manager,
251	"management_chain": management_chain,
252	"to": set(To),
253	"body": body,
254	}
255	mails.append(m)	×
256
257	return mails	×
258
259	def reorganize_to_bag(self, data):	1✔
260	return data	×

mozilla / relman-auto-nag / #4403

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous