#4649

pending completion

Build # #4649

Build Type

push

coveralls-python

Committed by

suhaibmujahid

Commit Message

Create a standalone logic to determine version status flags

Run Details

713 of 3543 branches covered (20.12%)

106 of 106 new or added lines in 2 files covered. (100.0%)

1917 of 8684 relevant lines covered (22.08%)

0.22 hits per line

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

79.84

/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


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 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] | None:
        """Detect the status for the version flags that should to 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 None

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

        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")
            if bug_id not in self.bugs
        }

        if not bug_ids:
            return

        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.
        """

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

mozilla / relman-auto-nag / #4649

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