From 68052c2ea3da43d00b6b04e36ffe34d7522ac808 Mon Sep 17 00:00:00 2001 From: Hook25 Date: Fri, 6 Jun 2025 13:57:01 +0200 Subject: [PATCH 1/6] Init commit of the script to get detailed test failure reporting --- backend/scripts/test_failure_reason.py | 150 +++++++++++++++++++++++++ 1 file changed, 150 insertions(+) create mode 100644 backend/scripts/test_failure_reason.py diff --git a/backend/scripts/test_failure_reason.py b/backend/scripts/test_failure_reason.py new file mode 100644 index 000000000..75b603afe --- /dev/null +++ b/backend/scripts/test_failure_reason.py @@ -0,0 +1,150 @@ +import os +import csv +import json +import string +import urllib.request +from multiprocessing import Pool +from argparse import ArgumentParser + +access_token = os.getenv("C3_TOKEN") + + +def parse_args(): + ap = ArgumentParser() + ap.add_argument("--filter", choices=("id", "artifact_name"), default="id") + ap.add_argument("csvs", nargs="+") + ap.add_argument("test_id") + return ap.parse_args() + + +def get_summary(sub_id): + """ + Get the validation results of a given list of submissions from the C3 API. + """ + api_url = ( + f"https://certification.canonical.com/api/v2/reports/summary/{sub_id}" + ) + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {access_token}", + } + # Convert the payload to a JSON string + # Create the request object with the data and headers + req = urllib.request.Request(api_url, headers=headers, method="GET") + # Send the request and get the response + with urllib.request.urlopen(req) as response: + result = json.loads(response.read().decode("utf-8")) + + return result + + +def slugify(_string: str): + if not _string: + return _string + + valid_chars = frozenset(f"_{string.ascii_letters}{string.digits}") + # Python identifiers cannot start with a digit + if _string[0].isdigit(): + _string = "_" + _string + return "".join(c if c in valid_chars else "_" for c in _string) + + +def get_filter(filter_name, filter_param): + + def id_filter(x: dict) -> bool: + # if a test fails, the test run fail, so its 2 failed + return ( + x["TestCase.name"] == filter_param + or x["TestCase.template_id"] == filter_param + ) and x["TestResult.status"] == "FAILED" + + def artifact_filter(x: dict) -> bool: + return ( + x["Artefact.name"] == filter_param + and x["TestResult.status"] == "FAILED" + ) + + return {"id": id_filter, "artifact_name": artifact_filter}[filter_name] + + +def main(): + args = parse_args() + lines_of_interest = [] + filter_f = get_filter(args.filter, args.test_id) + + for f_path in args.csvs: + with open(f_path) as f: + reader = csv.DictReader(f, delimiter=",") + lines_of_interest += list(filter(filter_f, reader)) + + artefact_id = (x["Artefact.name"] for x in lines_of_interest) + machine_id = (x["TestExecution.c3_link"][45:57] for x in lines_of_interest) + sub_links = (x["TestExecution.c3_link"] for x in lines_of_interest) + sub_links = [f"{x}test-results/fail" for x in sub_links] + print(f"Found {len(sub_links)} failures") + + relevant_template_id = { + x["TestCase.template_id"] for x in lines_of_interest + } + relevant_ids = { + x["TestCase.name"] for x in lines_of_interest + } | relevant_template_id + relevant_ids = relevant_ids - {None, ""} + + file_name = slugify(args.test_id) + + with open(f"{file_name}.url", "w+") as f: + f.writelines("\n".join(sub_links)) + + sub_ids = [x.rsplit("/", 3)[-3] for x in sub_links] + sub_summaries = [] + print("Downloading all submissions") + with Pool(20) as p: + for i, sub_summary in enumerate(p.imap(get_summary, sub_ids)): + print(f"Done {i+1}/{len(sub_ids)}") + sub_summaries.append(sub_summary["results"][0]["testresult_set"]) + + job_objects = [ + list( + filter( + lambda x: ( + x["name"] in relevant_ids + or x["template_id"] in relevant_ids + ) + and x["status"] == "fail", + sub, + ) + ) + for sub in sub_summaries + ] + fieldnames = [ + "Submission Link", + "Job ID", + "Template ID", + "Job log", + "Machine", + "Artefact id", + ] + results = zip(sub_links, job_objects, machine_id, artefact_id) + results = ( + ( + sub_link, + job_object["name"], + job_object.get("template_id", ""), + job_object["io_log"], + machine_id, + artefact_id, + ) + for (sub_link, job_objects, machine_id, artefact_id) in results + for job_object in job_objects + ) + result_rows = [dict(zip(fieldnames, result_row)) for result_row in results] + + with open(f"{file_name}.csv", "w+") as f: + writer = csv.DictWriter(f, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(result_rows) + + +if __name__ == "__main__": + main() From 066a77c884078744491a115d56005f76ac314dbc Mon Sep 17 00:00:00 2001 From: Hook25 Date: Fri, 6 Jun 2025 16:36:59 +0200 Subject: [PATCH 2/6] Also expose the TO link Minor: bad hack to add an s, ops --- backend/scripts/test_failure_reason.py | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/backend/scripts/test_failure_reason.py b/backend/scripts/test_failure_reason.py index 75b603afe..1b2a7bd08 100644 --- a/backend/scripts/test_failure_reason.py +++ b/backend/scripts/test_failure_reason.py @@ -78,6 +78,10 @@ def main(): lines_of_interest += list(filter(filter_f, reader)) artefact_id = (x["Artefact.name"] for x in lines_of_interest) + to_links = [ + f"https://test-observer.canonical.com/#/{x["Artefact.family"]}s/{x["Artefact.id"]}" + for x in lines_of_interest + ] machine_id = (x["TestExecution.c3_link"][45:57] for x in lines_of_interest) sub_links = (x["TestExecution.c3_link"] for x in lines_of_interest) sub_links = [f"{x}test-results/fail" for x in sub_links] @@ -118,6 +122,7 @@ def main(): for sub in sub_summaries ] fieldnames = [ + "TestObserver Link", "Submission Link", "Job ID", "Template ID", @@ -125,9 +130,10 @@ def main(): "Machine", "Artefact id", ] - results = zip(sub_links, job_objects, machine_id, artefact_id) + results = zip(to_links, sub_links, job_objects, machine_id, artefact_id) results = ( ( + to_link, sub_link, job_object["name"], job_object.get("template_id", ""), @@ -135,7 +141,13 @@ def main(): machine_id, artefact_id, ) - for (sub_link, job_objects, machine_id, artefact_id) in results + for ( + to_link, + sub_link, + job_objects, + machine_id, + artefact_id, + ) in results for job_object in job_objects ) result_rows = [dict(zip(fieldnames, result_row)) for result_row in results] From eeae01d6116d4d967f4428c5785dd3fe1820398b Mon Sep 17 00:00:00 2001 From: Hook25 Date: Thu, 3 Jul 2025 15:17:43 +0200 Subject: [PATCH 3/6] Cache submissions instead of re-downloading them 20 times Minor: more fields to the report Minor: Better TO link (now straight to the artifact --- backend/scripts/test_failure_reason.py | 42 ++++++++++++++++++++------ 1 file changed, 32 insertions(+), 10 deletions(-) diff --git a/backend/scripts/test_failure_reason.py b/backend/scripts/test_failure_reason.py index 1b2a7bd08..60452bcb7 100644 --- a/backend/scripts/test_failure_reason.py +++ b/backend/scripts/test_failure_reason.py @@ -35,7 +35,7 @@ def get_summary(sub_id): with urllib.request.urlopen(req) as response: result = json.loads(response.read().decode("utf-8")) - return result + return (sub_id, result) def slugify(_string: str): @@ -79,10 +79,15 @@ def main(): artefact_id = (x["Artefact.name"] for x in lines_of_interest) to_links = [ - f"https://test-observer.canonical.com/#/{x["Artefact.family"]}s/{x["Artefact.id"]}" + f"https://test-observer.canonical.com/#/{x["Artefact.family"]}s/{x["Artefact.id"]}?q={x["Environment.name"]}" for x in lines_of_interest ] machine_id = (x["TestExecution.c3_link"][45:57] for x in lines_of_interest) + environment_name = (x["Environment.name"] for x in lines_of_interest) + checkbox_version = ( + x["TestExecution.checkbox_version"] for x in lines_of_interest + ) + sub_links = (x["TestExecution.c3_link"] for x in lines_of_interest) sub_links = [f"{x}test-results/fail" for x in sub_links] print(f"Found {len(sub_links)} failures") @@ -101,12 +106,15 @@ def main(): f.writelines("\n".join(sub_links)) sub_ids = [x.rsplit("/", 3)[-3] for x in sub_links] - sub_summaries = [] + unique_ids = list(set(sub_ids)) + sub_summaries = {} print("Downloading all submissions") - with Pool(20) as p: - for i, sub_summary in enumerate(p.imap(get_summary, sub_ids)): - print(f"Done {i+1}/{len(sub_ids)}") - sub_summaries.append(sub_summary["results"][0]["testresult_set"]) + with Pool(min(len(unique_ids), 20)) as p: + for i, (sub_id, sub_summary) in enumerate( + p.imap(get_summary, unique_ids) + ): + print(f"Done {i+1}/{len(unique_ids)}") + sub_summaries[sub_id] = sub_summary["results"][0]["testresult_set"] job_objects = [ list( @@ -116,10 +124,10 @@ def main(): or x["template_id"] in relevant_ids ) and x["status"] == "fail", - sub, + sub_summaries[sub_id], ) ) - for sub in sub_summaries + for sub_id in sub_ids ] fieldnames = [ "TestObserver Link", @@ -128,9 +136,19 @@ def main(): "Template ID", "Job log", "Machine", + "Environment Name", + "Checkbox Version", "Artefact id", ] - results = zip(to_links, sub_links, job_objects, machine_id, artefact_id) + results = zip( + to_links, + sub_links, + job_objects, + machine_id, + environment_name, + checkbox_version, + artefact_id, + ) results = ( ( to_link, @@ -139,6 +157,8 @@ def main(): job_object.get("template_id", ""), job_object["io_log"], machine_id, + environment_name, + checkbox_version, artefact_id, ) for ( @@ -146,6 +166,8 @@ def main(): sub_link, job_objects, machine_id, + environment_name, + checkbox_version, artefact_id, ) in results for job_object in job_objects From fc7b30a408edcc7c2f57695336027e9ddfc041ca Mon Sep 17 00:00:00 2001 From: Hook25 Date: Fri, 18 Jul 2025 13:43:09 +0200 Subject: [PATCH 4/6] Only include family snap or deb --- backend/scripts/test_failure_reason.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/backend/scripts/test_failure_reason.py b/backend/scripts/test_failure_reason.py index 60452bcb7..6ab5ceba6 100644 --- a/backend/scripts/test_failure_reason.py +++ b/backend/scripts/test_failure_reason.py @@ -64,7 +64,15 @@ def artifact_filter(x: dict) -> bool: and x["TestResult.status"] == "FAILED" ) - return {"id": id_filter, "artifact_name": artifact_filter}[filter_name] + def both(f): + def _f(x: dict) -> bool: + return x["Artefact.family"] in ["snap", "deb"] and f(x) + + return _f + + return {"id": both(id_filter), "artifact_name": both(artifact_filter)}[ + filter_name + ] def main(): From 2b7eee641c6667f7777e0810bd29e2b67dba49cb Mon Sep 17 00:00:00 2001 From: Hook25 Date: Thu, 19 Feb 2026 10:37:03 +0100 Subject: [PATCH 5/6] Also export date --- backend/scripts/test_failure_reason.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/backend/scripts/test_failure_reason.py b/backend/scripts/test_failure_reason.py index 6ab5ceba6..8b8293e8e 100644 --- a/backend/scripts/test_failure_reason.py +++ b/backend/scripts/test_failure_reason.py @@ -95,6 +95,7 @@ def main(): checkbox_version = ( x["TestExecution.checkbox_version"] for x in lines_of_interest ) + test_date = (x["TestResult.created_at"] for x in lines_of_interest) sub_links = (x["TestExecution.c3_link"] for x in lines_of_interest) sub_links = [f"{x}test-results/fail" for x in sub_links] @@ -119,7 +120,7 @@ def main(): print("Downloading all submissions") with Pool(min(len(unique_ids), 20)) as p: for i, (sub_id, sub_summary) in enumerate( - p.imap(get_summary, unique_ids) + p.imap_unordered(get_summary, unique_ids) ): print(f"Done {i+1}/{len(unique_ids)}") sub_summaries[sub_id] = sub_summary["results"][0]["testresult_set"] @@ -147,6 +148,7 @@ def main(): "Environment Name", "Checkbox Version", "Artefact id", + "Test date", ] results = zip( to_links, @@ -156,6 +158,7 @@ def main(): environment_name, checkbox_version, artefact_id, + test_date, ) results = ( ( @@ -168,6 +171,7 @@ def main(): environment_name, checkbox_version, artefact_id, + test_date, ) for ( to_link, @@ -177,6 +181,7 @@ def main(): environment_name, checkbox_version, artefact_id, + test_date, ) in results for job_object in job_objects ) From 9eff016489d9bbc5e2965e733cb14efd3e29d481 Mon Sep 17 00:00:00 2001 From: Hook25 Date: Thu, 5 Mar 2026 11:01:24 +0100 Subject: [PATCH 6/6] Add capability to query more than 1 test id and retries --- backend/scripts/test_failure_reason.py | 60 +++++++++++++++++--------- 1 file changed, 40 insertions(+), 20 deletions(-) diff --git a/backend/scripts/test_failure_reason.py b/backend/scripts/test_failure_reason.py index 8b8293e8e..997a0fee2 100644 --- a/backend/scripts/test_failure_reason.py +++ b/backend/scripts/test_failure_reason.py @@ -2,18 +2,23 @@ import csv import json import string +import time import urllib.request from multiprocessing import Pool from argparse import ArgumentParser +RETRY = 10 + access_token = os.getenv("C3_TOKEN") +if not access_token: + raise SystemExit("C3_TOKEN required") def parse_args(): ap = ArgumentParser() ap.add_argument("--filter", choices=("id", "artifact_name"), default="id") ap.add_argument("csvs", nargs="+") - ap.add_argument("test_id") + ap.add_argument("--test-ids", nargs="+", required=True) return ap.parse_args() @@ -21,26 +26,34 @@ def get_summary(sub_id): """ Get the validation results of a given list of submissions from the C3 API. """ - api_url = ( - f"https://certification.canonical.com/api/v2/reports/summary/{sub_id}" - ) - headers = { - "Content-Type": "application/json", - "Authorization": f"Bearer {access_token}", - } - # Convert the payload to a JSON string - # Create the request object with the data and headers - req = urllib.request.Request(api_url, headers=headers, method="GET") - # Send the request and get the response - with urllib.request.urlopen(req) as response: - result = json.loads(response.read().decode("utf-8")) + for i in range(RETRY): + try: + api_url = f"https://certification.canonical.com/api/v2/reports/summary/{sub_id}" + headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {access_token}", + } + # Convert the payload to a JSON string + # Create the request object with the data and headers + req = urllib.request.Request( + api_url, headers=headers, method="GET" + ) + # Send the request and get the response + with urllib.request.urlopen(req, timeout=60) as response: + result = json.loads(response.read().decode("utf-8")) - return (sub_id, result) + return (sub_id, result) + except Exception as e: + if i == RETRY - 1: + raise + print(f"C3 failed with exception: {e}", flush=True) + time.sleep(min(i * 10, 60)) def slugify(_string: str): if not _string: return _string + _string = _string.replace("com.canonical.certification::", "") valid_chars = frozenset(f"_{string.ascii_letters}{string.digits}") # Python identifiers cannot start with a digit @@ -75,12 +88,11 @@ def _f(x: dict) -> bool: ] -def main(): - args = parse_args() +def do_test_id(filter_arg, csvs, test_id): lines_of_interest = [] - filter_f = get_filter(args.filter, args.test_id) + filter_f = get_filter(filter_arg, test_id) - for f_path in args.csvs: + for f_path in csvs: with open(f_path) as f: reader = csv.DictReader(f, delimiter=",") lines_of_interest += list(filter(filter_f, reader)) @@ -109,7 +121,7 @@ def main(): } | relevant_template_id relevant_ids = relevant_ids - {None, ""} - file_name = slugify(args.test_id) + file_name = slugify(test_id) with open(f"{file_name}.url", "w+") as f: f.writelines("\n".join(sub_links)) @@ -193,5 +205,13 @@ def main(): writer.writerows(result_rows) +def main(): + args = parse_args() + tot = len(args.test_ids) + for i, test_id in enumerate(args.test_ids, 1): + print(f"Doing [{i}/{tot}]: {test_id}") + do_test_id(args.filter, args.csvs, test_id) + + if __name__ == "__main__": main()