helpers.py

from __future__ import annotations

import multiprocessing
import os
import random
import re
from typing import Any
from typing import overload
from typing import Sequence
from typing import TYPE_CHECKING

import pre_commit.constants as C
from pre_commit import parse_shebang
from pre_commit.hook import Hook
from pre_commit.prefix import Prefix
from pre_commit.util import cmd_output_b
from pre_commit.xargs import xargs

if TYPE_CHECKING:
    from typing import NoReturn

FIXED_RANDOM_SEED = 1542676187

SHIMS_RE = re.compile(r'[/\\]shims[/\\]')


def exe_exists(exe: str) -> bool:
    found = parse_shebang.find_executable(exe)
    if found is None:  # exe exists
        return False

    homedir = os.path.expanduser('~')
    try:
        common: str | None = os.path.commonpath((found, homedir))
    except ValueError:  # on windows, different drives raises ValueError
        common = None

    return (
        # it is not in a /shims/ directory
        not SHIMS_RE.search(found) and
        (
            # the homedir is / (docker, service user, etc.)
            os.path.dirname(homedir) == homedir or
            # the exe is not contained in the home directory
            common != homedir
        )
    )


def run_setup_cmd(prefix: Prefix, cmd: tuple[str, ...], **kwargs: Any) -> None:
    cmd_output_b(*cmd, cwd=prefix.prefix_dir, **kwargs)


@overload
def environment_dir(d: None, language_version: str) -> None: ...
@overload
def environment_dir(d: str, language_version: str) -> str: ...


def environment_dir(d: str | None, language_version: str) -> str | None:
    if d is None:
        return None
    else:
        return f'{d}-{language_version}'


def assert_version_default(binary: str, version: str) -> None:
    if version != C.DEFAULT:
        raise AssertionError(
            f'for now, pre-commit requires system-installed {binary} -- '
            f'you selected `language_version: {version}`',
        )


def assert_no_additional_deps(
        lang: str,
        additional_deps: Sequence[str],
) -> None:
    if additional_deps:
        raise AssertionError(
            f'for now, pre-commit does not support '
            f'additional_dependencies for {lang} -- '
            f'you selected `additional_dependencies: {additional_deps}`',
        )


def basic_get_default_version() -> str:
    return C.DEFAULT


def basic_healthy(prefix: Prefix, language_version: str) -> bool:
    return True


def no_install(
        prefix: Prefix,
        version: str,
        additional_dependencies: Sequence[str],
) -> NoReturn:
    raise AssertionError('This type is not installable')


def target_concurrency(hook: Hook) -> int:
    if hook.require_serial or 'PRE_COMMIT_NO_CONCURRENCY' in os.environ:
        return 1
    else:
        # Travis appears to have a bunch of CPUs, but we can't use them all.
        if 'TRAVIS' in os.environ:
            return 2
        else:
            try:
                return multiprocessing.cpu_count()
            except NotImplementedError:
                return 1


def _shuffled(seq: Sequence[str]) -> list[str]:
    """Deterministically shuffle"""
    fixed_random = random.Random()
    fixed_random.seed(FIXED_RANDOM_SEED, version=1)

    seq = list(seq)
    fixed_random.shuffle(seq)
    return seq


def run_xargs(
        hook: Hook,
        cmd: tuple[str, ...],
        file_args: Sequence[str],
        **kwargs: Any,
) -> tuple[int, bytes]:
    # Shuffle the files so that they more evenly fill out the xargs partitions,
    # but do it deterministically in case a hook cares about ordering.
    file_args = _shuffled(file_args)
    kwargs['target_concurrency'] = target_concurrency(hook)
    return xargs(cmd, file_args, **kwargs)