    def _enable() -> None:
        def bool_errcheck(result, func, args):
def format_color(text: str, color: str, use_color_setting: bool) -> str:
def use_color(setting: str) -> bool:
def add_color_option(parser: argparse.ArgumentParser) -> None:
        STD_ERROR_HANDLE = -12
        ENABLE_VIRTUAL_TERMINAL_PROCESSING = 4
RED = '\033[41m'
GREEN = '\033[42m'
YELLOW = '\033[43;30m'
TURQUOISE = '\033[46;30m'
SUBTLE = '\033[2m'
NORMAL = '\033[m'
COLOR_CHOICES = ('auto', 'always', 'never') def install(
        config_file: str,
        store: Store,
        hook_types: Sequence[str],
        overwrite: bool = False,
        hooks: bool = False,
        skip_on_missing_config: bool = False,
        git_dir: str | None = None,
) -> int:
    if git_dir is None and git.has_core_hookpaths_set():
        logger.error(
            'Cowardly refusing to install hooks with `core.hooksPath` set.\n'
            'hint: `git config --unset-all core.hooksPath`',
        )
        return 1

    for hook_type in hook_types:
        _install_hook_script(
            config_file, hook_type,
            overwrite=overwrite,
            skip_on_missing_config=skip_on_missing_config,
            git_dir=git_dir,
        )

    if hooks:
        install_hooks(config_file, store)

    return 0 def _compute_cols(hooks: Sequence[Hook]) -> int:
    """Compute the number of columns to display hook messages.  The widest
    that will be displayed is in the no files skipped case:

        Hook name...(no files to check) Skipped
    """
    if hooks:
        name_len = max(_len_cjk(hook.name) for hook in hooks)
    else:
        name_len = 0

    cols = name_len + 3 + len(NO_FILES) + 1 + len(SKIPPED)
    return max(cols, 80) def _full_msg(
        *,
        start: str,
        cols: int,
        end_msg: str,
        end_color: str,
        use_color: bool,
        postfix: str = '',
) -> str:
    dots = '.' * (cols - _len_cjk(start) - len(postfix) - len(end_msg) - 1)
    end = color.format_color(end_msg, end_color, use_color)
    return f'{start}{dots}{postfix}{end}\n' def _get_skips(environ: MutableMapping[str, str]) -> set[str]:
    skips = environ.get('SKIP', '')
    return {skip.strip() for skip in skips.split(',') if skip.strip()} def _has_unmerged_paths() -> bool:
    _, stdout, _ = cmd_output_b('git', 'ls-files', '--unmerged')
    return bool(stdout.strip()) def _start_msg(*, start: str, cols: int, end_len: int) -> str:
    dots = '.' * (cols - _len_cjk(start) - end_len - 1)
    return f'{start}{dots}' class Classifier:
    def __init__(self, filenames: Collection[str]) -> None:
        self.filenames = [f for f in filenames if os.path.lexists(f)]

    @functools.lru_cache(maxsize=None)
    def _types_for_file(self, filename: str) -> set[str]:
        return tags_from_path(filename)

    def by_types(
            self,
            names: Sequence[str],
            types: Collection[str],
            types_or: Collection[str],
            exclude_types: Collection[str],
    ) -> list[str]:
        types = frozenset(types)
        types_or = frozenset(types_or)
        exclude_types = frozenset(exclude_types)
        ret = []
        for filename in names:
            tags = self._types_for_file(filename)
            if (
                    tags >= types and
                    (not types_or or tags & types_or) and
                    not tags & exclude_types
            ):
                ret.append(filename)
        return ret

    def filenames_for_hook(self, hook: Hook) -> tuple[str, ...]:
        names = self.filenames
        names = filter_by_include_exclude(names, hook.files, hook.exclude)
        names = self.by_types(
            names,
            hook.types,
            hook.types_or,
            hook.exclude_types,
        )
        return tuple(names)

    @classmethod
    def from_config(
            cls,
            filenames: Collection[str],
            include: str,
            exclude: str,
    ) -> Classifier:
        # on windows we normalize all filenames to use forward slashes
        # this makes it easier to filter using the `files:` regex
        # this also makes improperly quoted shell-based hooks work better
        # see #1173
        if os.altsep == '/' and os.sep == '\\':
            filenames = [f.replace(os.sep, os.altsep) for f in filenames]
        filenames = filter_by_include_exclude(filenames, include, exclude)
        return Classifier(filenames) def filter_by_include_exclude(
        names: Collection[str],
        include: str,
        exclude: str,
) -> list[str]:
    include_re, exclude_re = re.compile(include), re.compile(exclude)
    return [
        filename for filename in names
        if include_re.search(filename)
        if not exclude_re.search(filename)
    ] def run(
        config_file: str,
        store: Store,
        args: argparse.Namespace,
        environ: MutableMapping[str, str] = os.environ,
) -> int:
    stash = not args.all_files and not args.files

    # Check if we have unresolved merge conflict files and fail fast.
    if _has_unmerged_paths():
        logger.error('Unmerged files.  Resolve before committing.')
        return 1
    if bool(args.from_ref) != bool(args.to_ref):
        logger.error('Specify both --from-ref and --to-ref.')
        return 1
    if stash and _has_unstaged_config(config_file):
        logger.error(
            f'Your pre-commit configuration is unstaged.\n'
            f'`git add {config_file}` to fix this.',
        )
        return 1
    if (
            args.hook_stage in {'prepare-commit-msg', 'commit-msg'} and
            not args.commit_msg_filename
    ):
        logger.error(
            f'`--commit-msg-filename` is required for '
            f'`--hook-stage {args.hook_stage}`',
        )
        return 1
    # prevent recursive post-checkout hooks (#1418)
    if (
            args.hook_stage == 'post-checkout' and
            environ.get('_PRE_COMMIT_SKIP_POST_CHECKOUT')
    ):
        return 0

    # Expose from-ref / to-ref as environment variables for hooks to consume
    if args.from_ref and args.to_ref:
        # legacy names
        environ['PRE_COMMIT_ORIGIN'] = args.from_ref
        environ['PRE_COMMIT_SOURCE'] = args.to_ref
        # new names
        environ['PRE_COMMIT_FROM_REF'] = args.from_ref
        environ['PRE_COMMIT_TO_REF'] = args.to_ref

    if (
        args.remote_name and args.remote_url and
        args.remote_branch and args.local_branch
    ):
        environ['PRE_COMMIT_LOCAL_BRANCH'] = args.local_branch
        environ['PRE_COMMIT_REMOTE_BRANCH'] = args.remote_branch
        environ['PRE_COMMIT_REMOTE_NAME'] = args.remote_name
        environ['PRE_COMMIT_REMOTE_URL'] = args.remote_url

    if args.checkout_type:
        environ['PRE_COMMIT_CHECKOUT_TYPE'] = args.checkout_type

    if args.is_squash_merge:
        environ['PRE_COMMIT_IS_SQUASH_MERGE'] = args.is_squash_merge

    if args.rewrite_command:
        environ['PRE_COMMIT_REWRITE_COMMAND'] = args.rewrite_command

    # Set pre_commit flag
    environ['PRE_COMMIT'] = '1'

    with contextlib.ExitStack() as exit_stack:
        if stash:
            exit_stack.enter_context(staged_files_only(store.directory))

        config = load_config(config_file)
        hooks = [
            hook
            for hook in all_hooks(config, store)
            if not args.hook or hook.id == args.hook or hook.alias == args.hook
            if args.hook_stage in hook.stages
        ]

        if args.hook and not hooks:
            output.write_line(
                f'No hook with id `{args.hook}` in stage `{args.hook_stage}`',
            )
            return 1

        skips = _get_skips(environ)
        to_install = [hook for hook in hooks if hook.id not in skips]
        install_hook_envs(to_install, store)

        return _run_hooks(config, hooks, skips, args)

    # https://github.com/python/mypy/issues/7726
    raise AssertionError('unreachable') 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 make_executable(filename: str) -> None:
    original_mode = os.stat(filename).st_mode
    new_mode = original_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
    os.chmod(filename, new_mode) def auto_namedtuple(classname='auto_namedtuple', **kwargs):
    """Returns an automatic namedtuple object.

    Args:
        classname - The class name for the returned object.
        **kwargs - Properties to give the returned object.
    """
    return (collections.namedtuple(classname, kwargs.keys())(**kwargs)) def add_config_to_repo(git_path, config, config_file=C.CONFIG_FILE):
    write_config(git_path, config, config_file=config_file)
    cmd_output('git', 'add', config_file, cwd=git_path)
    git_commit(msg=add_config_to_repo.__name__, cwd=git_path)
    return git_path def git_dir(tempdir_factory):
    path = tempdir_factory.get()
    cmd_output('git', 'init', path)
    return path def make_consuming_repo(tempdir_factory, repo_source):
    path = make_repo(tempdir_factory, repo_source)
    config = make_config_from_repo(path)
    git_path = git_dir(tempdir_factory)
    return add_config_to_repo(git_path, config) @contextlib.contextmanager
def modify_config(path='.', commit=True):
    """Modify the config yielded by this context to write to
    .pre-commit-config.yaml
    """
    config_path = os.path.join(path, C.CONFIG_FILE)
    with open(config_path) as f:
        config = yaml_load(f.read())
    yield config
    with open(config_path, 'w', encoding='UTF-8') as config_file:
        config_file.write(yaml_dump(config))
    if commit:
        git_commit(msg=modify_config.__name__, cwd=path) def read_config(directory, config_file=C.CONFIG_FILE):
    config_path = os.path.join(directory, config_file)
    with open(config_path) as f:
        config = yaml_load(f.read())
    return config def sample_meta_config():
    return {'repo': 'meta', 'hooks': [{'id': 'check-useless-excludes'}]} def write_config(directory, config, config_file=C.CONFIG_FILE):
    if type(config) is not list and 'repos' not in config:
        assert isinstance(config, dict), config
        config = {'repos': [config]}
    with open(os.path.join(directory, config_file), 'w') as outfile:
        outfile.write(yaml_dump(config)) def cmd_output_mocked_pre_commit_home(
        *args, tempdir_factory, pre_commit_home=None, env=None, **kwargs,
):
    if pre_commit_home is None:
        pre_commit_home = tempdir_factory.get()
    env = env if env is not None else os.environ
    kwargs.setdefault('stderr', subprocess.STDOUT)
    # Don't want to write to the home directory
    env = dict(env, PRE_COMMIT_HOME=pre_commit_home)
    ret, out, _ = cmd_output(*args, env=env, **kwargs)
    return ret, out.replace('\r\n', '\n'), None @contextlib.contextmanager
def cwd(path):
    original_cwd = os.getcwd()
    os.chdir(path)
    try:
        yield
    finally:
        os.chdir(original_cwd) def git_commit(*args, fn=cmd_output, msg='commit!', all_files=True, **kwargs):
    kwargs.setdefault('stderr', subprocess.STDOUT)

    cmd = ('git', 'commit', '--allow-empty', '--no-gpg-sign', *args)
    if all_files:  # allow skipping `-a` with `all_files=False`
        cmd += ('-a',)
    if msg is not None:  # allow skipping `-m` with `msg=None`
        cmd += ('-m', msg)
    ret, out, _ = fn(*cmd, **kwargs)
    return ret, out.replace('\r\n', '\n') 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='',
):
    # These are mutually exclusive
    assert not (all_files and files)
    return auto_namedtuple(
        all_files=all_files,
        files=files,
        color=color,
        verbose=verbose,
        hook=hook,
        remote_branch=remote_branch,
        local_branch=local_branch,
        from_ref=from_ref,
        to_ref=to_ref,
        remote_name=remote_name,
        remote_url=remote_url,
        hook_stage=hook_stage,
        show_diff_on_failure=show_diff_on_failure,
        commit_msg_filename=commit_msg_filename,
        checkout_type=checkout_type,
        is_squash_merge=is_squash_merge,
        rewrite_command=rewrite_command,
    )
import os.path
import shlex
import sys
import time
import pytest
import pre_commit.constants as C
from typing import MutableMapping
from unittest import mock
from pre_commit import color
from pre_commit.commands.install_uninstall import install
from pre_commit.commands.run import _compute_cols
from pre_commit.commands.run import _full_msg
from pre_commit.commands.run import _get_skips
from pre_commit.commands.run import _has_unmerged_paths
from pre_commit.commands.run import _start_msg
from pre_commit.commands.run import Classifier
from pre_commit.commands.run import filter_by_include_exclude
from pre_commit.commands.run import run
from pre_commit.util import cmd_output
from pre_commit.util import make_executable
from testing.auto_namedtuple import auto_namedtuple
from testing.fixtures import add_config_to_repo
from testing.fixtures import git_dir
from testing.fixtures import make_consuming_repo
from testing.fixtures import modify_config
from testing.fixtures import read_config
from testing.fixtures import sample_meta_config
from testing.fixtures import write_config
from testing.util import cmd_output_mocked_pre_commit_home
from testing.util import cwd
from testing.util import git_commit
from testing.util import run_opts
from __future__ import annotations





def test_start_msg():
    ret = _start_msg(start='start', end_len=5, cols=15)
    # 4 dots: 15 - 5 - 5 - 1
    assert ret == 'start....'


def test_full_msg():
    ret = _full_msg(
        start='start',
        end_msg='end',
        end_color='',
        use_color=False,
        cols=15,
    )
    # 6 dots: 15 - 5 - 3 - 1
    assert ret == 'start......end\n'


def test_full_msg_with_cjk():
    ret = _full_msg(
        start='啊あ아',
        end_msg='end',
        end_color='',
        use_color=False,
        cols=15,
    )
    # 5 dots: 15 - 6 - 3 - 1
    assert ret == '啊あ아.....end\n'


def test_full_msg_with_color():
    ret = _full_msg(
        start='start',
        end_msg='end',
        end_color=color.RED,
        use_color=True,
        cols=15,
    )
    # 6 dots: 15 - 5 - 3 - 1
    assert ret == f'start......{color.RED}end{color.NORMAL}\n'


def test_full_msg_with_postfix():

    ret = _full_msg(