ci: run_n_monitor, arguments review and unicode

Coding style homogenous for all the methods in the file. Document
unicode symbols used.

Signed-off-by: Sergi Blanch Torne <sergi.blanch.torne@collabora.com>
Reviewed-by: Daniel Stone <daniels@collabora.com>
Part-of: <https://gitlab.freedesktop.org/mesa/mesa/-/merge_requests/29444>

bin/ci/ci_run_n_monitor.py

@@ -63,7 +63,10 @@ COMPLETED_STATUSES = {"success", "failed"}
 RUNNING_STATUSES = {"created", "pending", "running"}
 
 
-def print_job_status(job, new_status=False) -> None:
+def print_job_status(
+    job: gitlab.v4.objects.ProjectPipelineJob,
+    new_status: bool = False,
+) -> None:
     """It prints a nice, colored job status with a link to the job."""
     if job.status in {"canceled", "canceling"}:
         return
@@ -75,7 +78,7 @@ def print_job_status(job, new_status=False) -> None:
 
     print(
         STATUS_COLORS[job.status]
-        + "🞋 job "
+        + "🞋 job "  # U+1F78B Round target
         + link2print(job.web_url, job.name)
         + (f" has new status: {job.status}" if new_status else f" :: {job.status}")
         + (f" ({pretty_duration(duration)})" if job.started_at else "")
@@ -99,24 +102,24 @@ def job_duration(job: gitlab.v4.objects.ProjectPipelineJob) -> float:
 def pretty_wait(sec: int) -> None:
     """shows progressbar in dots"""
     for val in range(sec, 0, -1):
-        print(f"⏲  {val} seconds", end="\r")
+        print(f"⏲  {val} seconds", end="\r")  # U+23F2 Timer clock
         time.sleep(1)
 
 
 def monitor_pipeline(
-    project,
+    project: gitlab.v4.objects.Project,
-    pipeline,
+    pipeline: gitlab.v4.objects.ProjectPipeline,
     target_jobs_regex: re.Pattern,
-    dependencies,
+    dependencies: set[str],
     force_manual: bool,
     stress: int,
 ) -> tuple[Optional[int], Optional[int], Dict[str, Dict[int, Tuple[float, str, str]]]]:
     """Monitors pipeline and delegate canceling jobs"""
     statuses: dict[str, str] = defaultdict(str)
     target_statuses: dict[str, str] = defaultdict(str)
-    stress_status_counter = defaultdict(lambda: defaultdict(int))
+    stress_status_counter: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
     execution_times = defaultdict(lambda: defaultdict(tuple))
-    target_id = None
+    target_id: int = -1
 
     # Pre-populate the stress status counter for already completed target jobs.
     if stress:
@@ -215,10 +218,10 @@ def monitor_pipeline(
 
 def get_pipeline_job(
     pipeline: gitlab.v4.objects.ProjectPipeline,
-    id: int,
+    job_id: int,
 ) -> gitlab.v4.objects.ProjectPipelineJob:
     pipeline_jobs = pipeline.jobs.list(all=True)
-    return [j for j in pipeline_jobs if j.id == id][0]
+    return [j for j in pipeline_jobs if j.id == job_id][0]
 
 
 def enable_job(
@@ -246,27 +249,33 @@ def enable_job(
         job = get_pipeline_job(pipeline, pjob.id)
 
     if action_type == "target":
-        jtype = "🞋 "
+        jtype = "🞋"  # U+1F78B Round target
     elif action_type == "retry":
-        jtype = "↻"
+        jtype = "↻"  # U+21BB Clockwise open circle arrow
     else:
-        jtype = "(dependency)"
+        jtype = "↪"  # U+21AA Left Arrow Curving Right
 
     print(Fore.MAGENTA + f"{jtype} job {job.name} manually enabled" + Style.RESET_ALL)
 
     return job
 
 
-def cancel_job(project, job) -> None:
+def cancel_job(
+    project: gitlab.v4.objects.Project,
+    job: gitlab.v4.objects.ProjectPipelineJob
+) -> None:
     """Cancel GitLab job"""
     if job.status not in RUNNING_STATUSES:
         return
     pjob = project.jobs.get(job.id, lazy=True)
     pjob.cancel()
-    print(f"♲ {job.name}", end=" ")
+    print(f"♲ {job.name}", end=" ")  # U+2672 Universal Recycling symbol
 
 
-def cancel_jobs(project, to_cancel) -> None:
+def cancel_jobs(
+    project: gitlab.v4.objects.Project,
+    to_cancel: list
+) -> None:
     """Cancel unwanted GitLab jobs"""
     if not to_cancel:
         return
@@ -277,7 +286,10 @@ def cancel_jobs(project, to_cancel) -> None:
     print()
 
 
-def print_log(project, job_id) -> None:
+def print_log(
+    project: gitlab.v4.objects.Project,
+    job_id: int
+) -> None:
     """Print job log into output"""
     printed_lines = 0
     while True:
@@ -295,7 +307,7 @@ def print_log(project, job_id) -> None:
         pretty_wait(REFRESH_WAIT_LOG)
 
 
-def parse_args() -> None:
+def parse_args() -> argparse.Namespace:
     """Parse args"""
     parser = argparse.ArgumentParser(
         description="Tool to trigger a subset of container jobs "
@@ -363,7 +375,9 @@ def parse_args() -> None:
 
 
 def print_detected_jobs(
-    target_dep_dag: "Dag", dependency_jobs: Iterable[str], target_jobs: Iterable[str]
+    target_dep_dag: "Dag",
+    dependency_jobs: Iterable[str],
+    target_jobs: Iterable[str],
 ) -> None:
     def print_job_set(color: str, kind: str, job_set: Iterable[str]):
         print(
@@ -380,10 +394,12 @@ def print_detected_jobs(
     print_job_set(Fore.BLUE, "target", target_jobs)
 
 
-def find_dependencies(token: str | None,
+def find_dependencies(
-                      target_jobs_regex: re.Pattern,
+    token: str | None,
-                      project_path: str,
+    target_jobs_regex: re.Pattern,
-                      iid: int) -> set[str]:
+    project_path: str,
+    iid: int
+) -> set[str]:
     """
     Find the dependencies of the target jobs in a GitLab pipeline.
 
@@ -427,10 +443,10 @@ def print_monitor_summary(
     """Summary of the test execution"""
     t_end = time.perf_counter()
     spend_minutes = (t_end - t_start) / 60
-    print(f"⏲ Duration of script execution: {spend_minutes:0.1f} minutes")
+    print(f"⏲ Duration of script execution: {spend_minutes:0.1f} minutes")  # U+23F2 Timer clock
     if len(execution_collection) == 0:
         return
-    print(f"⏲ Jobs execution times:")
+    print(f"⏲ Jobs execution times:")  # U+23F2 Timer clock
     job_names = list(execution_collection.keys())
     job_names.sort()
     name_field_pad = len(max(job_names, key=len)) + 2
@@ -457,7 +473,7 @@ def link2print(url: str, text: str) -> str:
     return f"{URL_START}{url}\a{text}{URL_END}"
 
 
-if __name__ == "__main__":
+def main() -> None:
     try:
         t_start = time.perf_counter()
 
@@ -519,8 +535,7 @@ if __name__ == "__main__":
         target = '|'.join(args.target)
         target = target.strip()
 
-        deps = set()
+        print("🞋 job: " + Fore.BLUE + target + Style.RESET_ALL)  # U+1F78B Round target
-        print("🞋 job: " + Fore.BLUE + target + Style.RESET_ALL)
 
         # Implicitly include `parallel:` jobs
         target = f'({target})' + r'( \d+/\d+)?'
@@ -545,3 +560,7 @@ if __name__ == "__main__":
         sys.exit(ret)
     except KeyboardInterrupt:
         sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()