1595

pending completion

Build # 1595

Build Type

Pull #756

travis-ci

Committed by

web-flow

Commit Message

Update auto_nag/scripts/tracked_bad_severity.py

Co-Authored-By: Sylvestre Ledru <sledru@mozilla.com>

Pull Request Pull Request #756: tracked_bad_severity: add a comment to explain

Run Details

369 of 1802 branches covered (20.48%)

1 of 1 new or added line in 1 file covered. (100.0%)

1283 of 4993 relevant lines covered (25.7%)

0.26 hits per line

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

35.91

/auto_nag/utils.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
import datetime
import json
import os
import random
import re

import dateutil.parser
import humanize
import pytz
import requests
import six
from dateutil.relativedelta import relativedelta
from libmozdata import release_calendar as rc
from libmozdata import utils as lmdutils
from libmozdata import versions as lmdversions
from libmozdata.bugzilla import Bugzilla
from libmozdata.hgmozilla import Mercurial

try:
    from urllib.parse import urlencode
except ImportError:
    from urllib import urlencode


_CONFIG = None
_CYCLE_SPAN = None
_MERGE_DAY = None
_TRIAGE_OWNERS = None
_DEFAULT_ASSIGNEES = None
_CURRENT_VERSIONS = None
_CONFIG_PATH = "./auto_nag/scripts/configs/"


BZ_FIELD_PAT = re.compile(r"^[fovj]([0-9]+)$")
PAR_PAT = re.compile(r"\([^\)]*\)")
BRA_PAT = re.compile(r"\[[^\]]*\]")
DIA_PAT = re.compile("<[^>]*>")
UTC_PAT = re.compile(r"UTC\+[^ \t]*")
COL_PAT = re.compile(":[^:]*")
BACKOUT_PAT = re.compile("^back(s|(ed))?[ \t]*out", re.I)
BUG_PAT = re.compile(r"^bug[s]?[ \t]*([0-9]+)", re.I)


def _get_config():
    global _CONFIG
    if _CONFIG is None:
        try:
            with open(_CONFIG_PATH + "/tools.json", "r") as In:
                _CONFIG = json.load(In)
        except IOError:
            _CONFIG = {}
    return _CONFIG


def get_config(name, entry, default=None):
    conf = _get_config()
    if name not in conf:
        name = "common"
    tool_conf = conf[name]
    if entry in tool_conf:
        return tool_conf[entry]
    tool_conf = conf["common"]
    return tool_conf.get(entry, default)


def init_random():
    now = datetime.datetime.utcnow()
    now = now.timestamp()
    random.seed(now)


def get_signatures(sgns):
    if not sgns:
        return set()

    res = set()
    sgns = map(lambda x: x.strip(), sgns.split("[@"))
    for s in filter(None, sgns):
        try:
            i = s.rindex("]")
            res.add(s[:i].strip())
        except ValueError:
            res.add(s)
    return res


def add_signatures(old, new):
    added_sgns = "[@ " + "]\n[@ ".join(sorted(new)) + "]"
    if old:
        return old + "\n" + added_sgns
    return added_sgns


def get_empty_assignees(params, negation=False):
    n = get_last_field_num(params)
    n = int(n)
    params.update(
        {
            "j" + str(n): "OR",
            "f" + str(n): "OP",
            "f" + str(n + 1): "assigned_to",
            "o" + str(n + 1): "equals",
            "v" + str(n + 1): "nobody@mozilla.org",
            "f" + str(n + 2): "assigned_to",
            "o" + str(n + 2): "regexp",
            "v" + str(n + 2): r"^.*\.bugs$",
            "f" + str(n + 3): "assigned_to",
            "o" + str(n + 3): "isempty",
            "f" + str(n + 4): "CP",
        }
    )
    if negation:
        params["n" + str(n)] = 1

    return params


def is_no_assignee(mail):
    return mail == "nobody@mozilla.org" or mail.endswith(".bugs") or mail == ""


def get_login_info():
    with open(_CONFIG_PATH + "config.json", "r") as In:
        return json.load(In)


def get_private():
    with open(_CONFIG_PATH + "config.json", "r") as In:
        return json.load(In)["private"]


def plural(sword, data, pword=""):
    if isinstance(data, six.integer_types):
        p = data != 1
    else:
        p = len(data) != 1
    if not p:
        return sword
    if pword:
        return pword
    return sword + "s"


def search_prev_merge(beta):
    tables = rc.get_all()

    # the first table is the future and the second is the recent past
    table = tables[1]
    central = table[0].index("Central")
    central = rc.get_versions(table[1][central])[0][0]

    # just check consistency
    assert beta == central

    merge = table[0].index("Merge Date")

    return lmdutils.get_date_ymd(table[1][merge])


def get_cycle_span():
    global _CYCLE_SPAN
    if _CYCLE_SPAN is None:
        cal = get_release_calendar()
        now = lmdutils.get_date_ymd("today")
        cycle = None
        for i, c in enumerate(cal):
            if now < c["merge"]:
                if i == 0:
                    cycle = [search_prev_merge(c["beta"]), c["merge"]]
                else:
                    cycle = [cal[i - 1]["merge"], c["merge"]]
                break
        if cycle:
            _CYCLE_SPAN = "-".join(x.strftime("%Y%m%d") for x in cycle)

    return _CYCLE_SPAN


def get_next_release_date():
    return rc.get_next_release_date()


def get_release_calendar():
    return rc.get_calendar()


def get_merge_day():
    global _MERGE_DAY
    if _MERGE_DAY is None:
        cal = get_release_calendar()
        _MERGE_DAY = cal[0]["merge"]
    return _MERGE_DAY


def is_merge_day():
    next_merge = get_merge_day()
    today = lmdutils.get_date_ymd("today")

    return next_merge == today


def get_report_bugs(channel, op="+"):
    url = "https://bugzilla.mozilla.org/page.cgi?id=release_tracking_report.html"
    params = {
        "q": "approval-mozilla-{}:{}:{}:0:and:".format(channel, op, get_cycle_span())
    }

    # allow_redirects=False avoids to load the data
    # and we'll just get the redirected url to get all the bug ids we need
    r = requests.get(url, params=params, allow_redirects=False)

    # something like https://bugzilla.mozilla.org/buglist.cgi?bug_id=1493711,1502766,1499908
    url = r.headers["Location"]

    return url.split("=")[1].split(",")


def get_flag(version, name, channel):
    if name in ["status", "tracking"]:
        if channel == "esr":
            return "cf_{}_firefox_esr{}".format(name, version)
        return "cf_{}_firefox{}".format(name, version)
    elif name == "approval":
        if channel == "esr":
            return "approval-mozilla-esr{}".format(version)
        return "approval-mozilla-{}".format(channel)


def get_needinfo(bug, days=-1):
    now = datetime.datetime.utcnow().replace(tzinfo=pytz.utc)
    for flag in bug.get("flags", []):
        if flag.get("name", "") == "needinfo" and flag["status"] == "?":
            date = flag["modification_date"]
            date = dateutil.parser.parse(date)
            if (now - date).days >= days:
                yield flag


def get_last_field_num(params):
    s = set()
    for k in params.keys():
        m = BZ_FIELD_PAT.match(k)
        if m:
            s.add(int(m.group(1)))

    x = max(s) + 1 if s else 1
    return str(x)


def add_prod_comp_to_query(params, prod_comp):
    n = int(get_last_field_num(params))
    params[f"j{n}"] = "OR"
    params[f"f{n}"] = "OP"
    n += 1
    for pc in prod_comp:
        prod, comp = pc.split("::")
        params[f"j{n}"] = "AND"
        params[f"f{n}"] = "OP"
        n += 1
        params[f"f{n}"] = "product"
        params[f"o{n}"] = "equals"
        params[f"v{n}"] = prod
        n += 1
        params[f"f{n}"] = "component"
        params[f"o{n}"] = "equals"
        params[f"v{n}"] = comp
        n += 1
        params[f"f{n}"] = "CP"
        n += 1
    params[f"f{n}"] = "CP"


def get_bz_search_url(params):
    return "https://bugzilla.mozilla.org/buglist.cgi?" + urlencode(params, doseq=True)


def has_bot_set_ni(bug):
    bot = get_config("common", "bot_bz_mail")
    for flag in get_needinfo(bug):
        if flag["setter"] in bot:
            return True
    return False


def get_triage_owners():
    global _TRIAGE_OWNERS
    if _TRIAGE_OWNERS is not None:
        return _TRIAGE_OWNERS

    # accessible is the union of:
    #  selectable (all product we can see)
    #  enterable (all product a user can file bugs into).
    prods = get_config("common", "products")
    url = "https://bugzilla.mozilla.org/rest/product"
    params = {
        "type": "accessible",
        "include_fields": ["name", "components.name", "components.triage_owner"],
        "names": prods,
    }
    r = requests.get(url, params=params)
    products = r.json()["products"]
    _TRIAGE_OWNERS = {}
    for prod in products:
        prod_name = prod["name"]
        for comp in prod["components"]:
            owner = comp["triage_owner"]
            if owner and not is_no_assignee(owner):
                comp_name = comp["name"]
                pc = f"{prod_name}::{comp_name}"
                if owner not in _TRIAGE_OWNERS:
                    _TRIAGE_OWNERS[owner] = [pc]
                else:
                    _TRIAGE_OWNERS[owner].append(pc)
    return _TRIAGE_OWNERS


def get_default_assignees():
    global _DEFAULT_ASSIGNEES
    if _DEFAULT_ASSIGNEES is not None:
        return _DEFAULT_ASSIGNEES

    # accessible is the union of:
    #  selectable (all product we can see)
    #  enterable (all product a user can file bugs into).
    prods = get_config("common", "products")
    url = "https://bugzilla.mozilla.org/rest/product"
    params = {
        "type": "accessible",
        "include_fields": ["name", "components.name", "components.default_assigned_to"],
        "names": prods,
    }
    r = requests.get(url, params=params)
    products = r.json()["products"]
    _DEFAULT_ASSIGNEES = {}
    for prod in products:
        prod_name = prod["name"]
        _DEFAULT_ASSIGNEES[prod_name] = dap = {}
        for comp in prod["components"]:
            comp_name = comp["name"]
            assignee = comp["default_assigned_to"]
            dap[comp_name] = assignee
    return _DEFAULT_ASSIGNEES


def organize(bugs, columns, key=None):
    if isinstance(bugs, dict):
        # we suppose that the values are the bugdata dict
        bugs = bugs.values()

    def identity(x):
        return x

    def bugid_key(x):
        return -int(x)

    lambdas = {"id": bugid_key}

    def mykey(p):
        return tuple(lambdas.get(c, identity)(x) for x, c in zip(p, columns))

    if len(columns) >= 2:
        res = [tuple(info[c] for c in columns) for info in bugs]
    else:
        c = columns[0]
        res = [info[c] for info in bugs]

    return sorted(res, key=mykey if not key else key)


def merge_bz_changes(c1, c2):
    if not c1:
        return c2
    if not c2:
        return c1

    assert set(c1.keys()).isdisjoint(
        c2.keys()
    ), "Merge changes with common keys is not a good idea"
    c = copy.deepcopy(c1)
    c.update(c2)

    return c


def is_test_file(path):
    e = os.path.splitext(path)[1][1:].lower()
    return "test" in path and e not in {"ini", "list", "in", "py", "json", "manifest"}


def get_better_name(name):
    if not name:
        return ""

    def repl(m):
        if m.start(0) == 0:
            return m.group(0)
        return ""

    if name.startswith("Nobody;"):
        s = "Nobody"
    else:
        s = PAR_PAT.sub("", name)
        s = BRA_PAT.sub("", s)
        s = DIA_PAT.sub("", s)
        s = COL_PAT.sub(repl, s)
        s = UTC_PAT.sub("", s)
        s = s.strip()
        if s.startswith(":"):
            s = s[1:]
    return s.encode("utf-8").decode("utf-8")


def is_backout(json):
    return json.get("backedoutby", "") != "" or bool(BACKOUT_PAT.search(json["desc"]))


def get_pushlog(startdate, enddate, channel="nightly"):
    """Get the pushlog from hg.mozilla.org"""
    # Get the pushes where startdate <= pushdate <= enddate
    # pushlog uses strict inequality, it's why we add +/- 1 second
    fmt = "%Y-%m-%d %H:%M:%S"
    startdate -= relativedelta(seconds=1)
    startdate = startdate.strftime(fmt)
    enddate += relativedelta(seconds=1)
    enddate = enddate.strftime(fmt)
    url = "{}/json-pushes".format(Mercurial.get_repo_url(channel))
    r = requests.get(
        url,
        params={"startdate": startdate, "enddate": enddate, "version": 2, "full": 1},
    )
    return r.json()


def get_bugs_from_desc(desc):
    """Get a bug number from the patch description"""
    return BUG_PAT.findall(desc)


def get_bugs_from_pushlog(startdate, enddate, channel="nightly"):
    pushlog = get_pushlog(startdate, enddate, channel=channel)
    bugs = set()
    for push in pushlog["pushes"].values():
        for chgset in push["changesets"]:
            if chgset.get("backedoutby", "") != "":
                continue
            desc = chgset["desc"]
            for bug in get_bugs_from_desc(desc):
                bugs.add(bug)
    return bugs


def get_checked_versions():
    # There are different reasons to not return versions:
    # i) we're merge day: the versions are changing
    # ii) not consecutive versions numbers
    # iii) bugzilla updated nightly version but p-d is not updated
    if is_merge_day():
        return {}

    versions = lmdversions.get(base=True)
    versions["central"] = versions["nightly"]

    v = [versions[k] for k in ["release", "beta", "central"]]
    versions = {k: str(v) for k, v in versions.items()}

    if v[0] + 2 == v[1] + 1 == v[2]:
        nightly_bugzilla = get_nightly_version_from_bz()
        if v[2] != nightly_bugzilla:
            from . import logger

            logger.info("Versions mismatch between Bugzilla and product-details")
            return {}
        return versions

    from . import logger

    logger.info("Not consecutive versions in product/details")
    return {}


def get_info_from_hg(json):
    res = {}
    push = json["pushdate"][0]
    push = datetime.datetime.utcfromtimestamp(push)
    push = lmdutils.as_utc(push)
    res["date"] = lmdutils.get_date_str(push)
    res["backedout"] = json.get("backedoutby", "") != ""
    m = BUG_PAT.search(json["desc"])
    res["bugid"] = m.group(1) if m else ""

    return res


def bz_ignore_case(s):
    return "[" + "][".join(c + c.upper() for c in s) + "]"


def check_product_component(data, bug):
    prod = bug["product"]
    comp = bug["component"]
    pc = prod + "::" + comp
    return pc in data or comp in data


def get_components(data):
    res = []
    for comp in data:
        if "::" in comp:
            _, comp = comp.split("::", 1)
        res.append(comp)
    return res


def get_products_components(data):
    prods = set()
    comps = set()
    for pc in data:
        if "::" in pc:
            p, c = pc.split("::", 1)
            prods.add(p)
        else:
            c = pc
        comps.add(c)
    return prods, comps


def ireplace(old, repl, text):
    return re.sub("(?i)" + re.escape(old), lambda m: repl, text)


def get_human_lag(date):
    today = pytz.utc.localize(datetime.datetime.utcnow())
    dt = dateutil.parser.parse(date)

    return humanize.naturaldelta(today - dt)


def get_nightly_version_from_bz():
    def bug_handler(bug, data):
        status = "cf_status_firefox"
        N = len(status)
        for k in bug.keys():
            if k.startswith(status):
                k = k[N:]
                if k.isdigit():
                    data.append(int(k))

    data = []
    Bugzilla(bugids=["1234567"], bughandler=bug_handler, bugdata=data).get_data().wait()

    return max(data)


def nice_round(val):
    return int(round(100 * val))

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	import datetime	1✔
7	import json	1✔
8	import os	1✔
9	import random	1✔
10	import re	1✔
11
12	import dateutil.parser	1✔
13	import humanize	1✔
14	import pytz	1✔
15	import requests	1✔
16	import six	1✔
17	from dateutil.relativedelta import relativedelta	1✔
18	from libmozdata import release_calendar as rc	1✔
19	from libmozdata import utils as lmdutils	1✔
20	from libmozdata import versions as lmdversions	1✔
21	from libmozdata.bugzilla import Bugzilla	1✔
22	from libmozdata.hgmozilla import Mercurial	1✔
23
24	try:	1✔
25	from urllib.parse import urlencode	1✔
26	except ImportError:	×
27	from urllib import urlencode	×
28
29
30	_CONFIG = None	1✔
31	_CYCLE_SPAN = None	1✔
32	_MERGE_DAY = None	1✔
33	_TRIAGE_OWNERS = None	1✔
34	_DEFAULT_ASSIGNEES = None	1✔
35	_CURRENT_VERSIONS = None	1✔
36	_CONFIG_PATH = "./auto_nag/scripts/configs/"	1✔
37
38
39	BZ_FIELD_PAT = re.compile(r"^[fovj]([0-9]+)$")	1✔
40	PAR_PAT = re.compile(r"\([^\)]*\)")	1✔
41	BRA_PAT = re.compile(r"\[[^\]]*\]")	1✔
42	DIA_PAT = re.compile("<[^>]*>")	1✔
43	UTC_PAT = re.compile(r"UTC\+[^ \t]*")	1✔
44	COL_PAT = re.compile(":[^:]*")	1✔
45	BACKOUT_PAT = re.compile("^back(s\|(ed))?[ \t]*out", re.I)	1✔
46	BUG_PAT = re.compile(r"^bug[s]?[ \t]*([0-9]+)", re.I)	1✔
47
48
49	def _get_config():	1✔
50	global _CONFIG
51	if _CONFIG is None:	1✔
52	try:	1✔
53	with open(_CONFIG_PATH + "/tools.json", "r") as In:	1✔
54	_CONFIG = json.load(In)	1✔
55	except IOError:	×
56	_CONFIG = {}	×
57	return _CONFIG	1✔
58
59
60	def get_config(name, entry, default=None):	1✔
61	conf = _get_config()	1✔
62	if name not in conf:	1✔
63	name = "common"	1✔
64	tool_conf = conf[name]	1✔
65	if entry in tool_conf:	1✔
66	return tool_conf[entry]	1✔
67	tool_conf = conf["common"]	1✔
68	return tool_conf.get(entry, default)	1✔
69
70
71	def init_random():	1✔
72	now = datetime.datetime.utcnow()	1✔
73	now = now.timestamp()	1✔
74	random.seed(now)	1✔
75
76
77	def get_signatures(sgns):	1✔
78	if not sgns:	1!
79	return set()	×
80
81	res = set()	1✔
82	sgns = map(lambda x: x.strip(), sgns.split("[@"))	1✔
83	for s in filter(None, sgns):	1✔
84	try:	1✔
85	i = s.rindex("]")	1✔
86	res.add(s[:i].strip())	1✔
87	except ValueError:	×
88	res.add(s)	×
89	return res	1✔
90
91
92	def add_signatures(old, new):	1✔
93	added_sgns = "[@ " + "]\n[@ ".join(sorted(new)) + "]"	1✔
94	if old:	1✔
95	return old + "\n" + added_sgns	1✔
96	return added_sgns	1✔
97
98
99	def get_empty_assignees(params, negation=False):	1✔
100	n = get_last_field_num(params)	1✔
101	n = int(n)	1✔
102	params.update(	1✔
103	{
104	"j" + str(n): "OR",
105	"f" + str(n): "OP",
106	"f" + str(n + 1): "assigned_to",
107	"o" + str(n + 1): "equals",
108	"v" + str(n + 1): "nobody@mozilla.org",
109	"f" + str(n + 2): "assigned_to",
110	"o" + str(n + 2): "regexp",
111	"v" + str(n + 2): r"^.*\.bugs$",
112	"f" + str(n + 3): "assigned_to",
113	"o" + str(n + 3): "isempty",
114	"f" + str(n + 4): "CP",
115	}
116	)
117	if negation:	1!
118	params["n" + str(n)] = 1	×
119
120	return params	1✔
121
122
123	def is_no_assignee(mail):	1✔
124	return mail == "nobody@mozilla.org" or mail.endswith(".bugs") or mail == ""	1✔
125
126
127	def get_login_info():	1✔
128	with open(_CONFIG_PATH + "config.json", "r") as In:	×
129	return json.load(In)	×
130
131
132	def get_private():	1✔
133	with open(_CONFIG_PATH + "config.json", "r") as In:	×
134	return json.load(In)["private"]	×
135
136
137	def plural(sword, data, pword=""):	1✔
138	if isinstance(data, six.integer_types):	×
139	p = data != 1	×
140	else:
141	p = len(data) != 1	×
142	if not p:	×
143	return sword	×
144	if pword:	×
145	return pword	×
146	return sword + "s"	×
147
148
149	def search_prev_merge(beta):	1✔
150	tables = rc.get_all()	×
151
152	# the first table is the future and the second is the recent past
153	table = tables[1]	×
154	central = table[0].index("Central")	×
155	central = rc.get_versions(table[1][central])[0][0]	×
156
157	# just check consistency
158	assert beta == central	×
159
160	merge = table[0].index("Merge Date")	×
161
162	return lmdutils.get_date_ymd(table[1][merge])	×
163
164
165	def get_cycle_span():	1✔
166	global _CYCLE_SPAN
167	if _CYCLE_SPAN is None:	×
168	cal = get_release_calendar()	×
169	now = lmdutils.get_date_ymd("today")	×
170	cycle = None	×
171	for i, c in enumerate(cal):	×
172	if now < c["merge"]:	×
173	if i == 0:	×
174	cycle = [search_prev_merge(c["beta"]), c["merge"]]	×
175	else:
176	cycle = [cal[i - 1]["merge"], c["merge"]]	×
177	break	×
178	if cycle:	×
179	_CYCLE_SPAN = "-".join(x.strftime("%Y%m%d") for x in cycle)	×
180
181	return _CYCLE_SPAN	×
182
183
184	def get_next_release_date():	1✔
185	return rc.get_next_release_date()	×
186
187
188	def get_release_calendar():	1✔
189	return rc.get_calendar()	1✔
190
191
192	def get_merge_day():	1✔
193	global _MERGE_DAY
194	if _MERGE_DAY is None:	1✔
195	cal = get_release_calendar()	1✔
196	_MERGE_DAY = cal[0]["merge"]	1✔
197	return _MERGE_DAY	1✔
198
199
200	def is_merge_day():	1✔
201	next_merge = get_merge_day()	1✔
202	today = lmdutils.get_date_ymd("today")	1✔
203
204	return next_merge == today	1✔
205
206
207	def get_report_bugs(channel, op="+"):	1✔
208	url = "https://bugzilla.mozilla.org/page.cgi?id=release_tracking_report.html"	×
209	params = {	×
210	"q": "approval-mozilla-{}:{}:{}:0:and:".format(channel, op, get_cycle_span())
211	}
212
213	# allow_redirects=False avoids to load the data
214	# and we'll just get the redirected url to get all the bug ids we need
215	r = requests.get(url, params=params, allow_redirects=False)	×
216
217	# something like https://bugzilla.mozilla.org/buglist.cgi?bug_id=1493711,1502766,1499908
218	url = r.headers["Location"]	×
219
220	return url.split("=")[1].split(",")	×
221
222
223	def get_flag(version, name, channel):	1✔
224	if name in ["status", "tracking"]:	×
225	if channel == "esr":	×
226	return "cf_{}_firefox_esr{}".format(name, version)	×
227	return "cf_{}_firefox{}".format(name, version)	×
228	elif name == "approval":	×
229	if channel == "esr":	×
230	return "approval-mozilla-esr{}".format(version)	×
231	return "approval-mozilla-{}".format(channel)	×
232
233
234	def get_needinfo(bug, days=-1):	1✔
235	now = datetime.datetime.utcnow().replace(tzinfo=pytz.utc)	×
236	for flag in bug.get("flags", []):	×
237	if flag.get("name", "") == "needinfo" and flag["status"] == "?":	×
238	date = flag["modification_date"]	×
239	date = dateutil.parser.parse(date)	×
240	if (now - date).days >= days:	×
241	yield flag	×
242
243
244	def get_last_field_num(params):	1✔
245	s = set()	1✔
246	for k in params.keys():	1✔
247	m = BZ_FIELD_PAT.match(k)	1✔
248	if m:	1✔
249	s.add(int(m.group(1)))	1✔
250
251	x = max(s) + 1 if s else 1	1✔
252	return str(x)	1✔
253
254
255	def add_prod_comp_to_query(params, prod_comp):	1✔
256	n = int(get_last_field_num(params))	×
257	params[f"j{n}"] = "OR"	×
258	params[f"f{n}"] = "OP"	×
259	n += 1	×
260	for pc in prod_comp:	×
261	prod, comp = pc.split("::")	×
262	params[f"j{n}"] = "AND"	×
263	params[f"f{n}"] = "OP"	×
264	n += 1	×
265	params[f"f{n}"] = "product"	×
266	params[f"o{n}"] = "equals"	×
267	params[f"v{n}"] = prod	×
268	n += 1	×
269	params[f"f{n}"] = "component"	×
270	params[f"o{n}"] = "equals"	×
271	params[f"v{n}"] = comp	×
272	n += 1	×
273	params[f"f{n}"] = "CP"	×
274	n += 1	×
275	params[f"f{n}"] = "CP"	×
276
277
278	def get_bz_search_url(params):	1✔
279	return "https://bugzilla.mozilla.org/buglist.cgi?" + urlencode(params, doseq=True)	1✔
280
281
282	def has_bot_set_ni(bug):	1✔
283	bot = get_config("common", "bot_bz_mail")	×
284	for flag in get_needinfo(bug):	×
285	if flag["setter"] in bot:	×
286	return True	×
287	return False	×
288
289
290	def get_triage_owners():	1✔
291	global _TRIAGE_OWNERS
292	if _TRIAGE_OWNERS is not None:	×
293	return _TRIAGE_OWNERS	×
294
295	# accessible is the union of:
296	# selectable (all product we can see)
297	# enterable (all product a user can file bugs into).
298	prods = get_config("common", "products")	×
299	url = "https://bugzilla.mozilla.org/rest/product"	×
300	params = {	×
301	"type": "accessible",
302	"include_fields": ["name", "components.name", "components.triage_owner"],
303	"names": prods,
304	}
305	r = requests.get(url, params=params)	×
306	products = r.json()["products"]	×
307	_TRIAGE_OWNERS = {}	×
308	for prod in products:	×
309	prod_name = prod["name"]	×
310	for comp in prod["components"]:	×
311	owner = comp["triage_owner"]	×
312	if owner and not is_no_assignee(owner):	×
313	comp_name = comp["name"]	×
314	pc = f"{prod_name}::{comp_name}"	×
315	if owner not in _TRIAGE_OWNERS:	×
316	_TRIAGE_OWNERS[owner] = [pc]	×
317	else:
318	_TRIAGE_OWNERS[owner].append(pc)	×
319	return _TRIAGE_OWNERS	×
320
321
322	def get_default_assignees():	1✔
323	global _DEFAULT_ASSIGNEES
324	if _DEFAULT_ASSIGNEES is not None:	×
325	return _DEFAULT_ASSIGNEES	×
326
327	# accessible is the union of:
328	# selectable (all product we can see)
329	# enterable (all product a user can file bugs into).
330	prods = get_config("common", "products")	×
331	url = "https://bugzilla.mozilla.org/rest/product"	×
332	params = {	×
333	"type": "accessible",
334	"include_fields": ["name", "components.name", "components.default_assigned_to"],
335	"names": prods,
336	}
337	r = requests.get(url, params=params)	×
338	products = r.json()["products"]	×
339	_DEFAULT_ASSIGNEES = {}	×
340	for prod in products:	×
341	prod_name = prod["name"]	×
342	_DEFAULT_ASSIGNEES[prod_name] = dap = {}	×
343	for comp in prod["components"]:	×
344	comp_name = comp["name"]	×
345	assignee = comp["default_assigned_to"]	×
346	dap[comp_name] = assignee	×
347	return _DEFAULT_ASSIGNEES	×
348
349
350	def organize(bugs, columns, key=None):	1✔
351	if isinstance(bugs, dict):	×
352	# we suppose that the values are the bugdata dict
353	bugs = bugs.values()	×
354
355	def identity(x):	×
356	return x	×
357
358	def bugid_key(x):	×
359	return -int(x)	×
360
361	lambdas = {"id": bugid_key}	×
362
363	def mykey(p):	×
364	return tuple(lambdas.get(c, identity)(x) for x, c in zip(p, columns))	×
365
366	if len(columns) >= 2:	×
367	res = [tuple(info[c] for c in columns) for info in bugs]	×
368	else:
369	c = columns[0]	×
370	res = [info[c] for info in bugs]	×
371
372	return sorted(res, key=mykey if not key else key)	×
373
374
375	def merge_bz_changes(c1, c2):	1✔
376	if not c1:	×
377	return c2	×
378	if not c2:	×
379	return c1	×
380
381	assert set(c1.keys()).isdisjoint(	×
382	c2.keys()
383	), "Merge changes with common keys is not a good idea"
384	c = copy.deepcopy(c1)	×
385	c.update(c2)	×
386
387	return c	×
388
389
390	def is_test_file(path):	1✔
391	e = os.path.splitext(path)[1][1:].lower()	×
392	return "test" in path and e not in {"ini", "list", "in", "py", "json", "manifest"}	×
393
394
395	def get_better_name(name):	1✔
396	if not name:	×
397	return ""	×
398
399	def repl(m):	×
400	if m.start(0) == 0:	×
401	return m.group(0)	×
402	return ""	×
403
404	if name.startswith("Nobody;"):	×
405	s = "Nobody"	×
406	else:
407	s = PAR_PAT.sub("", name)	×
408	s = BRA_PAT.sub("", s)	×
409	s = DIA_PAT.sub("", s)	×
410	s = COL_PAT.sub(repl, s)	×
411	s = UTC_PAT.sub("", s)	×
412	s = s.strip()	×
413	if s.startswith(":"):	×
414	s = s[1:]	×
415	return s.encode("utf-8").decode("utf-8")	×
416
417
418	def is_backout(json):	1✔
419	return json.get("backedoutby", "") != "" or bool(BACKOUT_PAT.search(json["desc"]))	1✔
420
421
422	def get_pushlog(startdate, enddate, channel="nightly"):	1✔
423	"""Get the pushlog from hg.mozilla.org"""
424	# Get the pushes where startdate <= pushdate <= enddate
425	# pushlog uses strict inequality, it's why we add +/- 1 second
426	fmt = "%Y-%m-%d %H:%M:%S"	×
427	startdate -= relativedelta(seconds=1)	×
428	startdate = startdate.strftime(fmt)	×
429	enddate += relativedelta(seconds=1)	×
430	enddate = enddate.strftime(fmt)	×
431	url = "{}/json-pushes".format(Mercurial.get_repo_url(channel))	×
432	r = requests.get(	×
433	url,
434	params={"startdate": startdate, "enddate": enddate, "version": 2, "full": 1},
435	)
436	return r.json()	×
437
438
439	def get_bugs_from_desc(desc):	1✔
440	"""Get a bug number from the patch description"""
441	return BUG_PAT.findall(desc)	×
442
443
444	def get_bugs_from_pushlog(startdate, enddate, channel="nightly"):	1✔
445	pushlog = get_pushlog(startdate, enddate, channel=channel)	×
446	bugs = set()	×
447	for push in pushlog["pushes"].values():	×
448	for chgset in push["changesets"]:	×
449	if chgset.get("backedoutby", "") != "":	×
450	continue	×
451	desc = chgset["desc"]	×
452	for bug in get_bugs_from_desc(desc):	×
453	bugs.add(bug)	×
454	return bugs	×
455
456
457	def get_checked_versions():	1✔
458	# There are different reasons to not return versions:
459	# i) we're merge day: the versions are changing
460	# ii) not consecutive versions numbers
461	# iii) bugzilla updated nightly version but p-d is not updated
462	if is_merge_day():	1!
463	return {}	×
464
465	versions = lmdversions.get(base=True)	1✔
466	versions["central"] = versions["nightly"]	1✔
467
468	v = [versions[k] for k in ["release", "beta", "central"]]	1✔
469	versions = {k: str(v) for k, v in versions.items()}	1✔
470
471	if v[0] + 2 == v[1] + 1 == v[2]:	1!
472	nightly_bugzilla = get_nightly_version_from_bz()	1✔
473	if v[2] != nightly_bugzilla:	1!
474	from . import logger	×
475
476	logger.info("Versions mismatch between Bugzilla and product-details")	×
477	return {}	×
478	return versions	1✔
479
480	from . import logger	×
481
482	logger.info("Not consecutive versions in product/details")	×
483	return {}	×
484
485
486	def get_info_from_hg(json):	1✔
487	res = {}	×
488	push = json["pushdate"][0]	×
489	push = datetime.datetime.utcfromtimestamp(push)	×
490	push = lmdutils.as_utc(push)	×
491	res["date"] = lmdutils.get_date_str(push)	×
492	res["backedout"] = json.get("backedoutby", "") != ""	×
493	m = BUG_PAT.search(json["desc"])	×
494	res["bugid"] = m.group(1) if m else ""	×
495
496	return res	×
497
498
499	def bz_ignore_case(s):	1✔
500	return "[" + "][".join(c + c.upper() for c in s) + "]"	×
501
502
503	def check_product_component(data, bug):	1✔
504	prod = bug["product"]	×
505	comp = bug["component"]	×
506	pc = prod + "::" + comp	×
507	return pc in data or comp in data	×
508
509
510	def get_components(data):	1✔
511	res = []	×
512	for comp in data:	×
513	if "::" in comp:	×
514	_, comp = comp.split("::", 1)	×
515	res.append(comp)	×
516	return res	×
517
518
519	def get_products_components(data):	1✔
520	prods = set()	×
521	comps = set()	×
522	for pc in data:	×
523	if "::" in pc:	×
524	p, c = pc.split("::", 1)	×
525	prods.add(p)	×
526	else:
527	c = pc	×
528	comps.add(c)	×
529	return prods, comps	×
530
531
532	def ireplace(old, repl, text):	1✔
533	return re.sub("(?i)" + re.escape(old), lambda m: repl, text)	×
534
535
536	def get_human_lag(date):	1✔
537	today = pytz.utc.localize(datetime.datetime.utcnow())	×
538	dt = dateutil.parser.parse(date)	×
539
540	return humanize.naturaldelta(today - dt)	×
541
542
543	def get_nightly_version_from_bz():	1✔
544	def bug_handler(bug, data):	1✔
545	status = "cf_status_firefox"	1✔
546	N = len(status)	1✔
547	for k in bug.keys():	1✔
548	if k.startswith(status):	1✔
549	k = k[N:]	1✔
550	if k.isdigit():	1✔
551	data.append(int(k))	1✔
552
553	data = []	1✔
554	Bugzilla(bugids=["1234567"], bughandler=bug_handler, bugdata=data).get_data().wait()	1✔
555
556	return max(data)	1✔
557
558
559	def nice_round(val):	1✔
560	return int(round(100 * val))	×

mozilla / relman-auto-nag / 1595

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