#4674

pending completion

Build # #4674

Build Type

push

coveralls-python

Committed by

suhaibmujahid

Commit Message

[bug/analyzer] Fix the bug handler

Run Details

716 of 3562 branches covered (20.1%)

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

1924 of 8713 relevant lines covered (22.08%)

0.22 hits per line

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

78.01

/bugbot/bug/analyzer.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/.


from functools import cached_property
from typing import Any, Iterable, NamedTuple

from libmozdata import versions as lmdversions
from libmozdata.bugzilla import Bugzilla

from bugbot import utils
from bugbot.components import ComponentName


class VersionStatus(NamedTuple):
    """A representation of a version status flag"""

    channel: str
    version: int
    status: str

    @property
    def flag(self) -> str:
        return utils.get_flag(self.version, "status", self.channel)


class BugAnalyzer:
    """A class to analyze a bug"""

    def __init__(self, bug: dict, store: "BugsStore"):
        """Constructor

        Args:
            bug: The bug to analyze
            store: The store of bugs
        """
        self._bug = bug
        self._store = store

    @property
    def id(self) -> int:
        """The bug id."""
        return self._bug["id"]

    @property
    def component(self) -> ComponentName:
        """The component that the bug is in."""
        return ComponentName(self._bug["product"], self._bug["component"])

    @property
    def is_security(self) -> bool:
        """Whether the bug is a security bug."""
        return any("core-security" in group for group in self._bug["groups"])

    @property
    def regressed_by_bugs(self) -> list["BugAnalyzer"]:
        """The bugs that regressed the bug."""
        return [
            self._store.get_bug_by_id(bug_id) for bug_id in self._bug["regressed_by"]
        ]

    @property
    def oldest_fixed_firefox_version(self) -> int | None:
        """The oldest version of Firefox that was fixed by this bug."""
        fixed_versions = sorted(
            int(key[len("cf_status_firefox") :])
            for key, value in self._bug.items()
            if key.startswith("cf_status_firefox")
            and "esr" not in key
            and value in ("fixed", "verified")
        )

        if not fixed_versions:
            return None

        return fixed_versions[0]

    @property
    def latest_firefox_version_status(self) -> str | None:
        """The version status for the latest version of Firefox.

        The latest version is the highest version number that has a status flag
        set (not `---`).
        """
        versions_status = sorted(
            (int(key[len("cf_status_firefox") :]), value)
            for key, value in self._bug.items()
            if value != "---"
            and key.startswith("cf_status_firefox")
            and "esr" not in key
        )

        if not versions_status:
            return None

        return versions_status[-1][1]

    def get_field(self, field: str) -> Any:
        """Get a field value from the bug.

        Args:
            field: The field name.

        Returns:
            The field value. If the field is not found, `None` is returned.
        """
        return self._bug.get(field)

    def detect_version_status_updates(self) -> list[VersionStatus]:
        """Detect the status for the version flags that should be updated.

        The status of the version flags is determined by the status of the
        regressor bug.

        Returns:
            A list of `VersionStatus` objects.
        """
        if len(self._bug["regressed_by"]) > 1:
            # Currently only bugs with one regressor are supported
            return []

        regressor_bug = self.regressed_by_bugs[0]
        regressed_version = regressor_bug.oldest_fixed_firefox_version
        if not regressed_version:
            return []

        fixed_version = self.oldest_fixed_firefox_version

        # If the latest status flag is wontfix or fix-optional, we ignore
        # setting flags with the status "affected" to newer versions.
        is_latest_wontfix = self.latest_firefox_version_status in (
            "wontfix",
            "fix-optional",
        )

        flag_updates = []
        for flag, channel, version in self._store.current_version_flags:
            if flag not in self._bug and channel == "esr":
                # It is okay if an ESR flag is absent (we try two, the current
                # and the previous). However, the absence of other flags is a
                # sign of something wrong.
                continue
            if self._bug[flag] != "---":
                # We don't override existing flags
                # XXX maybe check for consistency?
                continue
            if fixed_version and fixed_version <= version:
                # Bug was fixed in an earlier version, don't set the flag
                continue
            if (
                version >= regressed_version
                # ESR: If the regressor was uplifted, so the regression affects
                # this version.
                or regressor_bug.get_field(flag) in ("fixed", "verified")
            ):
                if is_latest_wontfix:
                    continue

                flag_updates.append(VersionStatus(channel, version, "affected"))
            else:
                flag_updates.append(VersionStatus(channel, version, "unaffected"))

        return flag_updates


class BugNotInStoreError(LookupError):
    """The bug was not found the bugs store."""


class BugsStore:
    """A class to retrieve bugs."""

    def __init__(self, bugs: Iterable[dict] = (), versions_map: dict[str, int] = None):
        self.bugs = {bug["id"]: BugAnalyzer(bug, self) for bug in bugs}
        self.versions_map = versions_map

    def get_bug_by_id(self, bug_id: int) -> BugAnalyzer:
        """Get a bug by its id.

        Args:
            bug_id: The id of the bug to retrieve.

        Returns:
            A `BugAnalyzer` object representing the bug.

        Raises:
            BugNotFoundError: The bug was not found in the store.
        """
        try:
            return self.bugs[bug_id]
        except KeyError as error:
            raise BugNotInStoreError(f"Bug {bug_id} is not the bugs store") from error

    def fetch_regressors(self, include_fields: list[str] = None):
        """Fetches the regressors for all the bugs in the store.

        Args:
            include_fields: The fields to include when fetching the bugs.
        """
        bug_ids = (
            bug_id
            for bug in self.bugs.values()
            if bug.get_field("regressed_by")
            for bug_id in bug.get_field("regressed_by")
        )

        self.fetch_bugs(bug_ids, include_fields)

    def fetch_bugs(self, bug_ids: Iterable[int], include_fields: list[str] = None):
        """Fetches the bugs from Bugzilla.

        Args:
            bug_ids: The ids of the bugs to fetch.
            include_fields: The fields to include when fetching the bugs.
        """
        bug_ids = {
            bug_id
            for bug_id in bug_ids
            # TODO: We only fetch bugs that aren't already in the store.
            # However, the new fetch request might be specifying fields that
            # aren't in the existing bug. We need at some point to handle such
            # cases (currently, we do not have this requirement).
            if bug_id not in self.bugs
        }
        if not bug_ids:
            return

        def bug_handler(bug):
            self.bugs[bug["id"]] = BugAnalyzer(bug, self)

        Bugzilla(bug_ids, bughandler=bug_handler, include_fields=include_fields).wait()

    @cached_property
    def current_version_flags(self) -> list[tuple[str, str, int]]:
        """The current version flags."""
        active_versions = []

        channel_version_map = (
            self.versions_map if self.versions_map else lmdversions.get(base=True)
        )
        for channel in ("release", "beta", "nightly"):
            version = int(channel_version_map[channel])
            flag = utils.get_flag(version, "status", channel)
            active_versions.append((flag, channel, version))

        esr_versions = {
            channel_version_map["esr"],
            channel_version_map["esr_previous"],
        }
        for version in esr_versions:
            channel = "esr"
            flag = utils.get_flag(version, "status", channel)
            active_versions.append((flag, channel, version))

        return active_versions

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
6	from functools import cached_property	1✔
7	from typing import Any, Iterable, NamedTuple	1✔
8
9	from libmozdata import versions as lmdversions	1✔
10	from libmozdata.bugzilla import Bugzilla	1✔
11
12	from bugbot import utils	1✔
13	from bugbot.components import ComponentName	1✔
14
15
16	class VersionStatus(NamedTuple):	1✔
17	"""A representation of a version status flag"""
18
19	channel: str	1✔
20	version: int	1✔
21	status: str	1✔
22
23	@property	1✔
24	def flag(self) -> str:	1✔
25	return utils.get_flag(self.version, "status", self.channel)	1✔
26
27
28	class BugAnalyzer:	1✔
29	"""A class to analyze a bug"""
30
31	def __init__(self, bug: dict, store: "BugsStore"):	1✔
32	"""Constructor
33
34	Args:
35	bug: The bug to analyze
36	store: The store of bugs
37	"""
38	self._bug = bug	1✔
39	self._store = store	1✔
40
41	@property	1✔
42	def id(self) -> int:	1✔
43	"""The bug id."""
44	return self._bug["id"]	×
45
46	@property	1✔
47	def component(self) -> ComponentName:	1✔
48	"""The component that the bug is in."""
49	return ComponentName(self._bug["product"], self._bug["component"])	×
50
51	@property	1✔
52	def is_security(self) -> bool:	1✔
53	"""Whether the bug is a security bug."""
54	return any("core-security" in group for group in self._bug["groups"])	×
55
56	@property	1✔
57	def regressed_by_bugs(self) -> list["BugAnalyzer"]:	1✔
58	"""The bugs that regressed the bug."""
59	return [	1✔
60	self._store.get_bug_by_id(bug_id) for bug_id in self._bug["regressed_by"]
61	]
62
63	@property	1✔
64	def oldest_fixed_firefox_version(self) -> int \| None:	1✔
65	"""The oldest version of Firefox that was fixed by this bug."""
66	fixed_versions = sorted(	1✔
67	int(key[len("cf_status_firefox") :])
68	for key, value in self._bug.items()
69	if key.startswith("cf_status_firefox")
70	and "esr" not in key
71	and value in ("fixed", "verified")
72	)
73
74	if not fixed_versions:	1✔
75	return None	1✔
76
77	return fixed_versions[0]	1✔
78
79	@property	1✔
80	def latest_firefox_version_status(self) -> str \| None:	1✔
81	"""The version status for the latest version of Firefox.
82
83	The latest version is the highest version number that has a status flag
84	set (not `---`).
85	"""
86	versions_status = sorted(	1✔
87	(int(key[len("cf_status_firefox") :]), value)
88	for key, value in self._bug.items()
89	if value != "---"
90	and key.startswith("cf_status_firefox")
91	and "esr" not in key
92	)
93
94	if not versions_status:	1✔
95	return None	1✔
96
97	return versions_status[-1][1]	1✔
98
99	def get_field(self, field: str) -> Any:	1✔
100	"""Get a field value from the bug.
101
102	Args:
103	field: The field name.
104
105	Returns:
106	The field value. If the field is not found, `None` is returned.
107	"""
108	return self._bug.get(field)	1✔
109
110	def detect_version_status_updates(self) -> list[VersionStatus]:	1✔
111	"""Detect the status for the version flags that should be updated.
112
113	The status of the version flags is determined by the status of the
114	regressor bug.
115
116	Returns:
117	A list of `VersionStatus` objects.
118	"""
119	if len(self._bug["regressed_by"]) > 1:	1!
120	# Currently only bugs with one regressor are supported
121	return []	×
122
123	regressor_bug = self.regressed_by_bugs[0]	1✔
124	regressed_version = regressor_bug.oldest_fixed_firefox_version	1✔
125	if not regressed_version:	1!
126	return []	×
127
128	fixed_version = self.oldest_fixed_firefox_version	1✔
129
130	# If the latest status flag is wontfix or fix-optional, we ignore
131	# setting flags with the status "affected" to newer versions.
132	is_latest_wontfix = self.latest_firefox_version_status in (	1✔
133	"wontfix",
134	"fix-optional",
135	)
136
137	flag_updates = []	1✔
138	for flag, channel, version in self._store.current_version_flags:	1✔
139	if flag not in self._bug and channel == "esr":	1!
140	# It is okay if an ESR flag is absent (we try two, the current
141	# and the previous). However, the absence of other flags is a
142	# sign of something wrong.
143	continue	×
144	if self._bug[flag] != "---":	1✔
145	# We don't override existing flags
146	# XXX maybe check for consistency?
147	continue	1✔
148	if fixed_version and fixed_version <= version:	1!
149	# Bug was fixed in an earlier version, don't set the flag
150	continue	×
151	if (	1✔
152	version >= regressed_version
153	# ESR: If the regressor was uplifted, so the regression affects
154	# this version.
155	or regressor_bug.get_field(flag) in ("fixed", "verified")
156	):
157	if is_latest_wontfix:	1!
158	continue	×
159
160	flag_updates.append(VersionStatus(channel, version, "affected"))	1✔
161	else:
162	flag_updates.append(VersionStatus(channel, version, "unaffected"))	1✔
163
164	return flag_updates	1✔
165
166
167	class BugNotInStoreError(LookupError):	1✔
168	"""The bug was not found the bugs store."""
169
170
171	class BugsStore:	1✔
172	"""A class to retrieve bugs."""
173
174	def __init__(self, bugs: Iterable[dict] = (), versions_map: dict[str, int] = None):	1✔
175	self.bugs = {bug["id"]: BugAnalyzer(bug, self) for bug in bugs}	1✔
176	self.versions_map = versions_map	1✔
177
178	def get_bug_by_id(self, bug_id: int) -> BugAnalyzer:	1✔
179	"""Get a bug by its id.
180
181	Args:
182	bug_id: The id of the bug to retrieve.
183
184	Returns:
185	A `BugAnalyzer` object representing the bug.
186
187	Raises:
188	BugNotFoundError: The bug was not found in the store.
189	"""
190	try:	1✔
191	return self.bugs[bug_id]	1✔
192	except KeyError as error:	×
193	raise BugNotInStoreError(f"Bug {bug_id} is not the bugs store") from error	×
194
195	def fetch_regressors(self, include_fields: list[str] = None):	1✔
196	"""Fetches the regressors for all the bugs in the store.
197
198	Args:
199	include_fields: The fields to include when fetching the bugs.
200	"""
201	bug_ids = (	×
202	bug_id
203	for bug in self.bugs.values()
204	if bug.get_field("regressed_by")
205	for bug_id in bug.get_field("regressed_by")
206	)
207
208	self.fetch_bugs(bug_ids, include_fields)	×
209
210	def fetch_bugs(self, bug_ids: Iterable[int], include_fields: list[str] = None):	1✔
211	"""Fetches the bugs from Bugzilla.
212
213	Args:
214	bug_ids: The ids of the bugs to fetch.
215	include_fields: The fields to include when fetching the bugs.
216	"""
217	bug_ids = {	×
218	bug_id
219	for bug_id in bug_ids
220	# TODO: We only fetch bugs that aren't already in the store.
221	# However, the new fetch request might be specifying fields that
222	# aren't in the existing bug. We need at some point to handle such
223	# cases (currently, we do not have this requirement).
224	if bug_id not in self.bugs
225	}
226	if not bug_ids:	×
227	return	×
228
229	def bug_handler(bug):	×
230	self.bugs[bug["id"]] = BugAnalyzer(bug, self)	×
231
232	Bugzilla(bug_ids, bughandler=bug_handler, include_fields=include_fields).wait()	×
233
234	@cached_property	1✔
235	def current_version_flags(self) -> list[tuple[str, str, int]]:	1✔
236	"""The current version flags."""
237	active_versions = []	1✔
238
239	channel_version_map = (	1✔
240	self.versions_map if self.versions_map else lmdversions.get(base=True)
241	)
242	for channel in ("release", "beta", "nightly"):	1✔
243	version = int(channel_version_map[channel])	1✔
244	flag = utils.get_flag(version, "status", channel)	1✔
245	active_versions.append((flag, channel, version))	1✔
246
247	esr_versions = {	1✔
248	channel_version_map["esr"],
249	channel_version_map["esr_previous"],
250	}
251	for version in esr_versions:	1✔
252	channel = "esr"	1✔
253	flag = utils.get_flag(version, "status", channel)	1✔
254	active_versions.append((flag, channel, version))	1✔
255
256	return active_versions	1✔

mozilla / relman-auto-nag / #4674

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