tasks/libs/pipeline_tools.py
def print_job_status(job):
    """
    Prints notifications about job changes.
    """

    def print_job(name, stage, color, date, duration, status, link):
        print(
            color_message(
                f"[{date}] Job {name} (stage: {stage}) {status} [job duration: {duration // 60:.0f}m{duration % 60:2.0f}s]\n{link}".strip(),
                color,
            )
        )

    def print_retry(name, date):
        print(color_message(f"[{date}] Job {name} was retried", "grey"))

    name = job['name']
    stage = job['stage']
    allow_failure = job['allow_failure']
    duration = job['duration']
    date = job['finished_at']  # Date that is printed in the console log. In most cases, it's when the job finished.
    status = job['status']  # Gitlab job status
    job_status = None  # Status string printed in the console
    link = ''  # Link to the pipeline. Only filled for failing jobs, to be able to quickly go to the failing job.
    color = 'grey'  # Log output color

    # A None duration is set by Gitlab when the job gets canceled before it was started.
    # In that case, set a duration of 0s.
    if duration is None:
        duration = 0

    if status == 'success':
        job_status = 'succeeded'
        color = 'green'
    elif status == 'failed':
        if allow_failure:
            job_status = 'failed (allowed to fail)'
            color = 'orange'
        else:
            job_status = 'failed'
            color = 'red'
            link = f"Link: {job['web_url']}"
            # Only notify on real (not retried) failures
            # Best-effort, as there can be situations where the retried
            # job didn't get created yet
            if job.get('retried_old', None) is None:
                notify("Job failure", f"Job {name} failed.")
    elif status == 'canceled':
        job_status = 'was canceled'
        color = 'grey'
    elif status == 'running':
        job_status = 'started running'
        date = job['started_at']
        color = 'blue'
    else:
        return

    # Some logic to print the retry message in the correct order (before the new job or after the old job)
    if job.get('retried_new', None) is not None:
        print_retry(name, job['created_at'])
    print_job(name, stage, color, date, duration, job_status, link)
    if job.get('retried_old', None) is not None:
        print_retry(name, job['retried_created_at'])

    Prints notifications about job changes.
    
Prints notifications about job changes.
Prints notifications about job changes .
def print_job_status(job):
    def print_job(name, stage, color, date, duration, status, link):
        print(
            color_message(
                f"[{date}] Job {name} (stage: {stage}) {status} [job duration: {duration // 60:.0f}m{duration % 60:2.0f}s]\n{link}".strip(),
                color,
            )
        )
    def print_retry(name, date):
        print(color_message(f"[{date}] Job {name} was retried", "grey"))
    name = job['name']
    stage = job['stage']
    allow_failure = job['allow_failure']
    duration = job['duration']
    date = job['finished_at']  
    status = job['status']  
    job_status = None  
    link = ''  
    color = 'grey'  
    if duration is None:
        duration = 0
    if status == 'success':
        job_status = 'succeeded'
        color = 'green'
    elif status == 'failed':
        if allow_failure:
            job_status = 'failed (allowed to fail)'
            color = 'orange'
        else:
            job_status = 'failed'
            color = 'red'
            link = f"Link: {job['web_url']}"
            if job.get('retried_old', None) is None:
                notify("Job failure", f"Job {name} failed.")
    elif status == 'canceled':
        job_status = 'was canceled'
        color = 'grey'
    elif status == 'running':
        job_status = 'started running'
        date = job['started_at']
        color = 'blue'
    else:
        return
    if job.get('retried_new', None) is not None:
        print_retry(name, job['created_at'])
    print_job(name, stage, color, date, duration, job_status, link)
    if job.get('retried_old', None) is not None:
        print_retry(name, job['retried_created_at'])
def print_job_status ( job ) : def print_job ( name , stage , color , date , duration , status , link ) : print ( color_message ( f"[{date}] Job {name} (stage: {stage}) {status} [job duration: {duration // 60:.0f}m{duration % 60:2.0f}s]\n{link}" . strip ( ) , color , ) ) def print_retry ( name , date ) : print ( color_message ( f"[{date}] Job {name} was retried" , "grey" ) ) name = job [ 'name' ] stage = job [ 'stage' ] allow_failure = job [ 'allow_failure' ] duration = job [ 'duration' ] date = job [ 'finished_at' ] status = job [ 'status' ] job_status = None link = '' color = 'grey' if duration is None : duration = 0 if status == 'success' : job_status = 'succeeded' color = 'green' elif status == 'failed' : if allow_failure : job_status = 'failed (allowed to fail)' color = 'orange' else : job_status = 'failed' color = 'red' link = f"Link: {job['web_url']}" if job . get ( 'retried_old' , None ) is None : notify ( "Job failure" , f"Job {name} failed." ) elif status == 'canceled' : job_status = 'was canceled' color = 'grey' elif status == 'running' : job_status = 'started running' date = job [ 'started_at' ] color = 'blue' else : return if job . get ( 'retried_new' , None ) is not None : print_retry ( name , job [ 'created_at' ] ) print_job ( name , stage , color , date , duration , job_status , link ) if job . get ( 'retried_old' , None ) is not None : print_retry ( name , job [ 'retried_created_at' ] )
Prints notifications about job changes.
Prints notifications about job changes .
"""
    Prints notifications about job changes.
    """ # Date that is printed in the console log. In most cases, it's when the job finished. # Gitlab job status # Status string printed in the console # Link to the pipeline. Only filled for failing jobs, to be able to quickly go to the failing job. # Log output color # A None duration is set by Gitlab when the job gets canceled before it was started. # In that case, set a duration of 0s. # Only notify on real (not retried) failures # Best-effort, as there can be situations where the retried # job didn't get created yet # Some logic to print the retry message in the correct order (before the new job or after the old job)
job