#4853

Committed 15 Dec 2023 03:26AM CUT coverage: 21.899% (+0.02%) from 21.877%

Build # #4853

Build Type

push

coveralls-python

Committed by

suhaibmujahid

Commit Message

[round_robin] Do not crash on invalid duty start dates

Run Details

716 of 3590 branches covered (0.0%)

5 of 8 new or added lines in 2 files covered. (62.5%)

1928 of 8804 relevant lines covered (21.9%)

0.22 hits per line

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

85.27

/bugbot/round_robin_calendar.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 json
import os
import re
from bisect import bisect_left
from json.decoder import JSONDecodeError

import recurring_ical_events
import requests
from dateutil.parser import ParserError
from dateutil.relativedelta import relativedelta
from icalendar import Calendar as iCalendar
from libmozdata import utils as lmdutils

from bugbot import utils
from bugbot.people import People


class InvalidCalendar(Exception):
    pass


class BadFallback(Exception):
    pass


class InvalidDateError(ParserError):
    """Raised when a date in a rotation calendar is invalid"""


class Calendar:
    def __init__(self, fallback, team_name, people=None):
        self.people = People.get_instance() if people is None else people
        self.fallback = fallback
        self.fb_bzmail = self.people.get_bzmail_from_name(self.fallback)
        self.fb_mozmail = self.people.get_moz_mail(self.fb_bzmail)
        self.team_name = team_name
        self.team = []
        self.cache = {}

    def get_fallback(self):
        return self.fallback

    def get_fallback_bzmail(self):
        if not self.fb_bzmail:
            raise BadFallback("'{}' is an invalid fallback".format(self.fallback))
        return self.fb_bzmail

    def get_fallback_mozmail(self):
        if not self.fb_mozmail:
            raise BadFallback("'{}' is an invalid fallback".format(self.fallback))
        return self.fb_mozmail

    def get_team_name(self):
        return self.team_name

    def get_persons(self, date):
        return []

    def set_team(self, team, triagers):
        for p in team:
            if p in triagers and "bzmail" in triagers[p]:
                bzmail = triagers[p]["bzmail"]
            else:
                bzmail = self.people.get_bzmail_from_name(p)
            self.team.append((p, bzmail))

    @staticmethod
    def get(url, fallback, team_name, people=None):
        data = None
        if url.startswith("private://"):
            name = url.split("//", 1)[1]
            url = utils.get_private()[name]

        if url.startswith("http"):
            r = requests.get(url)
            data = r.text
        elif os.path.isfile(url):
            with open(url, "r") as In:
                data = In.read()
        else:
            data = url

        if data is None:
            raise InvalidCalendar("Cannot read calendar: {}".format(url))

        try:
            cal = json.loads(data)
            return JSONCalendar(cal, fallback, team_name, people=people)
        except JSONDecodeError:
            try:
                return ICSCalendar(data, fallback, team_name, people=people)
            except ValueError:
                raise InvalidCalendar(
                    f"Cannot decode calendar: {url} for team {team_name}"
                )

    def __str__(self):
        return f"""Round robin calendar:
team name: {self.team_name}
fallback: {self.fallback}, bz: {self.fb_bzmail}, moz: {self.fb_mozmail}
team: {self.team}"""

    def __repr__(self):
        return self.__str__()


class JSONCalendar(Calendar):
    def __init__(self, cal, fallback, team_name, people=None):
        super().__init__(fallback, team_name, people=people)
        start_dates = cal.get("duty-start-dates", {})
        if start_dates:
            try:
                dates = sorted((lmdutils.get_date_ymd(d), d) for d in start_dates)
            except ParserError as err:
                raise InvalidDateError(
                    f"Invalid duty start date for the {team_name} team: {err}"
                ) from err
            self.set_team(
                list(start_dates[d] for _, d in dates), cal.get("triagers", {})
            )
            self.dates = [d for d, _ in dates]
            cycle = self.guess_cycle()
            self.dates.append(self.dates[-1] + relativedelta(days=cycle))
            self.team.append(None)
        else:
            triagers = cal["triagers"]
            self.set_team(triagers.keys(), triagers)
            self.dates = []

    def get_persons(self, date):
        date = lmdutils.get_date_ymd(date)
        if date in self.cache:
            return self.cache[date]

        if not self.dates:
            # no dates so only triagers
            return self.team

        i = bisect_left(self.dates, date)
        if i == len(self.dates):
            self.cache[date] = []
            return []

        if date == self.dates[i]:
            person = self.team[i][0]
        else:
            person = self.team[i - 1][0] if i != 0 else self.team[0][0]

        self.cache[date] = res = [(person, self.people.get_bzmail_from_name(person))]

        return res

    def guess_cycle(self):
        diffs = [(x - y).days for x, y in zip(self.dates[1:], self.dates[:-1])]
        mean = sum(diffs) / len(diffs)
        return int(round(mean))


class ICSCalendar(Calendar):
    # The summary can be "[Gfx Triage] Foo Bar" or just "Foo Bar"
    SUM_PAT = re.compile(r"\s*(?:\[[^\]]*\])?\s*(.*)")

    def __init__(self, cal, fallback, team_name, people=None):
        super().__init__(fallback, team_name, people=people)
        self.cal = iCalendar.from_ical(cal)

    def get_person(self, p):
        g = ICSCalendar.SUM_PAT.match(p)
        if g:
            p = g.group(1)
            p = p.strip()
        return p

    def get_persons(self, date):
        date = lmdutils.get_date_ymd(date)
        if date in self.cache:
            return self.cache[date]

        events = recurring_ical_events.of(self.cal).between(date, date)
        persons = [self.get_person(event["SUMMARY"]) for event in events]
        self.cache[date] = res = [
            (person, self.people.get_bzmail_from_name(person)) for person in persons
        ]

        return res

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 json	1✔
6	import os	1✔
7	import re	1✔
8	from bisect import bisect_left	1✔
9	from json.decoder import JSONDecodeError	1✔
10
11	import recurring_ical_events	1✔
12	import requests	1✔
13	from dateutil.parser import ParserError	1✔
14	from dateutil.relativedelta import relativedelta	1✔
15	from icalendar import Calendar as iCalendar	1✔
16	from libmozdata import utils as lmdutils	1✔
17
18	from bugbot import utils	1✔
19	from bugbot.people import People	1✔
20
21
22	class InvalidCalendar(Exception):	1✔
23	pass	1✔
24
25
26	class BadFallback(Exception):	1✔
27	pass	1✔
28
29
30	class InvalidDateError(ParserError):	1✔
31	"""Raised when a date in a rotation calendar is invalid"""
32
33
34	class Calendar:	1✔
35	def __init__(self, fallback, team_name, people=None):	1✔
36	self.people = People.get_instance() if people is None else people	1✔
37	self.fallback = fallback	1✔
38	self.fb_bzmail = self.people.get_bzmail_from_name(self.fallback)	1✔
39	self.fb_mozmail = self.people.get_moz_mail(self.fb_bzmail)	1✔
40	self.team_name = team_name	1✔
41	self.team = []	1✔
42	self.cache = {}	1✔
43
44	def get_fallback(self):	1✔
45	return self.fallback	1✔
46
47	def get_fallback_bzmail(self):	1✔
48	if not self.fb_bzmail:	1!
49	raise BadFallback("'{}' is an invalid fallback".format(self.fallback))	×
50	return self.fb_bzmail	1✔
51
52	def get_fallback_mozmail(self):	1✔
53	if not self.fb_mozmail:	1✔
54	raise BadFallback("'{}' is an invalid fallback".format(self.fallback))	1✔
55	return self.fb_mozmail	1✔
56
57	def get_team_name(self):	1✔
58	return self.team_name	1✔
59
60	def get_persons(self, date):	1✔
61	return []	×
62
63	def set_team(self, team, triagers):	1✔
64	for p in team:	1✔
65	if p in triagers and "bzmail" in triagers[p]:	1!
66	bzmail = triagers[p]["bzmail"]	×
67	else:
68	bzmail = self.people.get_bzmail_from_name(p)	1✔
69	self.team.append((p, bzmail))	1✔
70
71	@staticmethod	1✔
72	def get(url, fallback, team_name, people=None):	1✔
73	data = None	1✔
74	if url.startswith("private://"):	1!
75	name = url.split("//", 1)[1]	×
76	url = utils.get_private()[name]	×
77
78	if url.startswith("http"):	1!
79	r = requests.get(url)	×
80	data = r.text	×
81	elif os.path.isfile(url):	1!
82	with open(url, "r") as In:	1✔
83	data = In.read()	1✔
84	else:
85	data = url	×
86
87	if data is None:	1!
88	raise InvalidCalendar("Cannot read calendar: {}".format(url))	×
89
90	try:	1✔
91	cal = json.loads(data)	1✔
92	return JSONCalendar(cal, fallback, team_name, people=people)	1✔
93	except JSONDecodeError:	1✔
94	try:	1✔
95	return ICSCalendar(data, fallback, team_name, people=people)	1✔
96	except ValueError:	×
97	raise InvalidCalendar(	×
98	f"Cannot decode calendar: {url} for team {team_name}"
99	)
100
101	def __str__(self):	1✔
102	return f"""Round robin calendar:	×
103	team name: {self.team_name}
104	fallback: {self.fallback}, bz: {self.fb_bzmail}, moz: {self.fb_mozmail}
105	team: {self.team}"""
106
107	def __repr__(self):	1✔
108	return self.__str__()	×
109
110
111	class JSONCalendar(Calendar):	1✔
112	def __init__(self, cal, fallback, team_name, people=None):	1✔
113	super().__init__(fallback, team_name, people=people)	1✔
114	start_dates = cal.get("duty-start-dates", {})	1✔
115	if start_dates:	1!
116	try:	1✔
117	dates = sorted((lmdutils.get_date_ymd(d), d) for d in start_dates)	1✔
NEW 118	except ParserError as err:	×
NEW 119	raise InvalidDateError(	×
120	f"Invalid duty start date for the {team_name} team: {err}"
121	) from err
122	self.set_team(	1✔
123	list(start_dates[d] for _, d in dates), cal.get("triagers", {})
124	)
125	self.dates = [d for d, _ in dates]	1✔
126	cycle = self.guess_cycle()	1✔
127	self.dates.append(self.dates[-1] + relativedelta(days=cycle))	1✔
128	self.team.append(None)	1✔
129	else:
130	triagers = cal["triagers"]	×
131	self.set_team(triagers.keys(), triagers)	×
132	self.dates = []	×
133
134	def get_persons(self, date):	1✔
135	date = lmdutils.get_date_ymd(date)	1✔
136	if date in self.cache:	1✔
137	return self.cache[date]	1✔
138
139	if not self.dates:	1!
140	# no dates so only triagers
141	return self.team	×
142
143	i = bisect_left(self.dates, date)	1✔
144	if i == len(self.dates):	1✔
145	self.cache[date] = []	1✔
146	return []	1✔
147
148	if date == self.dates[i]:	1✔
149	person = self.team[i][0]	1✔
150	else:
151	person = self.team[i - 1][0] if i != 0 else self.team[0][0]	1✔
152
153	self.cache[date] = res = [(person, self.people.get_bzmail_from_name(person))]	1✔
154
155	return res	1✔
156
157	def guess_cycle(self):	1✔
158	diffs = [(x - y).days for x, y in zip(self.dates[1:], self.dates[:-1])]	1✔
159	mean = sum(diffs) / len(diffs)	1✔
160	return int(round(mean))	1✔
161
162
163	class ICSCalendar(Calendar):	1✔
164	# The summary can be "[Gfx Triage] Foo Bar" or just "Foo Bar"
165	SUM_PAT = re.compile(r"\s(?:\[[^\]]\])?\s(.)")	1✔
166
167	def __init__(self, cal, fallback, team_name, people=None):	1✔
168	super().__init__(fallback, team_name, people=people)	1✔
169	self.cal = iCalendar.from_ical(cal)	1✔
170
171	def get_person(self, p):	1✔
172	g = ICSCalendar.SUM_PAT.match(p)	1✔
173	if g:	1!
174	p = g.group(1)	1✔
175	p = p.strip()	1✔
176	return p	1✔
177
178	def get_persons(self, date):	1✔
179	date = lmdutils.get_date_ymd(date)	1✔
180	if date in self.cache:	1✔
181	return self.cache[date]	1✔
182
183	events = recurring_ical_events.of(self.cal).between(date, date)	1✔
184	persons = [self.get_person(event["SUMMARY"]) for event in events]	1✔
185	self.cache[date] = res = [	1✔
186	(person, self.people.get_bzmail_from_name(person)) for person in persons
187	]
188
189	return res	1✔

mozilla / relman-auto-nag / #4853

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