Setting the file. One moment.
Fetch PR Checks · Iterate PR · getsentry/skills · Skills Docs
Back to the top of the page scripts/ fetch_pr_checks.py
Python · 305 lines · 10 KB
17 import argparse
18 import json
19 import re
20 import subprocess
21 import sys
22 from typing import Any
23
24 HUMAN_GATE_PATTERNS = [
25 r " (?i) review \s + required" ,
26 r " (?i) required \s + review" ,
27 r " (?i) requires \s + review" ,
28 r " (?i) required \s + approving \s + review" ,
29 r " (?i) approval \s + required" ,
30 r " (?i) waiting \s + for \s + approval" ,
31 r " (?i) manual \s + approval" ,
32 r " (?i) draft \s + ( pull \s + request | pr ) " ,
33 ]
34
35
36 def run_gh (args: list[ str ]) -> dict[ str , Any] | list[Any] | None :
37 """Run a gh CLI command and return parsed JSON output."""
38 try :
39 result = subprocess.run(
40 [ "gh" ] + args,
41 capture_output = True ,
42 text = True ,
43 check = True ,
44 )
45 return json.loads(result.stdout) if result.stdout.strip() else None
46 except subprocess.CalledProcessError as e:
47 print ( f "Error running gh { ' ' .join(args) } : { e.stderr } " , file = sys.stderr)
48 return None
49 except json.JSONDecodeError:
50 return None
51
52
53 def get_pr_info (pr_number: int | None = None ) -> dict[ str , Any] | None :
54 """Get PR info, optionally by number or for current branch."""
55 args = [
56 "pr" ,
57 "view" ,
58 "--json" ,
59 "number,url,headRefName,baseRefName,isDraft,reviewDecision" ,
60 ]
61 if pr_number:
62 args.insert( 2 , str (pr_number))
63 return run_gh(args)
64
65
66 def get_checks (pr_number: int | None = None ) -> list[dict[ str , Any]]:
67 """Get all checks for a PR."""
68 args = [ "gh" , "pr" , "checks" ]
69 if pr_number:
70 args.append( str (pr_number))
71 args.extend([ "--json" , "name,bucket,link,workflow,state,description,event" ])
72 try :
73 result = subprocess.run(
74 args,
75 capture_output = True ,
76 text = True ,
77 )
78 if not result.stdout.strip():
79 return []
80 try :
81 checks = json.loads(result.stdout)
82 return checks if isinstance (checks, list ) else []
83 except json.JSONDecodeError:
84 pass
85
86 checks = []
87 for line in result.stdout.strip().split( " \n " ):
88 if not line.strip():
89 continue
90 parts = line.split( " \t " )
91 if len (parts) >= 2 :
92 checks.append({
93 "name" : parts[ 0 ].strip(),
94 "bucket" : parts[ 1 ].strip(),
95 "link" : parts[ 3 ].strip() if len (parts) > 3 else "" ,
96 "workflow" : "" ,
97 })
98 return checks
99 except Exception :
100 return []
101
102
103 def is_human_gate_check (check: dict[ str , Any]) -> bool :
104 """Return true when a pending entry is a human review/approval gate."""
105 haystack = " " .join(
106 str (check.get(field, "" ))
107 for field in ( "name" , "state" , "description" , "workflow" )
108 )
109 return any (re.search(pattern, haystack) for pattern in HUMAN_GATE_PATTERNS )
110
111
112 def get_failed_runs (branch: str ) -> list[dict[ str , Any]]:
113 """Get recent failed workflow runs for a branch."""
114 result = run_gh([
115 "run" , "list" ,
116 "--branch" , branch,
117 "--limit" , "10" ,
118 "--json" , "databaseId,name,status,conclusion,headSha"
119 ])
120 if not isinstance (result, list ):
121 return []
122 # Return runs that failed or are in progress
123 return [r for r in result if r.get( "conclusion" ) == "failure" ]
124
125
126 def extract_failure_snippet (log_text: str , max_lines: int = 50 ) -> str :
127 """Extract relevant failure snippet from log text.
128
129 Looks for common failure markers and extracts surrounding context.
130 """
131 lines = log_text.split( " \n " )
132
133 # Patterns that indicate failure points (case-insensitive via re.IGNORECASE)
134 failure_patterns = [
135 r "error [ : \s] " ,
136 r "failed [ : \s] " ,
137 r "failure [ : \s] " ,
138 r "traceback" ,
139 r "exception" ,
140 r "assert ( ion ) ? . * failed" ,
141 r "FAILED" ,
142 r "panic:" ,
143 r "fatal:" ,
144 r "npm ERR!" ,
145 r "yarn error" ,
146 r "ModuleNotFoundError" ,
147 r "ImportError" ,
148 r "SyntaxError" ,
149 r "TypeError" ,
150 r "ValueError" ,
151 r "KeyError" ,
152 r "AttributeError" ,
153 r "NameError" ,
154 r "IndentationError" ,
155 r "=== . * FAILURES . * ===" ,
156 r "___ . * ___" , # pytest failure separators
157 ]
158
159 combined_pattern = "|" .join(failure_patterns)
160
161 # Find lines matching failure patterns
162 failure_indices = []
163 for i, line in enumerate (lines):
164 if re.search(combined_pattern, line, re. IGNORECASE ):
165 failure_indices.append(i)
166
167 if not failure_indices:
168 # No clear failure point, return last N lines
169 return " \n " .join(lines[ - max_lines:])
170
171 # Extract context around first failure point
172 # Include some context before and after
173 first_failure = failure_indices[ 0 ]
174 start = max ( 0 , first_failure - 5 )
175 end = min ( len (lines), first_failure + max_lines - 5 )
176
177 snippet_lines = lines[start:end]
178
179 # If there are more failures after our snippet, note it
180 remaining_failures = [i for i in failure_indices if i >= end]
181 if remaining_failures:
182 snippet_lines.append( f " \n ... ( { len (remaining_failures) } more error(s) follow)" )
183
184 return " \n " .join(snippet_lines)
185
186
187 def get_run_logs (run_id: int ) -> str | None :
188 """Get failed logs for a workflow run."""
189 try :
190 result = subprocess.run(
191 [ "gh" , "run" , "view" , str (run_id), "--log-failed" ],
192 capture_output = True ,
193 text = True ,
194 timeout = 60 ,
195 )
196 return result.stdout if result.stdout else result.stderr
197 except subprocess.TimeoutExpired:
198 return None
199 except subprocess.CalledProcessError:
200 return None
201
202
203 def main ():
204 parser = argparse.ArgumentParser( description = "Fetch PR CI checks with failure snippets" )
205 parser.add_argument( "--pr" , type = int , help = "PR number (defaults to current branch PR)" )
206 args = parser.parse_args()
207
208 # Get PR info
209 pr_info = get_pr_info(args.pr)
210 if not pr_info:
211 print (json.dumps({ "error" : "No PR found for current branch" }))
212 sys.exit( 1 )
213
214 pr_number = pr_info[ "number" ]
215 branch = pr_info[ "headRefName" ]
216
217 # Get checks
218 checks = get_checks(pr_number)
219
220 # Process checks and add failure snippets
221 processed_checks = []
222 failed_runs = None # Lazy load
223
224 for check in checks:
225 status = check.get( "bucket" , check.get( "state" , "unknown" ))
226 human_gate = status == "pending" and is_human_gate_check(check)
227 processed = {
228 "name" : check.get( "name" , "unknown" ),
229 "status" : status,
230 "link" : check.get( "link" , "" ),
231 "workflow" : check.get( "workflow" , "" ),
232 }
233 if check.get( "state" ):
234 processed[ "state" ] = check[ "state" ]
235 if check.get( "description" ):
236 processed[ "description" ] = check[ "description" ]
237 if human_gate:
238 processed[ "human_gate" ] = True
239
240 # For failures, try to get log snippet
241 if processed[ "status" ] == "fail" :
242 if failed_runs is None :
243 failed_runs = get_failed_runs(branch)
244
245 # Find matching run by workflow name
246 workflow_name = processed[ "workflow" ] or processed[ "name" ]
247 matching_run = next (
248 (r for r in failed_runs if workflow_name in r.get( "name" , "" )),
249 None
250 )
251
252 if matching_run:
253 logs = get_run_logs(matching_run[ "databaseId" ])
254 if logs:
255 processed[ "log_snippet" ] = extract_failure_snippet(logs)
256 processed[ "run_id" ] = matching_run[ "databaseId" ]
257
258 processed_checks.append(processed)
259
260 # Build output
261 output = {
262 "pr" : {
263 "number" : pr_number,
264 "url" : pr_info.get( "url" , "" ),
265 "branch" : branch,
266 "base" : pr_info.get( "baseRefName" , "" ),
267 "is_draft" : bool (pr_info.get( "isDraft" )),
268 "review_decision" : pr_info.get( "reviewDecision" , "" ),
269 },
270 "summary" : {
271 "total" : len (processed_checks),
272 "passed" : sum ( 1 for c in processed_checks if c[ "status" ] == "pass" ),
273 "failed" : sum ( 1 for c in processed_checks if c[ "status" ] == "fail" ),
274 "pending" : sum ( 1 for c in processed_checks if c[ "status" ] == "pending" ),
275 "actionable_pending" : sum (
276 1
277 for c in processed_checks
278 if c[ "status" ] == "pending" and not c.get( "human_gate" )
279 ),
280 "human_gate_pending" : sum (
281 1
282 for c in processed_checks
283 if c[ "status" ] == "pending" and c.get( "human_gate" )
284 ),
285 "skipped" : sum ( 1 for c in processed_checks if c[ "status" ] in ( "skipping" , "cancel" )),
286 },
287 "checks" : processed_checks,
288 }
289
290 if pr_info.get( "isDraft" ) and not processed_checks:
291 output[ "action_required" ] = "Draft PR has no registered checks; do not wait for CI indefinitely"
292 elif not processed_checks:
293 output[ "action_required" ] = "No registered checks; monitor before reporting NO_CHECKS_REGISTERED"
294 elif output[ "summary" ][ "actionable_pending" ]:
295 output[ "action_required" ] = "Wait for actionable checks to finish; poll feedback while waiting"
296 elif output[ "summary" ][ "failed" ]:
297 output[ "action_required" ] = "Address failed checks"
298 elif output[ "summary" ][ "pending" ] and not output[ "summary" ][ "actionable_pending" ]:
299 output[ "action_required" ] = "Only human review or approval gates remain pending"
300
301 print (json.dumps(output, indent = 2 ))
302
303
304 if __name__ == "__main__" :
305 main()