tasks/libs/pipeline_tools.py
def pipeline_status(gitlab, pipeline_id, job_status):
    """
    Checks the pipeline status and updates job statuses.
    """
    jobs = gitlab.all_jobs(pipeline_id)

    job_status = update_job_status(jobs, job_status)

    # Check pipeline status
    pipeline = gitlab.pipeline(pipeline_id)
    pipestatus = pipeline["status"].lower().strip()
    ref = pipeline["ref"]

    if pipestatus == "success":
        print(
            color_message(
                f"Pipeline https://gitlab.ddbuild.io/{gitlab.project_name}/pipelines/{pipeline_id} for {ref} succeeded",
                "green",
            )
        )
        notify("Pipeline success", f"Pipeline {pipeline_id} for {ref} succeeded.")
        return True, job_status

    if pipestatus == "failed":
        print(
            color_message(
                f"Pipeline https://gitlab.ddbuild.io/{gitlab.project_name}/pipelines/{pipeline_id} for {ref} failed",
                "red",
            )
        )
        notify("Pipeline failure", f"Pipeline {pipeline_id} for {ref} failed.")
        return True, job_status

    if pipestatus == "canceled":
        print(
            color_message(
                f"Pipeline https://gitlab.ddbuild.io/{gitlab.project_name}/pipelines/{pipeline_id} for {ref} was canceled",
                "grey",
            )
        )
        notify("Pipeline canceled", f"Pipeline {pipeline_id} for {ref} was canceled.")
        return True, job_status

    if pipestatus not in ["created", "running", "pending"]:
        raise ErrorMsg(f"Error: pipeline status {pipestatus.title()}")

    return False, job_status

    Checks the pipeline status and updates job statuses.
    
Checks the pipeline status and updates job statuses.
Checks the pipeline status and updates job statuses .
def pipeline_status(gitlab, pipeline_id, job_status):
    jobs = gitlab.all_jobs(pipeline_id)
    job_status = update_job_status(jobs, job_status)
    pipeline = gitlab.pipeline(pipeline_id)
    pipestatus = pipeline["status"].lower().strip()
    ref = pipeline["ref"]
    if pipestatus == "success":
        print(
            color_message(
                f"Pipeline https://gitlab.ddbuild.io/{gitlab.project_name}/pipelines/{pipeline_id} for {ref} succeeded",
                "green",
            )
        )
        notify("Pipeline success", f"Pipeline {pipeline_id} for {ref} succeeded.")
        return True, job_status
    if pipestatus == "failed":
        print(
            color_message(
                f"Pipeline https://gitlab.ddbuild.io/{gitlab.project_name}/pipelines/{pipeline_id} for {ref} failed",
                "red",
            )
        )
        notify("Pipeline failure", f"Pipeline {pipeline_id} for {ref} failed.")
        return True, job_status
    if pipestatus == "canceled":
        print(
            color_message(
                f"Pipeline https://gitlab.ddbuild.io/{gitlab.project_name}/pipelines/{pipeline_id} for {ref} was canceled",
                "grey",
            )
        )
        notify("Pipeline canceled", f"Pipeline {pipeline_id} for {ref} was canceled.")
        return True, job_status
    if pipestatus not in ["created", "running", "pending"]:
        raise ErrorMsg(f"Error: pipeline status {pipestatus.title()}")
    return False, job_status
def pipeline_status ( gitlab , pipeline_id , job_status ) : jobs = gitlab . all_jobs ( pipeline_id ) job_status = update_job_status ( jobs , job_status ) pipeline = gitlab . pipeline ( pipeline_id ) pipestatus = pipeline [ "status" ] . lower ( ) . strip ( ) ref = pipeline [ "ref" ] if pipestatus == "success" : print ( color_message ( f"Pipeline https://gitlab.ddbuild.io/{gitlab.project_name}/pipelines/{pipeline_id} for {ref} succeeded" , "green" , ) ) notify ( "Pipeline success" , f"Pipeline {pipeline_id} for {ref} succeeded." ) return True , job_status if pipestatus == "failed" : print ( color_message ( f"Pipeline https://gitlab.ddbuild.io/{gitlab.project_name}/pipelines/{pipeline_id} for {ref} failed" , "red" , ) ) notify ( "Pipeline failure" , f"Pipeline {pipeline_id} for {ref} failed." ) return True , job_status if pipestatus == "canceled" : print ( color_message ( f"Pipeline https://gitlab.ddbuild.io/{gitlab.project_name}/pipelines/{pipeline_id} for {ref} was canceled" , "grey" , ) ) notify ( "Pipeline canceled" , f"Pipeline {pipeline_id} for {ref} was canceled." ) return True , job_status if pipestatus not in [ "created" , "running" , "pending" ] : raise ErrorMsg ( f"Error: pipeline status {pipestatus.title()}" ) return False , job_status
Checks the pipeline status and updates job statuses.
Checks the pipeline status and updates job statuses .
"""
    Checks the pipeline status and updates job statuses.
    """ # Check pipeline status
gitlab
pipeline_id
job_status