in scripts/generate_landings_risk_report.py [0:0]
def retrieve_test_info(self, days: int) -> dict[str, Any]:
logger.info("Download previous test info...")
db.download(TEST_INFOS_DB)
dates = [
datetime.utcnow() - timedelta(days=day)
for day in reversed(range(min(days, 90)))
]
logger.info("Get previously gathered test info...")
test_infos = {
test_info["date"]: test_info for test_info in db.read(TEST_INFOS_DB)
}
prev_skips = None
for date in tqdm(dates):
date_str = date.strftime("%Y-%m-%d")
# Gather the latest three days again, as the data might have changed.
if date_str in test_infos and date < datetime.utcnow() - timedelta(days=3):
prev_skips = test_infos[date_str]["skips"]
continue
test_infos[date_str] = {
"date": date_str,
"bugs": [
{"id": item["bug_id"], "count": item["bug_count"]}
for item in test_scheduling.get_failure_bugs(date, date)
],
"skips": {},
}
try:
test_info = test_scheduling.get_test_info(date)
for component in test_info["tests"].keys():
test_infos[date_str]["skips"][component] = sum(
1 for test in test_info["tests"][component] if "skip-if" in test
)
except requests.exceptions.HTTPError:
# If we couldn't find a test info artifact for the given date, assume the number of skip-ifs didn't change from the previous day.
assert prev_skips is not None
test_infos[date_str]["skips"] = prev_skips
prev_skips = test_infos[date_str]["skips"]
db.write(
TEST_INFOS_DB,
(
test_infos[date.strftime("%Y-%m-%d")]
for date in dates
if date.strftime("%Y-%m-%d") in test_infos
),
)
zstd_compress(TEST_INFOS_DB)
return test_infos