NO_FS_MONITOR = ('-c', 'core.useBuiltinFSMonitor=false')
def zsplit(s: str) -> list[str]:
def no_git_env(
        _env: MutableMapping[str, str] | None = None,
) -> dict[str, str]:
def get_root() -> str:
def get_git_dir(git_root: str = '.') -> str:
def get_remote_url(git_root: str) -> str:
def is_in_merge_conflict() -> bool:
def parse_merge_msg_for_conflicts(merge_msg: bytes) -> list[str]:
def get_conflicted_files() -> set[str]:
def get_staged_files(cwd: str | None = None) -> list[str]:
def intent_to_add_files() -> list[str]:
def get_all_files() -> list[str]:
def get_changed_files(old: str, new: str) -> list[str]:
def head_rev(remote: str) -> str:
def has_diff(*args: str, repo: str = '.') -> bool:
def has_core_hookpaths_set() -> bool:
def init_repo(path: str, remote: str) -> None:
def commit(repo: str = '.') -> None:
def git_path(name: str, repo: str = '.') -> str:
def check_for_cygwin_mismatch() -> None: CONFIG_SCHEMA = cfgv.Map(
    'Config', None,

    cfgv.RequiredRecurse('repos', cfgv.Array(CONFIG_REPO_DICT)),
    cfgv.OptionalRecurse(
        'default_language_version', DEFAULT_LANGUAGE_VERSION, {},
    ),
    cfgv.Optional(
        'default_stages',
        cfgv.check_array(cfgv.check_one_of(C.STAGES)),
        C.STAGES,
    ),
    cfgv.Optional('files', check_string_regex, ''),
    cfgv.Optional('exclude', check_string_regex, '^$'),
    cfgv.Optional('fail_fast', cfgv.check_bool, False),
    cfgv.Optional(
        'minimum_pre_commit_version',
        cfgv.check_and(cfgv.check_string, check_min_version),
        '0',
    ),
    cfgv.WarnAdditionalKeys(
        (
            'repos',
            'default_language_version',
            'default_stages',
            'files',
            'exclude',
            'fail_fast',
            'minimum_pre_commit_version',
            'ci',
        ),
        warn_unknown_keys_root,
    ),
    OptionalSensibleRegexAtTop('files', cfgv.check_string),
    OptionalSensibleRegexAtTop('exclude', cfgv.check_string),

    # do not warn about configuration for pre-commit.ci
    cfgv.OptionalNoDefault('ci', cfgv.check_type(dict)),
) def check_type_tag(tag: str) -> None:
def check_min_version(version: str) -> None:
def _make_argparser(filenames_help: str) -> argparse.ArgumentParser:
def validate_manifest_main(argv: Sequence[str] | None = None) -> int:
    def check(self, dct: dict[str, Any]) -> None:
    def check(self, dct: dict[str, Any]) -> None:
    def check(self, dct: dict[str, Any]) -> None:
    def _cond(key: str) -> cfgv.Conditional:
    def check(self, dct: dict[str, Any]) -> None:
    def apply_default(self, dct: dict[str, Any]) -> None:
def _entry(modname: str) -> str:
def warn_unknown_keys_root(
        extra: Sequence[str],
        orig_keys: Sequence[str],
        dct: dict[str, str],
) -> None:
def warn_unknown_keys_repo(
        extra: Sequence[str],
        orig_keys: Sequence[str],
        dct: dict[str, str],
) -> None:
    def check(self, dct: dict[str, Any]) -> None:
def ordered_load_normalize_legacy_config(contents: str) -> dict[str, Any]:
def validate_config_main(argv: Sequence[str] | None = None) -> int:
MANIFEST_HOOK_DICT = cfgv.Map(
    'Hook', 'id',

    cfgv.Required('id', cfgv.check_string),
    cfgv.Required('name', cfgv.check_string),
    cfgv.Required('entry', cfgv.check_string),
    cfgv.Required('language', cfgv.check_one_of(all_languages)),
    cfgv.Optional('alias', cfgv.check_string, ''),

    cfgv.Optional('files', check_string_regex, ''),
    cfgv.Optional('exclude', check_string_regex, '^$'),
    cfgv.Optional('types', cfgv.check_array(check_type_tag), ['file']),
    cfgv.Optional('types_or', cfgv.check_array(check_type_tag), []),
    cfgv.Optional('exclude_types', cfgv.check_array(check_type_tag), []),

    cfgv.Optional(
        'additional_dependencies', cfgv.check_array(cfgv.check_string), [],
    ),
    cfgv.Optional('args', cfgv.check_array(cfgv.check_string), []),
    cfgv.Optional('always_run', cfgv.check_bool, False),
    cfgv.Optional('fail_fast', cfgv.check_bool, False),
    cfgv.Optional('pass_filenames', cfgv.check_bool, True),
    cfgv.Optional('description', cfgv.check_string, ''),
    cfgv.Optional('language_version', cfgv.check_string, C.DEFAULT),
    cfgv.Optional('log_file', cfgv.check_string, ''),
    cfgv.Optional('minimum_pre_commit_version', cfgv.check_string, '0'),
    cfgv.Optional('require_serial', cfgv.check_bool, False),
    cfgv.Optional('stages', cfgv.check_array(cfgv.check_one_of(C.STAGES)), []),
    cfgv.Optional('verbose', cfgv.check_bool, False),
)
MANIFEST_SCHEMA = cfgv.Array(MANIFEST_HOOK_DICT)
LOCAL = 'local'
META = 'meta'
META_HOOK_DICT = cfgv.Map(
    'Hook', 'id',
    cfgv.Required('id', cfgv.check_string),
    cfgv.Required('id', cfgv.check_one_of(tuple(k for k, _ in _meta))),
    # language must be system
    cfgv.Optional('language', cfgv.check_one_of({'system'}), 'system'),
    # entry cannot be overridden
    NotAllowed('entry', cfgv.check_any),
    *(
        # default to the hook definition for the meta hooks
        cfgv.ConditionalOptional(key, cfgv.check_any, value, 'id', hook_id)
        for hook_id, values in _meta
        for key, value in values
    ),
    *(
        # default to the "manifest" parsing
        cfgv.OptionalNoDefault(item.key, item.check_fn)
        # these will always be defaulted above
        if item.key in {'name', 'language', 'entry'} else
        item
        for item in MANIFEST_HOOK_DICT.items
    ),
)
CONFIG_HOOK_DICT = cfgv.Map(
    'Hook', 'id',

    cfgv.Required('id', cfgv.check_string),

    # All keys in manifest hook dict are valid in a config hook dict, but
    # are optional.
    # No defaults are provided here as the config is merged on top of the
    # manifest.
    *(
        cfgv.OptionalNoDefault(item.key, item.check_fn)
        for item in MANIFEST_HOOK_DICT.items
        if item.key != 'id'
    ),
    OptionalSensibleRegexAtHook('files', cfgv.check_string),
    OptionalSensibleRegexAtHook('exclude', cfgv.check_string),
)
CONFIG_REPO_DICT = cfgv.Map(
    'Repository', 'repo',

    cfgv.Required('repo', cfgv.check_string),

    cfgv.ConditionalRecurse(
        'hooks', cfgv.Array(CONFIG_HOOK_DICT),
        'repo', cfgv.NotIn(LOCAL, META),
    ),
    cfgv.ConditionalRecurse(
        'hooks', cfgv.Array(MANIFEST_HOOK_DICT),
        'repo', LOCAL,
    ),
    cfgv.ConditionalRecurse(
        'hooks', cfgv.Array(META_HOOK_DICT),
        'repo', META,
    ),

    MigrateShaToRev(),
    WarnMutableRev(
        'rev',
        cfgv.check_string,
        '',
        'repo',
        cfgv.NotIn(LOCAL, META),
        True,
    ),
    cfgv.WarnAdditionalKeys(('repo', 'rev', 'hooks'), warn_unknown_keys_repo),
)
DEFAULT_LANGUAGE_VERSION = cfgv.Map(
    'DefaultLanguageVersion', None,
    cfgv.NoAdditionalKeys(all_languages),
    *(cfgv.Optional(x, cfgv.check_string, C.DEFAULT) for x in all_languages),
)
CONFIG_SCHEMA = cfgv.Map(
    'Config', None,

    cfgv.RequiredRecurse('repos', cfgv.Array(CONFIG_REPO_DICT)),
    cfgv.OptionalRecurse(
        'default_language_version', DEFAULT_LANGUAGE_VERSION, {},
    ),
    cfgv.Optional(
        'default_stages',
        cfgv.check_array(cfgv.check_one_of(C.STAGES)),
        C.STAGES,
    ),
    cfgv.Optional('files', check_string_regex, ''),
    cfgv.Optional('exclude', check_string_regex, '^$'),
    cfgv.Optional('fail_fast', cfgv.check_bool, False),
    cfgv.Optional(
        'minimum_pre_commit_version',
        cfgv.check_and(cfgv.check_string, check_min_version),
        '0',
    ),
    cfgv.WarnAdditionalKeys(
        (
            'repos',
            'default_language_version',
            'default_stages',
            'files',
            'exclude',
            'fail_fast',
            'minimum_pre_commit_version',
            'ci',
        ),
        warn_unknown_keys_root,
    ),
    OptionalSensibleRegexAtTop('files', cfgv.check_string),
    OptionalSensibleRegexAtTop('exclude', cfgv.check_string),

    # do not warn about configuration for pre-commit.ci
    cfgv.OptionalNoDefault('ci', cfgv.check_type(dict)),
)
class InvalidManifestError(FatalError):
class WarnMutableRev(cfgv.ConditionalOptional):
class OptionalSensibleRegexAtHook(cfgv.OptionalNoDefault):
class OptionalSensibleRegexAtTop(cfgv.OptionalNoDefault):
class MigrateShaToRev:
class NotAllowed(cfgv.OptionalNoDefault):
class InvalidConfigError(FatalError): @contextlib.contextmanager
def envcontext(
        patch: PatchesT,
        _env: MutableMapping[str, str] | None = None,
) -> Generator[None, None, None]:
    """In this context, `os.environ` is modified according to `patch`.

    `patch` is an iterable of 2-tuples (key, value):
        `key`: string
        `value`:
            - string: `environ[key] == value` inside the context.
            - UNSET: `key not in environ` inside the context.
            - template: A template is a tuple of strings and Var which will be
              replaced with the previous environment
    """
    env = os.environ if _env is None else _env
    before = dict(env)

    for k, v in patch:
        if v is UNSET:
            env.pop(k, None)
        elif isinstance(v, tuple):
            env[k] = format_env(v, before)
        else:
            env[k] = v

    try:
        yield
    finally:
        env.clear()
        env.update(before) class Hook(NamedTuple):
    src: str
    prefix: Prefix
    id: str
    name: str
    entry: str
    language: str
    alias: str
    files: str
    exclude: str
    types: Sequence[str]
    types_or: Sequence[str]
    exclude_types: Sequence[str]
    additional_dependencies: Sequence[str]
    args: Sequence[str]
    always_run: bool
    fail_fast: bool
    pass_filenames: bool
    description: str
    language_version: str
    log_file: str
    minimum_pre_commit_version: str
    require_serial: bool
    stages: Sequence[str]
    verbose: bool

    @property
    def cmd(self) -> tuple[str, ...]:
        return (*shlex.split(self.entry), *self.args)

    @property
    def install_key(self) -> tuple[Prefix, str, str, tuple[str, ...]]:
        return (
            self.prefix,
            self.language,
            self.language_version,
            tuple(self.additional_dependencies),
        )

    @classmethod
    def create(cls, src: str, prefix: Prefix, dct: dict[str, Any]) -> Hook:
        # TODO: have cfgv do this (?)
        extra_keys = set(dct) - _KEYS
        if extra_keys:
            logger.warning(
                f'Unexpected key(s) present on {src} => {dct["id"]}: '
                f'{", ".join(sorted(extra_keys))}',
            )
        return cls(src=src, prefix=prefix, **{k: dct[k] for k in _KEYS}) ENVIRONMENT_DIR = 'golangenv'
def get_env_patch(venv: str) -> PatchesT:
def in_env(prefix: Prefix) -> Generator[None, None, None]:
def guess_go_dir(remote_url: str) -> str:
def install_environment(
        prefix: Prefix,
        version: str,
        additional_dependencies: Sequence[str],
) -> None:
def run_hook(
        hook: Hook,
        file_args: Sequence[str],
        color: bool,
) -> tuple[int, bytes]: FIXED_RANDOM_SEED = 1542676187
SHIMS_RE = re.compile(r'[/\\]shims[/\\]')
def exe_exists(exe: str) -> bool:
def run_setup_cmd(prefix: Prefix, cmd: tuple[str, ...], **kwargs: Any) -> None:
def environment_dir(d: None, language_version: str) -> None: ...
def environment_dir(d: str, language_version: str) -> str: ...
def environment_dir(d: str | None, language_version: str) -> str | None:
def assert_version_default(binary: str, version: str) -> None:
def assert_no_additional_deps(
        lang: str,
        additional_deps: Sequence[str],
) -> None:
def basic_get_default_version() -> str:
def basic_healthy(prefix: Prefix, language_version: str) -> bool:
def no_install(
        prefix: Prefix,
        version: str,
        additional_dependencies: Sequence[str],
) -> NoReturn:
def target_concurrency(hook: Hook) -> int:
def _shuffled(seq: Sequence[str]) -> list[str]:
def run_xargs(
        hook: Hook,
        cmd: tuple[str, ...],
        file_args: Sequence[str],
        **kwargs: Any,
) -> tuple[int, bytes]: ENVIRONMENT_DIR = 'node_env'
def get_default_version() -> str:
def _envdir(prefix: Prefix, version: str) -> str:
def get_env_patch(venv: str) -> PatchesT:
def in_env(
        prefix: Prefix,
        language_version: str,
) -> Generator[None, None, None]:
def healthy(prefix: Prefix, language_version: str) -> bool:
def install_environment(
        prefix: Prefix, version: str, additional_dependencies: Sequence[str],
) -> None:
def run_hook(
        hook: Hook,
        file_args: Sequence[str],
        color: bool,
) -> tuple[int, bytes]: ENVIRONMENT_DIR = 'py_env'
def _version_info(exe: str) -> str:
def _read_pyvenv_cfg(filename: str) -> dict[str, str]:
def bin_dir(venv: str) -> str:
def get_env_patch(venv: str) -> PatchesT:
def _find_by_py_launcher(
        version: str,
) -> str | None:  # pragma: no cover (windows only)
def _find_by_sys_executable() -> str | None:
    def _norm(path: str) -> str | None:
def get_default_version() -> str:  # pragma: no cover (platform dependent)
def _sys_executable_matches(version: str) -> bool:
def norm_version(version: str) -> str | None:
def in_env(
        prefix: Prefix,
        language_version: str,
) -> Generator[None, None, None]:
def healthy(prefix: Prefix, language_version: str) -> bool:
def install_environment(
        prefix: Prefix,
        version: str,
        additional_dependencies: Sequence[str],
) -> None:
def run_hook(
        hook: Hook,
        file_args: Sequence[str],
        color: bool,
) -> tuple[int, bytes]: ENVIRONMENT_DIR = 'rbenv'
def get_default_version() -> str:
def get_env_patch(
        venv: str,
        language_version: str,
) -> PatchesT:
def in_env(
        prefix: Prefix,
        language_version: str,
) -> Generator[None, None, None]:
def _extract_resource(filename: str, dest: str) -> None:
def _install_rbenv(
        prefix: Prefix,
        version: str,
) -> None:  # pragma: win32 no cover
def _install_ruby(
        prefix: Prefix,
        version: str,
) -> None:  # pragma: win32 no cover
def install_environment(
        prefix: Prefix, version: str, additional_dependencies: Sequence[str],
) -> None:
def run_hook(
        hook: Hook,
        file_args: Sequence[str],
        color: bool,
) -> tuple[int, bytes]: ENVIRONMENT_DIR = 'rustenv'
def get_env_patch(target_dir: str) -> PatchesT:
def in_env(prefix: Prefix) -> Generator[None, None, None]:
def _add_dependencies(
        cargo_toml_path: str,
        additional_dependencies: set[str],
) -> None:
def install_environment(
        prefix: Prefix,
        version: str,
        additional_dependencies: Sequence[str],
) -> None:
def run_hook(
        hook: Hook,
        file_args: Sequence[str],
        color: bool,
) -> tuple[int, bytes]: class Language(NamedTuple):
    ENVIRONMENT_DIR: str | None class Prefix(NamedTuple):
    prefix_dir: str

    def path(self, *parts: str) -> str:
        return os.path.normpath(os.path.join(self.prefix_dir, *parts))

    def exists(self, *parts: str) -> bool:
        return os.path.exists(self.path(*parts))

    def star(self, end: str) -> tuple[str, ...]:
        paths = os.listdir(self.prefix_dir)
        return tuple(path for path in paths if path.endswith(end)) def all_hooks(root_config: dict[str, Any], store: Store) -> tuple[Hook, ...]:
    return tuple(
        hook
        for repo in root_config['repos']
        for hook in _repository_hooks(repo, store, root_config)
    ) def install_hook_envs(hooks: Sequence[Hook], store: Store) -> None:
    def _need_installed() -> list[Hook]:
        seen: set[tuple[Prefix, str, str, tuple[str, ...]]] = set()
        ret = []
        for hook in hooks:
            if hook.install_key not in seen and not _hook_installed(hook):
                ret.append(hook)
            seen.add(hook.install_key)
        return ret

    if not _need_installed():
        return
    with store.exclusive_lock():
        # Another process may have already completed this work
        for hook in _need_installed():
            _hook_install(hook) def cmd_output(*cmd: str, **kwargs: Any) -> tuple[int, str, str | None]:
    returncode, stdout_b, stderr_b = cmd_output_b(*cmd, **kwargs)
    stdout = stdout_b.decode() if stdout_b is not None else None
    stderr = stderr_b.decode() if stderr_b is not None else None
    return returncode, stdout, stderr def cmd_output_b(
        *cmd: str,
        retcode: int | None = 0,
        **kwargs: Any,
) -> tuple[int, bytes, bytes | None]:
    _setdefault_kwargs(kwargs)

    try:
        cmd = parse_shebang.normalize_cmd(cmd)
    except parse_shebang.ExecutableNotFoundError as e:
        returncode, stdout_b, stderr_b = e.to_output()
    else:
        try:
            proc = subprocess.Popen(cmd, **kwargs)
        except OSError as e:
            returncode, stdout_b, stderr_b = _oserror_to_output(e)
        else:
            stdout_b, stderr_b = proc.communicate()
            returncode = proc.returncode

    if retcode is not None and retcode != returncode:
        raise CalledProcessError(returncode, cmd, retcode, stdout_b, stderr_b)

    return returncode, stdout_b, stderr_b def make_config_from_repo(repo_path, rev=None, hooks=None, check=True):
    manifest = load_manifest(os.path.join(repo_path, C.MANIFEST_FILE))
    config = {
        'repo': f'file://{repo_path}',
        'rev': rev or git.head_rev(repo_path),
        'hooks': hooks or [{'id': hook['id']} for hook in manifest],
    }

    if check:
        wrapped = validate({'repos': [config]}, CONFIG_SCHEMA)
        wrapped = apply_defaults(wrapped, CONFIG_SCHEMA)
        config, = wrapped['repos']
        return config
    else:
        return config def make_repo(tempdir_factory, repo_source):
    path = git_dir(tempdir_factory)
    copy_tree_to_path(get_resource_path(repo_source), path)
    cmd_output('git', 'add', '.', cwd=path)
    git_commit(msg=make_repo.__name__, cwd=path)
    return path @contextlib.contextmanager
def modify_manifest(path, commit=True):
    """Modify the manifest yielded by this context to write to
    .pre-commit-hooks.yaml.
    """
    manifest_path = os.path.join(path, C.MANIFEST_FILE)
    with open(manifest_path) as f:
        manifest = yaml_load(f.read())
    yield manifest
    with open(manifest_path, 'w') as manifest_file:
        manifest_file.write(yaml_dump(manifest))
    if commit:
        git_commit(msg=modify_manifest.__name__, cwd=path) @contextlib.contextmanager
def cwd(path):
    original_cwd = os.getcwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(original_cwd) def get_resource_path(path):
    return os.path.join(TESTING_DIR, 'resources', path) TESTING_DIR = os.path.abspath(os.path.dirname(__file__))
def docker_is_running() -> bool:  # pragma: win32 no cover
def get_resource_path(path):
def cmd_output_mocked_pre_commit_home(
        *args, tempdir_factory, pre_commit_home=None, env=None, **kwargs,
):
def run_opts(
        all_files=False,
        files=(),
        color=False,
        verbose=False,
        hook=None,
        remote_branch='',
        local_branch='',
        from_ref='',
        to_ref='',
        remote_name='',
        remote_url='',
        hook_stage='commit',
        show_diff_on_failure=False,
        commit_msg_filename='',
        checkout_type='',
        is_squash_merge='',
        rewrite_command='',
):
def cwd(path):
def git_commit(*args, fn=cmd_output, msg='commit!', all_files=True, **kwargs): TESTING_DIR = os.path.abspath(os.path.dirname(__file__))
def docker_is_running() -> bool:  # pragma: win32 no cover
def get_resource_path(path):
def cmd_output_mocked_pre_commit_home(
        *args, tempdir_factory, pre_commit_home=None, env=None, **kwargs,
):
def run_opts(
        all_files=False,
        files=(),
        color=False,
        verbose=False,
        hook=None,
        remote_branch='',
        local_branch='',
        from_ref='',
        to_ref='',
        remote_name='',
        remote_url='',
        hook_stage='commit',
        show_diff_on_failure=False,
        commit_msg_filename='',
        checkout_type='',
        is_squash_merge='',
        rewrite_command='',
):
def cwd(path):
def git_commit(*args, fn=cmd_output, msg='commit!', all_files=True, **kwargs): TESTING_DIR = os.path.abspath(os.path.dirname(__file__))
def docker_is_running() -> bool:  # pragma: win32 no cover
def get_resource_path(path):
def cmd_output_mocked_pre_commit_home(
        *args, tempdir_factory, pre_commit_home=None, env=None, **kwargs,
):
def run_opts(
        all_files=False,
        files=(),
        color=False,
        verbose=False,
        hook=None,
        remote_branch='',
        local_branch='',
        from_ref='',
        to_ref='',
        remote_name='',
        remote_url='',
        hook_stage='commit',
        show_diff_on_failure=False,
        commit_msg_filename='',
        checkout_type='',
        is_squash_merge='',
        rewrite_command='',
):
def cwd(path):
def git_commit(*args, fn=cmd_output, msg='commit!', all_files=True, **kwargs): TESTING_DIR = os.path.abspath(os.path.dirname(__file__))
def docker_is_running() -> bool:  # pragma: win32 no cover
def get_resource_path(path):
def cmd_output_mocked_pre_commit_home(
        *args, tempdir_factory, pre_commit_home=None, env=None, **kwargs,
):
def run_opts(
        all_files=False,
        files=(),
        color=False,
        verbose=False,
        hook=None,
        remote_branch='',
        local_branch='',
        from_ref='',
        to_ref='',
        remote_name='',
        remote_url='',
        hook_stage='commit',
        show_diff_on_failure=False,
        commit_msg_filename='',
        checkout_type='',
        is_squash_merge='',
        rewrite_command='',
):
def cwd(path):
def git_commit(*args, fn=cmd_output, msg='commit!', all_files=True, **kwargs): TESTING_DIR = os.path.abspath(os.path.dirname(__file__))
def docker_is_running() -> bool:  # pragma: win32 no cover
def get_resource_path(path):
def cmd_output_mocked_pre_commit_home(
        *args, tempdir_factory, pre_commit_home=None, env=None, **kwargs,
):
def run_opts(
        all_files=False,
        files=(),
        color=False,
        verbose=False,
        hook=None,
        remote_branch='',
        local_branch='',
        from_ref='',
        to_ref='',
        remote_name='',
        remote_url='',
        hook_stage='commit',
        show_diff_on_failure=False,
        commit_msg_filename='',
        checkout_type='',
        is_squash_merge='',
        rewrite_command='',
):
def cwd(path):
def git_commit(*args, fn=cmd_output, msg='commit!', all_files=True, **kwargs):
import os.path
import shutil
import sys
import cfgv
import pytest
import re_assert
import pre_commit.constants as C
from typing import Any
from unittest import mock
from pre_commit import git
from pre_commit.clientlib import CONFIG_SCHEMA
from pre_commit.clientlib import load_manifest
from pre_commit.envcontext import envcontext
from pre_commit.hook import Hook
from pre_commit.languages import golang
from pre_commit.languages import helpers
from pre_commit.languages import node
from pre_commit.languages import python
from pre_commit.languages import ruby
from pre_commit.languages import rust
from pre_commit.languages.all import languages
from pre_commit.prefix import Prefix
from pre_commit.repository import all_hooks
from pre_commit.repository import install_hook_envs
from pre_commit.util import cmd_output
from pre_commit.util import cmd_output_b
from testing.fixtures import make_config_from_repo
from testing.fixtures import make_repo
from testing.fixtures import modify_manifest
from testing.util import cwd
from testing.util import get_resource_path
from testing.util import skipif_cant_run_coursier
from testing.util import skipif_cant_run_docker
from testing.util import skipif_cant_run_lua
from testing.util import skipif_cant_run_swift
from testing.util import xfailif_windows
        b'0\n1\n2\n3\n4\n5\n',
    )


@pytest.mark.xfail(os.name == 'nt', reason='ptys are posix-only')
def test_output_isatty(tempdir_factory, store):
    _test_hook_repo(
        tempdir_factory, store, 'stdout_stderr_repo',
        'tty-check',
        [],
        b'stdin: False\nstdout: True\nstderr: True\n',
        color=True,
    )


def _make_grep_repo(entry, store, args=()):
    config = {
        'repo': 'local',
        'hooks': [{
            'id': 'grep-hook',
            'name': 'grep-hook',
            'language': 'pygrep',
            'entry': entry,
            'args': args,
            'types': ['text'],
        }],
    }
    return _get_hook(config, store, 'grep-hook')


@pytest.fixture
def greppable_files(tmpdir):
    with tmpdir.as_cwd():
        cmd_output_b('git', 'init', '.')
        tmpdir.join('f1').write_binary(b"hello'hi\nworld\n")
        tmpdir.join('f2').write_binary(b'foo\nbar\nbaz\n')
        tmpdir.join('f3').write_binary(b'[WARN] hi\n')
        yield tmpdir


def test_grep_hook_matching(greppable_files, store):
    hook = _make_grep_repo('ello', store)
    ret, out = _hook_run(hook, ('f1', 'f2', 'f3'), color=False)
    assert ret == 1
    assert _norm_out(out) == b"f1:1:hello'hi\n"


def test_grep_hook_case_insensitive(greppable_files, store):
    hook = _make_grep_repo('ELLO', store, args=['-i'])
    ret, out = _hook_run(hook, ('f1', 'f2', 'f3'), color=False)
    assert ret == 1
    assert _norm_out(out) == b"f1:1:hello'hi\n"


@pytest.mark.parametrize('regex', ('nope', "foo'bar", r'^\[INFO\]'))
def test_grep_hook_not_matching(regex, greppable_files, store):
    hook = _make_grep_repo(regex, store)
    ret, out = _hook_run(hook, ('f1', 'f2', 'f3'), color=False)
    assert (ret, out) == (0, b'')


def _norm_pwd(path):
    # Under windows bash's temp and windows temp is different.
    # This normalizes to the bash /tmp
    return cmd_output_b(
        'bash', '-c', f"cd '{path}' && pwd",
    )[1].strip()


def test_cwd_of_hook(in_git_dir, tempdir_factory, store):
    # Note: this doubles as a test for `system` hooks
    _test_hook_repo(
        tempdir_factory, store, 'prints_cwd_repo',
        'prints_cwd', ['-L'], _norm_pwd(in_git_dir.strpath) + b'\n',
    )


def test_lots_of_files(tempdir_factory, store):
    _test_hook_repo(
        tempdir_factory, store, 'script_hooks_repo',
        'bash_hook', [os.devnull] * 15000, mock.ANY,
    )


def test_additional_dependencies_roll_forward(tempdir_factory, store):
    path = make_repo(tempdir_factory, 'python_hooks_repo')

    config1 = make_config_from_repo(path)
    hook1 = _get_hook(config1, store, 'foo')
    with python.in_env(hook1.prefix, hook1.language_version):
        assert 'mccabe' not in cmd_output('pip', 'freeze', '-l')[1]

    # Make another repo with additional dependencies
    config2 = make_config_from_repo(path)
    config2['hooks'][0]['additional_dependencies'] = ['mccabe']
    hook2 = _get_hook(config2, store, 'foo')
    with python.in_env(hook2.prefix, hook2.language_version):
        assert 'mccabe' in cmd_output('pip', 'freeze', '-l')[1]

    # should not have affected original
    with python.in_env(hook1.prefix, hook1.language_version):
        assert 'mccabe' not in cmd_output('pip', 'freeze', '-l')[1]


def test_additional_ruby_dependencies_installed(tempdir_factory, store):
    path = make_repo(tempdir_factory, 'ruby_hooks_repo')
    config = make_config_from_repo(path)
    config['hooks'][0]['additional_dependencies'] = ['tins']
    hook = _get_hook(config, store, 'ruby_hook')
    with ruby.in_env(hook.prefix, hook.language_version):
        output = cmd_output('gem', 'list', '--local')[1]
        assert 'tins' in output


def test_additional_node_dependencies_installed(tempdir_factory, store):
    path = make_repo(tempdir_factory, 'node_hooks_repo')
    config = make_config_from_repo(path)
    # Careful to choose a small package that's not depped by npm
    config['hooks'][0]['additional_dependencies'] = ['lodash']
    hook = _get_hook(config, store, 'foo')

    with node.in_env(hook.prefix, hook.language_version):