check-merge-conflicts.py

"""Checks which pull requests contain merge conflicts with Blackened code."""
import os
import shlex
import subprocess
import tempfile

import github
import tqdm


OUTPUT_PATH = os.path.join(os.getcwd(), "no-conflict.numbers")
LOG_PATH = os.path.join(os.getcwd(), "no-conflict.log")

GITHUB_ACCESS_TOKEN = ...


def main() -> None:
    """Main function."""
    with tempfile.TemporaryDirectory() as temporary_directory:
        # Clone Ray and change working directory to repository.
        print("Cloning Ray")
        os.chdir(temporary_directory)
        run("git clone https://github.com/ray-project/ray")
        os.chdir(os.path.join(temporary_directory, "ray"))

        # Blacken Python code on master.
        print("Formatting Python code")
        assert current_branch_name() == "master"
        run("black .")
        run("git add --all")
        run('git commit -m "Format Python code with Black"')

        # Check which pull requests contain merge conflicts with master.
        print("Checking for merge conflicts")
        github_client = github.Github(GITHUB_ACCESS_TOKEN)
        ray_repo = github_client.get_repo("ray-project/ray")
        pull_requests = ray_repo.get_pulls(state="open")
        for pull_request in tqdm.tqdm(pull_requests, total=pull_requests.totalCount):
            if not will_conflict(pull_request, "master"):
                with open(OUTPUT_PATH, "a") as file:
                    file.write(str(pull_request.number) + "\n")


class checkout:
    """Creates and checks out a temporary branch based on a pull request.

    When the context manager exits, the context manager checks out the
    previously current branch and deletes the temporary branch.
    """

    def __init__(self, pull_request: github.PullRequest):
        """Initializes instance attributes.

        Arguments:
            pull_request: The pull request to base the branch on.
        """
        self._pull_request_id = pull_request.number
        self._temporary_branch_name = f"temp-{pull_request.number}"
        self._previous_branch_name = current_branch_name()

    def __enter__(self):
        """Creates and checks out a new branch based on the provided pull request."""
        # See https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/reviewing-changes-in-pull-requests/checking-out-pull-requests-locally.
        run(
            f"git fetch origin pull/{self._pull_request_id}/head:{self._temporary_branch_name}"
        )
        run(f"git checkout {self._temporary_branch_name}")
        return self

    def __exit__(self, type, value, traceback):
        """Checks out the previous branch and deletes the temporary branch."""
        run(f"git checkout {self._previous_branch_name}")
        run(f"git branch -D {self._temporary_branch_name}")


def will_conflict(pull_request: github.PullRequest, branch_name: str) -> bool:
    """Checks if a pull request contains merge conflicts with a local branch.

    Arguments:
        pull_request: The pull request to check for merge conflicts.
        branch_name: The name of the local branch to check against.

    Returns:
        True if the pull request contains merge conflicts with the specified branch and false otherwise.
    """
    with checkout(pull_request):
        # See https://stackoverflow.com/questions/10879331/how-to-check-the-conflict-of-two-branch-but-not-need-to-merge-them/10879368.
        return_code = run(f"git merge {branch_name} --no-ff --no-commit")
        assert return_code in {0, 1}
        conflict_exists = return_code == 1
        run("git merge --abort")
        return conflict_exists


def current_branch_name() -> str:
    """Returns the name of the current git branch."""
    process = subprocess.Popen(
        ["git", "branch", "--show-current"], stdout=subprocess.PIPE
    )
    stdout, _ = process.communicate()
    return stdout.decode().strip()


def run(command: str) -> int:
    """Runs a command, logs the output, and returns the return code."""
    with open(LOG_PATH, "a") as file:
        process = subprocess.run(
            shlex.split(command, posix=False), stderr=file, stdout=file
        )
    return process.returncode


if __name__ == "__main__":
    main()

requirements.txt

black
PyGithub
tqdm