1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
# 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/.
"""
Shared module for fetching intermittent test failure data from Treeherder and Bugzilla.
"""
import datetime
import re
from typing import Literal, Optional, TypedDict
import requests
from wpt_path_utils import resolve_wpt_path
USER_AGENT = "mach-intermittent-failures/1.0"
class BugzillaFailure(TypedDict):
bug_id: int
bug_count: int
class BugzillaSummary(TypedDict):
summary: str
id: int
status: Optional[str]
resolution: Optional[str]
creation_time: Optional[str]
last_change_time: Optional[str]
comment_count: Optional[int]
class IntermittentFailure(TypedDict):
bug_id: int
failure_count: int
summary: str
status: str
resolution: str
test_path: Optional[str]
creation_time: Optional[str]
last_change_time: Optional[str]
comment_count: Optional[int]
class IntermittentFailuresFetcher:
"""Fetches intermittent test failure data from Treeherder and Bugzilla APIs."""
def __init__(self, days: int = 7, threshold: int = 30, verbose: bool = False):
self.days = days
self.threshold = threshold
self.verbose = verbose
self.end_date = datetime.datetime.now()
self.start_date = self.end_date - datetime.timedelta(days=self.days)
def get_failures(self, branch: str = "trunk") -> list[IntermittentFailure]:
"""
Fetch intermittent failures that meet the threshold.
Returns a list of intermittent failures with bug information.
"""
bugzilla_failures = self._get_bugzilla_failures(branch)
bug_list = self._keep_bugs_above_threshold(bugzilla_failures)
if not bug_list:
return []
bug_summaries = self._get_bugzilla_summaries(bug_list)
results = []
for bug in bug_summaries:
bug_id = bug["id"]
if bug_id in bug_list:
result: IntermittentFailure = {
"bug_id": bug_id,
"failure_count": self._get_failure_count(bugzilla_failures, bug_id),
"summary": bug["summary"],
"status": bug.get("status", "UNKNOWN"),
"resolution": bug.get("resolution", ""),
"test_path": None,
"creation_time": bug.get("creation_time"),
"last_change_time": bug.get("last_change_time"),
"comment_count": bug.get("comment_count"),
}
if "single tracking bug" in bug["summary"]:
match = re.findall(
r" ([^\s]+\/?\.[a-z0-9-A-Z]+) \|", bug["summary"]
)
if match:
test_path = match[0]
if test_path.startswith("/"):
test_path = resolve_wpt_path(test_path)
result["test_path"] = test_path
results.append(result)
return results
def get_single_tracking_bugs_with_paths(
self, branch: str = "trunk"
) -> list[tuple[int, str]]:
"""
Get only single tracking bugs that have test paths.
This is what high_freq_skipfails uses.
Returns a list of (bug_id, test_path) tuples.
"""
failures = self.get_failures(branch)
results = []
for failure in failures:
if failure["test_path"] and "single tracking bug" in failure["summary"]:
results.append((failure["bug_id"], failure["test_path"]))
return results
def _get_bugzilla_failures(self, branch: str = "trunk") -> list[BugzillaFailure]:
"""Fetch failure data from Treeherder API."""
url = (
f"https://treeherder.mozilla.org/api/failures/"
f"?startday={self.start_date.date()}&endday={self.end_date.date()}"
f"&tree={branch}&failurehash=all"
)
if self.verbose:
print(f"[DEBUG] Fetching failures from Treeherder: {url}")
response = requests.get(url, headers={"User-agent": USER_AGENT})
response.raise_for_status()
return [
item
for item in response.json()
if "bug_id" in item and isinstance(item["bug_id"], int)
]
def _keep_bugs_above_threshold(
self, failure_list: list[BugzillaFailure]
) -> list[int]:
"""Filter bugs that have failure counts above the threshold."""
if not failure_list:
return []
bug_counts = {}
for failure in failure_list:
bug_id = failure["bug_id"]
bug_counts[bug_id] = bug_counts.get(bug_id, 0) + failure.get("bug_count", 1)
return [
bug_id for bug_id, count in bug_counts.items() if count >= self.threshold
]
def _get_failure_count(
self, failure_list: list[BugzillaFailure], bug_id: int
) -> int:
"""Get the total failure count for a specific bug."""
total = 0
for failure in failure_list:
if failure["bug_id"] == bug_id:
total += failure.get("bug_count", 1)
return total
def _get_bugzilla_summaries(self, bug_id_list: list[int]) -> list[BugzillaSummary]:
"""Fetch bug summaries from Bugzilla REST API."""
if not bug_id_list:
return []
url = (
f"https://bugzilla.mozilla.org/rest/bug"
f"?include_fields=summary,id,status,resolution,creation_time,last_change_time,comment_count"
f"&id={','.join(str(id) for id in bug_id_list)}"
)
if self.verbose:
print(f"[DEBUG] Fetching bug details from Bugzilla: {url}")
response = requests.get(url, headers={"User-agent": USER_AGENT})
response.raise_for_status()
json_response: dict[Literal["bugs"], list[BugzillaSummary]] = response.json()
return json_response.get("bugs", [])
|
