From d7784aa0c412c80cfcb33a29fd1b2ea69dbe8ab8 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Mon, 27 Feb 2023 11:32:22 +0100 Subject: Merging upstream version 3.1.0. Signed-off-by: Daniel Baumann --- pre_commit/lang_base.py | 196 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 196 insertions(+) create mode 100644 pre_commit/lang_base.py (limited to 'pre_commit/lang_base.py') diff --git a/pre_commit/lang_base.py b/pre_commit/lang_base.py new file mode 100644 index 0000000..9480c55 --- /dev/null +++ b/pre_commit/lang_base.py @@ -0,0 +1,196 @@ +from __future__ import annotations + +import contextlib +import multiprocessing +import os +import random +import re +import shlex +from typing import Any +from typing import ContextManager +from typing import Generator +from typing import NoReturn +from typing import Protocol +from typing import Sequence + +import pre_commit.constants as C +from pre_commit import parse_shebang +from pre_commit.prefix import Prefix +from pre_commit.util import cmd_output_b +from pre_commit.xargs import xargs + +FIXED_RANDOM_SEED = 1542676187 + +SHIMS_RE = re.compile(r'[/\\]shims[/\\]') + + +class Language(Protocol): + # Use `None` for no installation / environment + @property + def ENVIRONMENT_DIR(self) -> str | None: ... + # return a value to replace `'default` for `language_version` + def get_default_version(self) -> str: ... + # return whether the environment is healthy (or should be rebuilt) + def health_check(self, prefix: Prefix, version: str) -> str | None: ... + + # install a repository for the given language and language_version + def install_environment( + self, + prefix: Prefix, + version: str, + additional_dependencies: Sequence[str], + ) -> None: + ... + + # modify the environment for hook execution + def in_env(self, prefix: Prefix, version: str) -> ContextManager[None]: ... + + # execute a hook and return the exit code and output + def run_hook( + self, + prefix: Prefix, + entry: str, + args: Sequence[str], + file_args: Sequence[str], + *, + is_local: bool, + require_serial: bool, + color: bool, + ) -> tuple[int, bytes]: + ... + + +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 setup_cmd(prefix: Prefix, cmd: tuple[str, ...], **kwargs: Any) -> None: + cmd_output_b(*cmd, cwd=prefix.prefix_dir, **kwargs) + + +def environment_dir(prefix: Prefix, d: str, language_version: str) -> str: + return prefix.path(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_health_check(prefix: Prefix, language_version: str) -> str | None: + return None + + +def no_install( + prefix: Prefix, + version: str, + additional_dependencies: Sequence[str], +) -> NoReturn: + raise AssertionError('This language is not installable') + + +@contextlib.contextmanager +def no_env(prefix: Prefix, version: str) -> Generator[None, None, None]: + yield + + +def target_concurrency() -> int: + if '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( + cmd: tuple[str, ...], + file_args: Sequence[str], + *, + require_serial: bool, + color: bool, +) -> tuple[int, bytes]: + if require_serial: + jobs = 1 + else: + # 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) + jobs = target_concurrency() + return xargs(cmd, file_args, target_concurrency=jobs, color=color) + + +def hook_cmd(entry: str, args: Sequence[str]) -> tuple[str, ...]: + return (*shlex.split(entry), *args) + + +def basic_run_hook( + prefix: Prefix, + entry: str, + args: Sequence[str], + file_args: Sequence[str], + *, + is_local: bool, + require_serial: bool, + color: bool, +) -> tuple[int, bytes]: + return run_xargs( + hook_cmd(entry, args), + file_args, + require_serial=require_serial, + color=color, + ) -- cgit v1.2.3