ci,crnm: migrate colorama to rich

The links in the console are broken depending on the console type; for example,
when it runs within a GitLab job. This can be improved using rich. But as we
have a dependency on colorama too, we can migrate all the coloring to use this
other library too.

Signed-off-by: Sergi Blanch Torne <sergi.blanch.torne@collabora.com>
Part-of: <https://gitlab.freedesktop.org/mesa/mesa/-/merge_requests/37454>
This commit is contained in:
Sergi Blanch Torne 2025-09-18 12:31:49 +02:00 committed by Marge Bot
parent a6b11b58d9
commit 51c3f56aa3
5 changed files with 102 additions and 85 deletions

View file

@ -26,7 +26,6 @@ from typing import Callable, Dict, TYPE_CHECKING, Iterable, Literal, Optional, T
import gitlab
import gitlab.v4.objects
from colorama import Fore, Style
from gitlab_common import (
GITLAB_URL,
TOKEN_DIR,
@ -34,11 +33,11 @@ from gitlab_common import (
get_gitlab_project,
get_token_from_default_dir,
pretty_duration,
print_once,
read_token,
wait_for_pipeline,
)
from gitlab_gql import GitlabGQL, create_job_needs_dag, filter_dag, print_dag, print_formatted_list
from rich.console import Console
if TYPE_CHECKING:
from gitlab_gql import Dag
@ -47,16 +46,13 @@ REFRESH_WAIT_LOG = 10
REFRESH_WAIT_JOBS = 6
MAX_ENABLE_JOB_ATTEMPTS = 3
URL_START = "\033]8;;"
URL_END = "\033]8;;\a"
STATUS_COLORS = {
"created": "",
"running": Fore.BLUE,
"success": Fore.GREEN,
"failed": Fore.RED,
"canceled": Fore.MAGENTA,
"canceling": Fore.MAGENTA,
"running": "[blue]",
"success": "[green]",
"failed": "[red]",
"canceled": "[magenta]",
"canceling": "[magenta]",
"manual": "",
"pending": "",
"skipped": "",
@ -65,6 +61,9 @@ STATUS_COLORS = {
COMPLETED_STATUSES = frozenset({"success", "failed"})
RUNNING_STATUSES = frozenset({"created", "pending", "running"})
console = Console(highlight=False)
print = console.print
def print_job_status(
job: gitlab.v4.objects.ProjectPipelineJob,
@ -86,13 +85,12 @@ def print_job_status(
duration = job_duration(job)
print_once(
STATUS_COLORS[job.status]
+ f"{jtype:{type_field_pad}} " # U+1F78B Round target
+ link2print(job.web_url, job.name, name_field_pad)
+ (f" has new status: {job.status}" if new_status else f" {job.status}")
+ (f" ({pretty_duration(duration)})" if job.started_at else "")
+ Style.RESET_ALL
print(
f"{STATUS_COLORS[job.status]}"
f"{jtype:{type_field_pad}} " # U+1F78B Round target
f"{link2print(job.web_url, job.name, name_field_pad)} "
f"{f"has new status: {job.status} " if new_status else f"{job.status}"} "
f"{f"({pretty_duration(duration)})" if job.started_at else ""}"
)
@ -246,9 +244,8 @@ def monitor_pipeline(
continue
if jobs_waiting:
print(f"{Fore.YELLOW}Waiting for jobs to update status:")
print_formatted_list(jobs_waiting, indentation=8)
print(Style.RESET_ALL, end='')
print(f"[yellow]Waiting for jobs to update status:")
print_formatted_list(jobs_waiting, indentation=8, color="[yellow]")
pretty_wait(REFRESH_WAIT_JOBS)
continue
@ -270,10 +267,7 @@ def monitor_pipeline(
and not RUNNING_STATUSES.intersection(target_statuses.values())
):
print(
Fore.RED,
"Target in skipped state, aborting. Failed dependencies:",
deps_failed,
Fore.RESET,
f"[red]Target in skipped state, aborting. Failed dependencies:{deps_failed}"
)
return None, 1, execution_times
@ -349,9 +343,7 @@ def enable_job(
type_field_pad = len(jtype) if len(jtype) > type_field_pad else type_field_pad
name_field_pad = len(job_name) if len(job_name) > name_field_pad else name_field_pad
print(
Fore.MAGENTA +
f"{jtype:{type_field_pad}} {job.name:{name_field_pad}} manually enabled" +
Style.RESET_ALL
f"[magenta]{jtype:{type_field_pad}} {job.name:{name_field_pad}} manually enabled"
)
return True
@ -417,7 +409,7 @@ def print_log(
printed_lines = len(lines)
if job.status in COMPLETED_STATUSES:
print(Fore.GREEN + f"Job finished: {job.web_url}" + Style.RESET_ALL)
print(f"[green]Job finished: {job.web_url}")
return
pretty_wait(REFRESH_WAIT_LOG)
@ -552,15 +544,13 @@ def print_detected_jobs(
) -> None:
def print_job_set(color: str, kind: str, job_set: Iterable[str]):
job_list = list(job_set)
print(color + f"Running {len(job_list)} {kind} jobs:")
print_formatted_list(job_list, indentation=8)
print(Style.RESET_ALL)
print(f"{color}Running {len(job_list)} {kind} jobs:")
print_formatted_list(job_list, indentation=8, color=color)
print(Fore.YELLOW + "Detected target job and its dependencies:")
print_dag(target_dep_dag, indentation=8)
print(Style.RESET_ALL)
print_job_set(Fore.MAGENTA, "dependency", dependency_jobs)
print_job_set(Fore.BLUE, "target", target_jobs)
print("[yellow]Detected target job and its dependencies:")
print_dag(target_dep_dag, indentation=8, color="[yellow]")
print_job_set("[magenta]", "dependency", dependency_jobs)
print_job_set("[blue]", "target", target_jobs)
def find_dependencies(
@ -601,7 +591,7 @@ def find_dependencies(
target_dep_dag = filter_dag(dag, job_filter)
if not target_dep_dag:
print(Fore.RED + "The job(s) were not found in the pipeline." + Fore.RESET)
print("[red]The job(s) were not found in the pipeline.")
sys.exit(1)
dependency_jobs = set(chain.from_iterable(d["needs"] for d in target_dep_dag.values()))
@ -638,15 +628,16 @@ def __job_duration_record(dict_item: tuple) -> str:
"""
job_id = f"{dict_item[0]}" # dictionary key
job_duration, job_status, job_url = dict_item[1] # dictionary value, the tuple
return (f"{STATUS_COLORS[job_status]}"
f"{link2print(job_url, job_id)}: {pretty_duration(job_duration):>8}"
f"{Style.RESET_ALL}")
return (
f"{STATUS_COLORS[job_status]}"
f"{link2print(job_url, job_id)}: {pretty_duration(job_duration):>8}"
)
def link2print(url: str, text: str, text_pad: int = 0) -> str:
text = str(text)
text_pad = len(text) if text_pad < 1 else text_pad
return f"{URL_START}{url}\a{text:{text_pad}}{URL_END}"
return f"[link={url}]{text:{text_pad}}[/link]"
def main() -> None:
@ -711,7 +702,7 @@ def main() -> None:
target = '|'.join(args.target)
target = target.strip()
print("🞋 target job: " + Fore.BLUE + target + Style.RESET_ALL) # U+1F78B Round target
print(f"🞋 target job: [blue]{target}") # U+1F78B Round target
# Implicitly include `parallel:` jobs
target = f'({target})' + r'( \d+/\d+)?'
@ -721,18 +712,18 @@ def main() -> None:
include_stage = '|'.join(args.include_stage)
include_stage = include_stage.strip()
print("🞋 target from stages: " + Fore.BLUE + include_stage + Style.RESET_ALL) # U+1F78B Round target
print(f"🞋 target from stages: [blue]{include_stage}") # U+1F78B Round target
include_stage_regex = re.compile(include_stage)
exclude_stage = '|'.join(args.exclude_stage)
exclude_stage = exclude_stage.strip()
print("🞋 target excluding stages: " + Fore.BLUE + exclude_stage + Style.RESET_ALL) # U+1F78B Round target
print(f"🞋 target excluding stages: [blue]{exclude_stage}") # U+1F78B Round target
exclude_stage_regex = re.compile(exclude_stage)
print("🞋 target jobs with tags: " + Fore.BLUE + str(args.job_tags) + Style.RESET_ALL) # U+1F78B Round target
print(f"🞋 target jobs with tags: [blue]{str(args.job_tags)}") # U+1F78B Round target
job_tags_regexes = [re.compile(job_tag) for job_tag in args.job_tags]
def job_filter(

View file

@ -22,6 +22,7 @@ from gitlab_common import get_token_from_default_dir
from gql import Client, gql
from gql.transport.requests import RequestsHTTPTransport
from graphql import DocumentNode
from rich.console import Console
DEFAULT_TERMINAL_SIZE: int = 80 # columns
@ -40,6 +41,9 @@ Dag = dict[str, DagNode]
StageSeq = OrderedDict[str, set[str]]
console = Console(highlight=False)
print = console.print
def get_project_root_dir():
root_path = Path(__file__).parent.parent.parent.resolve()
@ -343,13 +347,13 @@ def filter_dag(dag: Dag, job_filter: callable) -> Dag:
})
def print_dag(dag: Dag, indentation: int = 0) -> None:
def print_dag(dag: Dag, indentation: int = 0, color: str = "") -> None:
for job, data in sorted(dag.items()):
print(f"{' '*indentation}{job}:")
print_formatted_list(list(data['needs']), indentation=indentation+8)
print(f"{color}{' '*indentation}{job}:")
print_formatted_list(list(data['needs']), indentation=indentation+8, color=color)
def print_formatted_list(elements: list[str], indentation: int = 0) -> None:
def print_formatted_list(elements: list[str], indentation: int = 0, color: str = "") -> None:
"""
When a list of elements is going to be printed, if it is longer than one line, reformat it to be multiple
lines with a 'ls' command style.
@ -364,7 +368,7 @@ def print_formatted_list(elements: list[str], indentation: int = 0) -> None:
except OSError:
h_size = DEFAULT_TERMINAL_SIZE
if indentation + sum(len(element) for element in elements) + (len(elements)*2) < h_size: # fits in one line
print(f"{' '*indentation}{', '.join([element for element in elements])}")
print(f"{color}{' '*indentation}{', '.join([element for element in elements])}")
return
column_separator_size = 2
column_width: int = len(max(elements, key=len)) + column_separator_size
@ -375,7 +379,7 @@ def print_formatted_list(elements: list[str], indentation: int = 0) -> None:
print(' '*indentation, end='')
for column in range(len(line)):
if line[column] is not None:
print(f"{line[column]:<{column_width}}", end='')
print(f"{color}{line[column]:<{column_width}}", end='')
print()

View file

@ -18,14 +18,16 @@ import io
from tabulate import tabulate
import gitlab
from colorama import Fore, Style
from gitlab_common import read_token
from rich import print
MARGE_BOT_USER_ID = 9716
def print_failures_csv(id):
url = 'https://gitlab.freedesktop.org/mesa/mesa/-/jobs/' + str(id) + '/artifacts/raw/results/failures.csv'
url = "https://gitlab.freedesktop.org/mesa/mesa"\
f"/-/jobs/{id}/artifacts/raw/results/failures.csv"
missing: int = 0
MAX_MISS: int = 20
try:
@ -37,25 +39,31 @@ def print_failures_csv(id):
for line in data[:]:
if line[1] == "UnexpectedImprovement(Pass)":
line[1] = Fore.GREEN + line[1] + Style.RESET_ALL
line[1] = f"[green]{line[1]}[/green]"
elif line[1] == "UnexpectedImprovement(Fail)":
line[1] = Fore.YELLOW + line[1] + Style.RESET_ALL
line[1] = f"[yellow]{line[1]}[/yellow]"
elif line[1] == "Crash" or line[1] == "Fail":
line[1] = Fore.RED + line[1] + Style.RESET_ALL
line[1] = f" [red]{line[1]}[/red]"
elif line[1] == "Missing":
if missing > MAX_MISS:
data.remove(line)
continue
missing += 1
line[1] = Fore.YELLOW + line[1] + Style.RESET_ALL
line[1] = f"[yellow]{line[1]}[/yellow]"
elif line[1] == "Fail":
line[1] = Fore.RED + line[1] + Style.RESET_ALL
line[1] = f"[red]{line[1]}[/red]"
else:
line[1] = Fore.WHITE + line[1] + Style.RESET_ALL
line[1] = f"[white]{line[1]}[/white]"
if missing > MAX_MISS:
data.append([Fore.RED + f"... more than {MAX_MISS} missing tests, something crashed?", "Missing" + Style.RESET_ALL])
headers = ["Test ", "Result"]
data.append(
[
f"[red]... more than {MAX_MISS} missing tests, "
"something crashed?[/red]",
"[red]Missing[/red]"
]
)
headers = [f"Test{"":<75}", "Result"]
print(tabulate(data, headers, tablefmt="plain"))
except Exception:
pass
@ -83,7 +91,8 @@ def parse_args() -> None:
parser.add_argument(
"--token",
metavar="token",
help="force GitLab token, otherwise it's read from ~/.config/gitlab-token",
help="force GitLab token, "
"otherwise it's read from ~/.config/gitlab-token",
)
return parser.parse_args()
@ -91,31 +100,47 @@ def parse_args() -> None:
if __name__ == "__main__":
args = parse_args()
token = read_token(args.token)
gl = gitlab.Gitlab(url="https://gitlab.freedesktop.org", private_token=token)
gl = gitlab.Gitlab(
url="https://gitlab.freedesktop.org",
private_token=token,
)
project = gl.projects.get("mesa/mesa")
print(
"\u001b]8;;https://gitlab.freedesktop.org/mesa/mesa/-/pipelines?page=1&scope=all&source=schedule\u001b\\Scheduled pipelines overview\u001b]8;;\u001b\\"
"[link=https://gitlab.freedesktop.org/mesa/mesa/-/pipelines?"
"page=1&scope=all&source=schedule]Scheduled pipelines overview[/link]"
)
pipelines = project.pipelines.list(
source="schedule", ordered_by="created_at", sort="desc", page=1, per_page=2
source="schedule",
ordered_by="created_at",
sort="desc",
page=1,
per_page=2,
)
print(
f"Old pipeline: {pipelines[1].created_at}\t\u001b]8;;{pipelines[1].web_url}\u001b\\{pipelines[1].status}\u001b]8;;\u001b\\\t{pipelines[1].sha}"
"Old pipeline:"
f" {pipelines[1].created_at}"
f"\t[link={pipelines[1].web_url}]{pipelines[1].status}[/link]"
f"\t{pipelines[1].sha}"
)
print(
f"New pipeline: {pipelines[0].created_at}\t\u001b]8;;{pipelines[0].web_url}\u001b\\{pipelines[0].status}\u001b]8;;\u001b\\\t{pipelines[0].sha}"
"New pipeline:"
f" {pipelines[0].created_at}"
f"\t[link={pipelines[0].web_url}]{pipelines[0].status}[/link]"
f"\t{pipelines[0].sha}"
)
print(
f"\nWebUI visual compare: https://gitlab.freedesktop.org/mesa/mesa/-/compare/{pipelines[1].sha}...{pipelines[0].sha}\n"
"\nWebUI visual compare: "
"https://gitlab.freedesktop.org/mesa/mesa/-/compare/"
f"{pipelines[1].sha}...{pipelines[0].sha}\n"
)
# regex part
if args.target:
target = "|".join(args.target)
target = target.strip()
print("🞋 jobs: " + Fore.BLUE + target + Style.RESET_ALL)
print(f"🞋 jobs: [blue]{target}[/blue]")
target = f"({target})" + r"( \d+/\d+)?"
else:
@ -147,17 +172,14 @@ if __name__ == "__main__":
previously_failed_job = job_failed_before(old_failed_jobs, job)
if previously_failed_job:
print(
Fore.YELLOW
+ f":: \u001b]8;;{job.web_url}\u001b\\{job.name}\u001b]8;;\u001b\\"
+ Fore.MAGENTA
+ f" \u001b]8;;{previously_failed_job.web_url}\u001b\\(previous run)\u001b]8;;\u001b\\"
+ Style.RESET_ALL
f"[yellow]"
f" :: [link={job.web_url}]{job.name}[/link][/yellow]"
f"[magenta]"
f" [link={previously_failed_job.web_url}](previous run)[/link]"
)
else:
print(
Fore.RED
+ f":: \u001b]8;;{job.web_url}\u001b\\{job.name}\u001b]8;;\u001b\\"
+ Style.RESET_ALL
f"[red]:: [link={job.web_url}]{job.name}[/link]"
)
print_failures_csv(job.id)
@ -168,7 +190,7 @@ if __name__ == "__main__":
commit = project.commits.get(pipelines[0].sha)
while True:
print(
f"{commit.id} \u001b]8;;{commit.web_url}\u001b\\{commit.title}\u001b]8;;\u001b\\"
f"{commit.id} [link={commit.web_url}]{commit.title}[/link]"
)
if commit.id == pipelines[1].sha:
break

View file

@ -1,6 +1,5 @@
-r requirements-lava.txt
PyYAML==6.*
colorama==0.4.*
filecache==0.81
flake8==7.*
gql==3.*
@ -9,6 +8,7 @@ pandas==2.*
plotly==5.*
python-dateutil==2.*
python-gitlab==4.*
rich==14.1.*
ruamel.yaml.clib==0.2.*
ruamel.yaml==0.17.*
tabulate==0.9.*

View file

@ -19,9 +19,9 @@ import sys
from ruamel.yaml import YAML
import gitlab
from colorama import Fore, Style
from gitlab_common import (get_gitlab_project, read_token, wait_for_pipeline,
get_gitlab_pipeline_from_url, TOKEN_DIR, get_token_from_default_dir)
from rich import print
DESCRIPTION_FILE = "export PIGLIT_REPLAY_DESCRIPTION_FILE=.*/install/(.*)$"
@ -53,7 +53,7 @@ def gather_results(
dev_name = device_name.group(1)
if not filename or not dev_name:
print(Fore.RED + "Couldn't find device name or YML file in the logs!" + Style.RESET_ALL)
print("[red]Couldn't find device name or YML file in the logs!")
return
print(f"👁 Found {dev_name} and file {filename}")
@ -86,11 +86,11 @@ def gather_results(
checksum: str = value['images'][0]['checksum_render']
if not checksum:
print(Fore.RED + f"{dev_name}: {trace}: checksum is missing! Crash?" + Style.RESET_ALL)
print(f"[red]{dev_name}: {trace}: checksum is missing! Crash?")
continue
if checksum == "error":
print(Fore.RED + f"{dev_name}: {trace}: crashed" + Style.RESET_ALL)
print(f"[red]{dev_name}: {trace}: crashed")
continue
if target['traces'][trace][dev_name].get('checksum') == checksum:
@ -99,11 +99,11 @@ def gather_results(
if "label" in target['traces'][trace][dev_name]:
print(
f"{dev_name}: {trace}: please verify that label "
f"{Fore.BLUE}{target['traces'][trace][dev_name]['label']}{Style.RESET_ALL} "
f"[blue]{target['traces'][trace][dev_name]['label']}[/blue] "
"is still valid"
)
print(Fore.GREEN + f'{dev_name}: {trace}: checksum updated' + Style.RESET_ALL)
print(f"[green]{dev_name}: {trace}: checksum updated")
target['traces'][trace][dev_name]['checksum'] = checksum
with open(traces_file[0], 'w', encoding='utf-8') as target_file: