lib.py

import asyncio
import errno
import json
import logging
import os
import stat
import sys
from functools import partial
from pathlib import Path
from platform import system
from shutil import rmtree, which
from subprocess import CalledProcessError
from sys import version_info
from tempfile import TemporaryDirectory
from typing import (
    Any,
    Callable,
    Dict,
    List,
    NamedTuple,
    Optional,
    Sequence,
    Tuple,
    Union,
)
from urllib.parse import urlparse

import click


TEN_MINUTES_SECONDS = 600
WINDOWS = system() == "Windows"
BLACK_BINARY = "black.exe" if WINDOWS else "black"
GIT_BINARY = "git.exe" if WINDOWS else "git"
LOG = logging.getLogger(__name__)


# Windows needs a ProactorEventLoop if you want to exec subprocesses
# Starting with 3.8 this is the default - can remove when Black >= 3.8
# mypy only respects sys.platform if directly in the evaluation
# https://mypy.readthedocs.io/en/latest/common_issues.html#python-version-and-system-platform-checks  # noqa: B950
if sys.platform == "win32":
    asyncio.set_event_loop(asyncio.ProactorEventLoop())


class Results(NamedTuple):
    stats: Dict[str, int] = {}
    failed_projects: Dict[str, CalledProcessError] = {}


async def _gen_check_output(
    cmd: Sequence[str],
    timeout: float = TEN_MINUTES_SECONDS,
    env: Optional[Dict[str, str]] = None,
    cwd: Optional[Path] = None,
    stdin: Optional[bytes] = None,
) -> Tuple[bytes, bytes]:
    process = await asyncio.create_subprocess_exec(
        *cmd,
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT,
        env=env,
        cwd=cwd,
    )
    try:
        (stdout, stderr) = await asyncio.wait_for(process.communicate(stdin), timeout)
    except asyncio.TimeoutError:
        process.kill()
        await process.wait()
        raise

    # A non-optional timeout was supplied to asyncio.wait_for, guaranteeing
    # a timeout or completed process.  A terminated Python process will have a
    # non-empty returncode value.
    assert process.returncode is not None

    if process.returncode != 0:
        cmd_str = " ".join(cmd)
        raise CalledProcessError(
            process.returncode, cmd_str, output=stdout, stderr=stderr
        )

    return (stdout, stderr)


def analyze_results(project_count: int, results: Results) -> int:
    failed_pct = round(((results.stats["failed"] / project_count) * 100), 2)
    success_pct = round(((results.stats["success"] / project_count) * 100), 2)

    if results.failed_projects:
        click.secho("\nFailed projects:\n", bold=True)

    for project_name, project_cpe in results.failed_projects.items():
        print(f"## {project_name}:")
        print(f" - Returned {project_cpe.returncode}")
        if project_cpe.stderr:
            print(f" - stderr:\n{project_cpe.stderr.decode('utf8')}")
        if project_cpe.stdout:
            print(f" - stdout:\n{project_cpe.stdout.decode('utf8')}")
        print("")

    click.secho("-- primer results 📊 --\n", bold=True)
    click.secho(
        f"{results.stats['success']} / {project_count} succeeded ({success_pct}%) ✅",
        bold=True,
        fg="green",
    )
    click.secho(
        f"{results.stats['failed']} / {project_count} FAILED ({failed_pct}%) 💩",
        bold=bool(results.stats["failed"]),
        fg="red",
    )
    s = "" if results.stats["disabled"] == 1 else "s"
    click.echo(f" - {results.stats['disabled']} project{s} disabled by config")
    s = "" if results.stats["wrong_py_ver"] == 1 else "s"
    click.echo(
        f" - {results.stats['wrong_py_ver']} project{s} skipped due to Python version"
    )
    click.echo(
        f" - {results.stats['skipped_long_checkout']} skipped due to long checkout"
    )

    if results.failed_projects:
        failed = ", ".join(results.failed_projects.keys())
        click.secho(f"\nFailed projects: {failed}\n", bold=True)

    return results.stats["failed"]


def _flatten_cli_args(cli_args: List[Union[Sequence[str], str]]) -> List[str]:
    """Allow a user to put long arguments into a list of strs
    to make the JSON human readable"""
    flat_args = []
    for arg in cli_args:
        if isinstance(arg, str):
            flat_args.append(arg)
            continue

        args_as_str = "".join(arg)
        flat_args.append(args_as_str)

    return flat_args


async def black_run(
    project_name: str,
    repo_path: Optional[Path],
    project_config: Dict[str, Any],
    results: Results,
    no_diff: bool = False,
) -> None:
    """Run Black and record failures"""
    if not repo_path:
        results.stats["failed"] += 1
        results.failed_projects[project_name] = CalledProcessError(
            69, [], f"{project_name} has no repo_path: {repo_path}".encode(), b""
        )
        return

    stdin_test = project_name.upper() == "STDIN"
    cmd = [str(which(BLACK_BINARY))]
    if "cli_arguments" in project_config and project_config["cli_arguments"]:
        cmd.extend(_flatten_cli_args(project_config["cli_arguments"]))
    cmd.append("--check")
    if not no_diff:
        cmd.append("--diff")

    # Workout if we should read in a python file or search from cwd
    stdin = None
    if stdin_test:
        cmd.append("-")
        stdin = repo_path.read_bytes()
    elif "base_path" in project_config:
        cmd.append(project_config["base_path"])
    else:
        cmd.append(".")

    timeout = (
        project_config["timeout_seconds"]
        if "timeout_seconds" in project_config
        else TEN_MINUTES_SECONDS
    )
    with TemporaryDirectory() as tmp_path:
        # Prevent reading top-level user configs by manipulating environment variables
        env = {
            **os.environ,
            "XDG_CONFIG_HOME": tmp_path,  # Unix-like
            "USERPROFILE": tmp_path,  # Windows (changes `Path.home()` output)
        }

        cwd_path = repo_path.parent if stdin_test else repo_path
        try:
            LOG.debug(f"Running black for {project_name}: {' '.join(cmd)}")
            _stdout, _stderr = await _gen_check_output(
                cmd, cwd=cwd_path, env=env, stdin=stdin, timeout=timeout
            )
        except asyncio.TimeoutError:
            results.stats["failed"] += 1
            LOG.error(f"Running black for {repo_path} timed out ({cmd})")
        except CalledProcessError as cpe:
            # TODO: Tune for smarter for higher signal
            # If any other return value than 1 we raise - can disable project in config
            if cpe.returncode == 1:
                if not project_config["expect_formatting_changes"]:
                    results.stats["failed"] += 1
                    results.failed_projects[repo_path.name] = cpe
                else:
                    results.stats["success"] += 1
                return
            elif cpe.returncode > 1:
                results.stats["failed"] += 1
                results.failed_projects[repo_path.name] = cpe
                return

            LOG.error(f"Unknown error with {repo_path}")
            raise

    # If we get here and expect formatting changes something is up
    if project_config["expect_formatting_changes"]:
        results.stats["failed"] += 1
        results.failed_projects[repo_path.name] = CalledProcessError(
            0, cmd, b"Expected formatting changes but didn't get any!", b""
        )
        return

    results.stats["success"] += 1


async def git_checkout_or_rebase(
    work_path: Path,
    project_config: Dict[str, Any],
    rebase: bool = False,
    *,
    depth: int = 1,
) -> Optional[Path]:
    """git Clone project or rebase"""
    git_bin = str(which(GIT_BINARY))
    if not git_bin:
        LOG.error("No git binary found")
        return None

    repo_url_parts = urlparse(project_config["git_clone_url"])
    path_parts = repo_url_parts.path[1:].split("/", maxsplit=1)

    repo_path: Path = work_path / path_parts[1].replace(".git", "")
    cmd = [git_bin, "clone", "--depth", str(depth), project_config["git_clone_url"]]
    cwd = work_path
    if repo_path.exists() and rebase:
        cmd = [git_bin, "pull", "--rebase"]
        cwd = repo_path
    elif repo_path.exists():
        return repo_path

    try:
        _stdout, _stderr = await _gen_check_output(cmd, cwd=cwd)
    except (asyncio.TimeoutError, CalledProcessError) as e:
        LOG.error(f"Unable to git clone / pull {project_config['git_clone_url']}: {e}")
        return None

    return repo_path


def handle_PermissionError(
    func: Callable[..., None], path: Path, exc: Tuple[Any, Any, Any]
) -> None:
    """
    Handle PermissionError during shutil.rmtree.

    This checks if the erroring function is either 'os.rmdir' or 'os.unlink', and that
    the error was EACCES (i.e. Permission denied). If true, the path is set writable,
    readable, and executable by everyone. Finally, it tries the error causing delete
    operation again.

    If the check is false, then the original error will be reraised as this function
    can't handle it.
    """
    excvalue = exc[1]
    LOG.debug(f"Handling {excvalue} from {func.__name__}... ")
    if func in (os.rmdir, os.unlink) and excvalue.errno == errno.EACCES:
        LOG.debug(f"Setting {path} writable, readable, and executable by everyone... ")
        os.chmod(path, stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO)  # chmod 0777
        func(path)  # Try the error causing delete operation again
    else:
        raise


async def load_projects_queue(
    config_path: Path,
    projects_to_run: List[str],
) -> Tuple[Dict[str, Any], asyncio.Queue]:
    """Load project config and fill queue with all the project names"""
    with config_path.open("r") as cfp:
        config = json.load(cfp)

    # TODO: Offer more options here
    # e.g. Run on X random packages etc.
    queue: asyncio.Queue = asyncio.Queue(maxsize=len(projects_to_run))
    for project in projects_to_run:
        await queue.put(project)

    return config, queue


async def project_runner(
    idx: int,
    config: Dict[str, Any],
    queue: asyncio.Queue,
    work_path: Path,
    results: Results,
    long_checkouts: bool = False,
    rebase: bool = False,
    keep: bool = False,
    no_diff: bool = False,
) -> None:
    """Check out project and run Black on it + record result"""
    loop = asyncio.get_event_loop()
    py_version = f"{version_info[0]}.{version_info[1]}"
    while True:
        try:
            project_name = queue.get_nowait()
        except asyncio.QueueEmpty:
            LOG.debug(f"project_runner {idx} exiting")
            return
        LOG.debug(f"worker {idx} working on {project_name}")

        project_config = config["projects"][project_name]

        # Check if disabled by config
        if "disabled" in project_config and project_config["disabled"]:
            results.stats["disabled"] += 1
            LOG.info(f"Skipping {project_name} as it's disabled via config")
            continue

        # Check if we should run on this version of Python
        if (
            "all" not in project_config["py_versions"]
            and py_version not in project_config["py_versions"]
        ):
            results.stats["wrong_py_ver"] += 1
            LOG.debug(f"Skipping {project_name} as it's not enabled for {py_version}")
            continue

        # Check if we're doing big projects / long checkouts
        if not long_checkouts and project_config["long_checkout"]:
            results.stats["skipped_long_checkout"] += 1
            LOG.debug(f"Skipping {project_name} as it's configured as a long checkout")
            continue

        repo_path: Optional[Path] = Path(__file__)
        stdin_project = project_name.upper() == "STDIN"
        if not stdin_project:
            repo_path = await git_checkout_or_rebase(work_path, project_config, rebase)
            if not repo_path:
                continue
        await black_run(project_name, repo_path, project_config, results, no_diff)

        if not keep and not stdin_project:
            LOG.debug(f"Removing {repo_path}")
            rmtree_partial = partial(
                rmtree, path=repo_path, onerror=handle_PermissionError
            )
            await loop.run_in_executor(None, rmtree_partial)

        LOG.info(f"Finished {project_name}")


async def process_queue(
    config_file: str,
    work_path: Path,
    workers: int,
    projects_to_run: List[str],
    keep: bool = False,
    long_checkouts: bool = False,
    rebase: bool = False,
    no_diff: bool = False,
) -> int:
    """
    Process the queue with X workers and evaluate results
    - Success is guaged via the config "expect_formatting_changes"

    Integer return equals the number of failed projects
    """
    results = Results()
    results.stats["disabled"] = 0
    results.stats["failed"] = 0
    results.stats["skipped_long_checkout"] = 0
    results.stats["success"] = 0
    results.stats["wrong_py_ver"] = 0

    config, queue = await load_projects_queue(Path(config_file), projects_to_run)
    project_count = queue.qsize()
    s = "" if project_count == 1 else "s"
    LOG.info(f"{project_count} project{s} to run Black over")
    if project_count < 1:
        return -1

    s = "" if workers == 1 else "s"
    LOG.debug(f"Using {workers} parallel worker{s} to run Black")
    # Wait until we finish running all the projects before analyzing
    await asyncio.gather(
        *[
            project_runner(
                i,
                config,
                queue,
                work_path,
                results,
                long_checkouts,
                rebase,
                keep,
                no_diff,
            )
            for i in range(workers)
        ]
    )

    LOG.info("Analyzing results")
    return analyze_results(project_count, results)


if __name__ == "__main__":  # pragma: nocover
    raise NotImplementedError("lib is a library, funnily enough.")