microsoft--promptflow/scripts/check_enforcer/check_enforcer.py

# Enforce the check of pipelines.
# This script will get the diff of the current branch and main branch, calculate the pipelines that should be triggered.
# Then it will check if the triggered pipelines are successful. This script will loop for 30*loop-times seconds at most.

# How many checks are triggered:
# 1. sdk checks: sdk_cli_tests, sdk_cli_azure_test, sdk_cli_global_config_tests are triggered.
# 2. examples checks: this script calculate the path filters and decide what should be triggered.

# Trigger checks and return the status of the checks:
# 1. If examples are not correctly generated, fail.
# 2. If required pipelines are not triggered within 6 rounds of loops, fail.
#     2.1 (special_care global variable could help on some pipelines that need to bypass the check)

# Check pipelines succeed or not:
# 1. These pipelines should return status within loop-times rounds.
# 2. If there is failed pipeline in the triggered pipelines, fail.

# Import necessary libraries
import os
import fnmatch
import subprocess
import time
import argparse
import json
import sys

# Define variables
github_repository = "microsoft/promptflow"
snippet_debug = os.getenv("SNIPPET_DEBUG", 0)
merge_commit = ""
loop_times = 40  # 40 * 30 seconds = 20 minutes
github_workspace = os.path.expanduser("~/promptflow/")

# Special cases for pipelines that need to be triggered more or less than default value 1.
# If 0, the pipeline will not be ignored in check enforcer.
# Please notice that the key should be the Job Name in the pipeline.
special_care = {
    "sdk_cli_tests": 4,
    "sdk_cli_azure_test_replay": 4,
    "sdk_cli_global_config_tests": 1,
    "tracing-e2e-test": 12,
    "tracing-unit-test": 12,
    "core_test": 4,
    "azureml_serving_test": 4,
    # "samples_connections_connection": 0,
}

# Copy from original yaml pipelines
checks = {
    "sdk_cli_tests": [
        "src/promptflow-core/**",
        "src/promptflow-devkit/**",
        "src/promptflow/**",
        "src/promptflow-tracing/**",
        "scripts/building/**",
        "src/promptflow-recording/**",
    ],
    "sdk_cli_global_config_tests": [
        "src/promptflow-core/**",
        "src/promptflow-devkit/**",
        "src/promptflow-tracing/**",
        "src/promptflow-azure/**",
        "src/promptflow/**",
        "scripts/building/**",
    ],
    "sdk_cli_azure_test_replay": [
        "src/promptflow/**",
        "scripts/building/**",
        "src/promptflow-tracing/**",
        "src/promptflow-core/**",
        "src/promptflow-devkit/**",
        "src/promptflow-azure/**",
    ],
    "tracing-e2e-test": [
        "src/promptflow-tracing/**",
    ],
    "tracing-unit-test": [
        "src/promptflow-tracing/**",
    ],
    "core_test": [
        "src/promptflow-tracing/**",
        "src/promptflow-core/**",
    ],
    "azureml_serving_test": [
        "src/promptflow-tracing/**",
        "src/promptflow-core/**",
    ],
}

reverse_checks = {}
pipelines = {}
pipelines_count = {}
failed_reason = ""


# Define functions
def trigger_checks(valid_status_array):
    global failed_reason
    global github_repository
    global merge_commit
    global snippet_debug
    global pipelines
    global pipelines_count

    output = subprocess.check_output(
        f"gh api /repos/{github_repository}/commits/{merge_commit}/check-suites?per_page=100",
        shell=True,
    )
    check_suites = json.loads(output)["check_suites"]
    for suite in check_suites:
        if snippet_debug != 0:
            print(f"check-suites id {suite['id']}")
        suite_id = suite["id"]
        output = subprocess.check_output(
            f"gh api /repos/{github_repository}/check-suites/{suite_id}/check-runs?per_page=100",
            shell=True,
        )
        check_runs = json.loads(output)["check_runs"]
        for run in check_runs:
            if snippet_debug != 0:
                print(f"check runs name {run['name']}")
            for key in pipelines.keys():
                value = pipelines[key]
                if value == 0:
                    continue
                if key in run["name"]:
                    pipelines_count[key] += 1
                    valid_status_array.append(run)

    for key in pipelines.keys():
        if pipelines_count[key] < pipelines[key]:
            print(
                f"[Failure]Pipeline {key} is triggered {pipelines_count[key]} times, less than {pipelines[key]} times."
            )
            failed_reason = "Not all pipelines are triggered."
        else:
            print(
                f"Pipeline {key} is triggered {pipelines_count[key]} times, more or equal to {pipelines[key]} times."
            )


def status_checks(valid_status_array):
    global failed_reason
    global pipelines
    global pipelines_count
    # Basic fact of sdk cli checked pipelines.
    failed_reason = ""

    # Loop through each valid status array.
    for status in valid_status_array:
        # Check if the pipeline was successful.
        if status["conclusion"] and status["conclusion"].lower() == "success":
            # Add 1 to the count of successful pipelines.
            pass
        # Check if the pipeline failed.
        elif status["conclusion"] and status["conclusion"].lower() == "failure":
            failed_reason = "Required pipelines are not successful."
        # Check if the pipeline is still running.
        else:
            if failed_reason == "":
                failed_reason = "Required pipelines are not finished."
        # Print the status of the pipeline to the console.
        print(status["name"] + " is checking.")


def trigger_prepare(input_paths):
    global github_workspace
    global checks
    global reverse_checks
    global pipelines
    global pipelines_count
    global failed_reason
    global special_care

    for input_path in input_paths:
        if "samples_connections_connection" in input_path:
            continue
        if input_path.startswith("migration-guide/"):
            continue
        if "-maf" in input_path:
            continue
        # Check if the input path contains "examples" or "samples".
        if "examples" in input_path or "samples" in input_path:
            sys.path.append(os.path.expanduser(github_workspace + "/scripts/readme"))
            from readme import main as readme_main

            os.chdir(os.path.expanduser(github_workspace))

            # Get the list of pipelines from the readme file.
            pipelines_samples = readme_main(check=True)

            git_diff_files = [
                item
                for item in subprocess.check_output(
                    ["git", "diff", "--name-only", "HEAD"]
                )
                .decode("utf-8")
                .split("\n")
                if item != ""
            ]
            for _ in git_diff_files:
                failed_reason = "Run readme generation before check in"
                return
            # Merge the pipelines from the readme file with the original list of pipelines.
            for key in pipelines_samples.keys():
                value = pipelines_samples[key]
                checks[key] = value

    # Reverse checks.
    for key in checks.keys():
        value = checks[key]
        for path in value:
            if path in reverse_checks:
                reverse_checks[path].append(key)
            else:
                reverse_checks[path] = [key]

    # Render pipelines and pipelines_count using input_paths.
    for input_path in input_paths:
        # Input pattern /**: input_path should match in the middle.
        # Input pattern /*: input_path should match last but one.
        # Other input pattern: input_path should match last.

        # Skip if input path is a markdown file.
        if input_path.endswith(".md"):
            continue

        keys = [
            key for key in reverse_checks.keys() if fnmatch.fnmatch(input_path, key)
        ]
        # Loop through each key in the list of keys.
        for key_item in keys:
            # Loop through each pipeline in the list of pipelines.
            for key in reverse_checks[key_item]:
                # Check if the pipeline is in the list of pipelines.
                if key in special_care:
                    pipelines[key] = special_care[key]
                else:
                    pipelines[key] = 1
                # Set the pipeline count to 0.
                pipelines_count[key] = 0


def run_checks():
    global github_repository
    global snippet_debug
    global merge_commit
    global loop_times
    global github_workspace
    global failed_reason

    if merge_commit == "":
        merge_commit = (
            subprocess.check_output(["git", "log", "-1"]).decode("utf-8").split("\n")
        )
        if snippet_debug != 0:
            print(merge_commit)
        for line in merge_commit:
            if "Merge" in line and "into" in line:
                merge_commit = line.split(" ")[-3]
                break
    if snippet_debug != 0:
        print("MergeCommit " + merge_commit)

    not_started_counter = 5

    os.chdir(github_workspace)
    # Get diff of current branch and main branch.
    try:
        git_merge_base = (
            subprocess.check_output(["git", "merge-base", "origin/main", "HEAD"])
            .decode("utf-8")
            .rstrip()
        )
        git_diff = (
            subprocess.check_output(
                ["git", "diff", "--name-only", "--diff-filter=d", f"{git_merge_base}"],
                stderr=subprocess.STDOUT,
            )
            .decode("utf-8")
            .rstrip()
            .split("\n")
        )
        for single_diff in git_diff:
            print(single_diff)
    except subprocess.CalledProcessError as e:
        print("Exception on process, rc=", e.returncode, "output=", e.output)
        raise e

    # Prepare how many pipelines should be triggered.
    trigger_prepare(git_diff)
    if failed_reason != "":
        raise Exception(failed_reason)

    # Loop for 20 minutes at most.
    for i in range(loop_times):
        # Wait for 30 seconds.
        time.sleep(30)

        # Reset the failed reason.
        failed_reason = ""
        # Reset the valid status array.
        valid_status_array = []

        # Get all triggered pipelines.
        # If not all pipelines are triggered, continue.
        trigger_checks(valid_status_array)
        if failed_reason != "":
            if not_started_counter == 0:
                raise Exception(failed_reason + " for 6 times.")
            print(failed_reason)
            not_started_counter -= 1
            continue

        # Get pipeline conclusion priority:
        # 1. Not successful, Fail.
        # 2. Not finished, Continue.
        # 3. Successful, Break.
        status_checks(valid_status_array)

        # Check if the failed reason contains "not successful".
        if "not successful" in failed_reason.lower():
            raise Exception(failed_reason)
        # Check if the failed reason contains "not finished".
        elif "not finished" in failed_reason.lower():
            print(failed_reason)
            continue
        # Otherwise, print that all required pipelines are successful.
        else:
            print("All required pipelines are successful.")
            break

    # Check if the failed reason is not empty.
    if failed_reason != "":
        raise Exception(failed_reason)


if __name__ == "__main__":
    # Run the checks.
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "-m",
        "--merge-commit",
        help="merge commit sha",
    )
    parser.add_argument(
        "-n",
        "--loop-times",
        type=int,
        help="Loop times",
    )
    parser.add_argument(
        "-t",
        "--github-workspace",
        help="base path of github workspace",
    )
    args = parser.parse_args()
    if args.merge_commit:
        merge_commit = args.merge_commit
    if args.loop_times:
        loop_times = args.loop_times
    if args.github_workspace:
        github_workspace = args.github_workspace
    run_checks()