MCPcopy Index your code
hub / github.com/streamlit/streamlit / main

Function main

scripts/fetch_flaky_tests.py:167–253  ·  view source on GitHub ↗
()

Source from the content-addressed store, hash-verified

165
166
167def main() -> None:
168 parser = argparse.ArgumentParser(
169 description="Fetch top flaky E2E tests from recent CI runs"
170 )
171 parser.add_argument(
172 "--days",
173 type=int,
174 default=4,
175 help="Number of days to look back (default: 4)",
176 )
177 parser.add_argument(
178 "--limit",
179 type=int,
180 default=100,
181 help="Maximum number of workflow runs to fetch (default: 100)",
182 )
183 parser.add_argument(
184 "--top",
185 type=int,
186 default=10,
187 help="Number of top flaky tests to return (default: 10)",
188 )
189 parser.add_argument(
190 "--min-reruns",
191 type=int,
192 default=2,
193 help="Minimum total reruns to include (default: 2)",
194 )
195 parser.add_argument(
196 "--json",
197 action="store_true",
198 help="Output as JSON",
199 )
200 args = parser.parse_args()
201
202 print(
203 f"Fetching successful playwright.yml runs from the last {args.days} days...",
204 file=sys.stderr,
205 )
206 runs = _fetch_successful_workflow_runs(days=args.days, limit=args.limit)
207 print(f"Found {len(runs)} successful runs", file=sys.stderr)
208
209 if not runs:
210 print("No successful runs found in the specified time range", file=sys.stderr)
211 sys.exit(0)
212
213 all_flaky_tests = []
214 with tempfile.TemporaryDirectory() as temp_dir:
215 temp_path = Path(temp_dir)
216 for i, run in enumerate(runs):
217 run_id = run["databaseId"]
218 print(
219 f"Processing run {i + 1}/{len(runs)} (ID: {run_id})...",
220 file=sys.stderr,
221 )
222
223 test_stats = _download_test_stats(run_id, temp_path)
224 if test_stats:

Callers 1

Calls 5

_download_test_statsFunction · 0.85
_extract_flaky_testsFunction · 0.85
_aggregate_flaky_testsFunction · 0.85
joinMethod · 0.45

Tested by

no test coverage detected

Used in the wild real call sites across dependent graphs

searching dependent graphs…