#4350

pending completion

Build # #4350

Build Type

push

coveralls-python

Committed by

web-flow

Commit Message

Update the protocol from git to https in Readme file (#1984)

Run Details

563 of 3079 branches covered (18.29%)

1795 of 7971 relevant lines covered (22.52%)

0.23 hits per line

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

84.34

/auto_nag/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.relativedelta import relativedelta
from icalendar import Calendar as iCalendar
from libmozdata import utils as lmdutils

from auto_nag import utils
from auto_nag.people import People


class InvalidCalendar(Exception):
    pass


class BadFallback(Exception):
    pass


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

mozilla / relman-auto-nag / #4350

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