#4767

Committed 13 Oct 2023 01:27AM CUT coverage: 22.091%. Remained the same

Build # #4767

Build Type

push

coveralls-python

Committed by

suhaibmujahid

Commit Message

Format the .pre-commit-config.yaml file

Run Details

716 of 3558 branches covered (0.0%)

1925 of 8714 relevant lines covered (22.09%)

0.22 hits per line

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

82.0

/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 = 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 = 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 = 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__(	1✔
175	self, bugs: Iterable[dict] = (), versions_map: dict[str, int] \| None = None
176	):
177	self.bugs = {bug["id"]: BugAnalyzer(bug, self) for bug in bugs}	1✔
178	self.versions_map = versions_map	1✔
179
180	def get_bug_by_id(self, bug_id: int) -> BugAnalyzer:	1✔
181	"""Get a bug by its id.
182
183	Args:
184	bug_id: The id of the bug to retrieve.
185
186	Returns:
187	A `BugAnalyzer` object representing the bug.
188
189	Raises:
190	BugNotFoundError: The bug was not found in the store.
191	"""
192	try:	1✔
193	return self.bugs[bug_id]	1✔
194	except KeyError as error:	×
195	raise BugNotInStoreError(f"Bug {bug_id} is not the bugs store") from error	×
196
197	def fetch_regressors(self, include_fields: list[str] \| None = None):	1✔
198	"""Fetches the regressors for all the bugs in the store.
199
200	Args:
201	include_fields: The fields to include when fetching the bugs.
202	"""
203	bug_ids = (	×
204	bug_id
205	for bug in self.bugs.values()
206	if bug.get_field("regressed_by")
207	for bug_id in bug.get_field("regressed_by")
208	)
209
210	self.fetch_bugs(bug_ids, include_fields)	×
211
212	def fetch_bugs(	1✔
213	self, bug_ids: Iterable[int], include_fields: list[str] \| None = None
214	):
215	"""Fetches the bugs from Bugzilla.
216
217	Args:
218	bug_ids: The ids of the bugs to fetch.
219	include_fields: The fields to include when fetching the bugs.
220	"""
221	bug_ids = {	×
222	bug_id
223	for bug_id in bug_ids
224	# TODO: We only fetch bugs that aren't already in the store.
225	# However, the new fetch request might be specifying fields that
226	# aren't in the existing bug. We need at some point to handle such
227	# cases (currently, we do not have this requirement).
228	if bug_id not in self.bugs
229	}
230	if not bug_ids:	×
231	return	×
232
233	def bug_handler(bug):	×
234	self.bugs[bug["id"]] = BugAnalyzer(bug, self)	×
235
236	Bugzilla(bug_ids, bughandler=bug_handler, include_fields=include_fields).wait()	×
237
238	@cached_property	1✔
239	def current_version_flags(self) -> list[tuple[str, str, int]]:	1✔
240	"""The current version flags."""
241	active_versions = []	1✔
242
243	channel_version_map = (	1✔
244	self.versions_map if self.versions_map else lmdversions.get(base=True)
245	)
246	for channel in ("release", "beta", "nightly"):	1✔
247	version = int(channel_version_map[channel])	1✔
248	flag = utils.get_flag(version, "status", channel)	1✔
249	active_versions.append((flag, channel, version))	1✔
250
251	esr_versions = {	1✔
252	channel_version_map["esr"],
253	channel_version_map["esr_previous"],
254	}
255	for version in esr_versions:	1✔
256	channel = "esr"	1✔
257	flag = utils.get_flag(version, "status", channel)	1✔
258	active_versions.append((flag, channel, version))	1✔
259
260	return active_versions	1✔

mozilla / relman-auto-nag / #4767

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