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: 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): def autoupdate(
        config_file: str,
        store: Store,
        tags_only: bool,
        freeze: bool,
        repos: Sequence[str] = (),
) -> int:
    """Auto-update the pre-commit config to the latest versions of repos."""
    migrate_config(config_file, quiet=True)
    retv = 0
    rev_infos: list[RevInfo | None] = []
    changed = False

    config = load_config(config_file)
    for repo_config in config['repos']:
        if repo_config['repo'] in {LOCAL, META}:
            continue

        info = RevInfo.from_config(repo_config)
        if repos and info.repo not in repos:
            rev_infos.append(None)
            continue

        output.write(f'Updating {info.repo} ... ')
        new_info = info.update(tags_only=tags_only, freeze=freeze)
        try:
            _check_hooks_still_exist_at_rev(repo_config, new_info, store)
        except RepositoryCannotBeUpdatedError as error:
            output.write_line(error.args[0])
            rev_infos.append(None)
            retv = 1
            continue

        if new_info.rev != info.rev:
            changed = True
            if new_info.frozen:
                updated_to = f'{new_info.frozen} (frozen)'
            else:
                updated_to = new_info.rev
            msg = f'updating {info.rev} -> {updated_to}.'
            output.write_line(msg)
            rev_infos.append(new_info)
        else:
            output.write_line('already up to date.')
            rev_infos.append(None)

    if changed:
        _write_new_config(config_file, rev_infos)

    return retv def gc(store: Store) -> int:
    with store.exclusive_lock():
        repos_removed = _gc_repos(store)
    output.write_line(f'{repos_removed} repo(s) removed.')
    return 0 def install_hooks(config_file: str, store: Store) -> int:
    install_hook_envs(all_hooks(load_config(config_file), store), store)
    return 0 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 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_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 sample_local_config():
    return {
        'repo': 'local',
        'hooks': [{
            'id': 'do_not_commit',
            'name': 'Block if "DO NOT COMMIT" is found',
            'entry': 'DO NOT COMMIT',
            'language': 'pygrep',
        }],
    } 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 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')
import os
import pre_commit.constants as C
from pre_commit import git
from pre_commit.clientlib import load_config
from pre_commit.commands.autoupdate import autoupdate
from pre_commit.commands.gc import gc
from pre_commit.commands.install_uninstall import install_hooks
from pre_commit.repository import all_hooks
from testing.fixtures import make_config_from_repo
from testing.fixtures import make_repo
from testing.fixtures import modify_config
from testing.fixtures import sample_local_config
from testing.fixtures import sample_meta_config
from testing.fixtures import write_config
from testing.util import git_commit
from __future__ import annotations




def _repo_count(store):
    return len(store.select_all_repos())


def _config_count(store):
    return len(store.select_all_configs())


def _remove_config_assert_cleared(store, cap_out):
    os.remove(C.CONFIG_FILE)
    assert not gc(store)
    assert _config_count(store) == 0
    assert _repo_count(store) == 0
    assert cap_out.get().splitlines()[-1] == '1 repo(s) removed.'


def test_gc(tempdir_factory, store, in_git_dir, cap_out):
    path = make_repo(tempdir_factory, 'script_hooks_repo')
    old_rev = git.head_rev(path)
    git_commit(cwd=path)

    write_config('.', make_config_from_repo(path, rev=old_rev))
    store.mark_config_used(C.CONFIG_FILE)

    # update will clone both the old and new repo, making the old one gc-able
    install_hooks(C.CONFIG_FILE, store)
    assert not autoupdate(C.CONFIG_FILE, store, freeze=False, tags_only=False)

    assert _config_count(store) == 1
    assert _repo_count(store) == 2
    assert not gc(store)
    assert _config_count(store) == 1
    assert _repo_count(store) == 1
    assert cap_out.get().splitlines()[-1] == '1 repo(s) removed.'

    _remove_config_assert_cleared(store, cap_out)


def test_gc_repo_not_cloned(tempdir_factory, store, in_git_dir, cap_out):
    path = make_repo(tempdir_factory, 'script_hooks_repo')
    write_config('.', make_config_from_repo(path))
    store.mark_config_used(C.CONFIG_FILE)

    assert _config_count(store) == 1
    assert _repo_count(store) == 0
    assert not gc(store)
    assert _config_count(store) == 1
    assert _repo_count(store) == 0
    assert cap_out.get().splitlines()[-1] == '0 repo(s) removed.'


def test_gc_meta_repo_does_not_crash(store, in_git_dir, cap_out):
    write_config('.', sample_meta_config())
    store.mark_config_used(C.CONFIG_FILE)
    assert not gc(store)
    assert cap_out.get().splitlines()[-1] == '0 repo(s) removed.'


def test_gc_local_repo_does_not_crash(store, in_git_dir, cap_out):
    write_config('.', sample_local_config())
    store.mark_config_used(C.CONFIG_FILE)
    assert not gc(store)
    assert cap_out.get().splitlines()[-1] == '0 repo(s) removed.'


def test_gc_unused_local_repo_with_env(store, in_git_dir, cap_out):
    config = {
        'repo': 'local',
        'hooks': [{
            'id': 'flake8', 'name': 'flake8', 'entry': 'flake8',
            # a `language: python` local hook will create an environment
            'types': ['python'], 'language': 'python',
        }],
    }
    write_config('.', config)
    store.mark_config_used(C.CONFIG_FILE)

    # this causes the repositories to be created
    all_hooks(load_config(C.CONFIG_FILE), store)

    assert _config_count(store) == 1
    assert _repo_count(store) == 1
    assert not gc(store)
    assert _config_count(store) == 1
    assert _repo_count(store) == 1
    assert cap_out.get().splitlines()[-1] == '0 repo(s) removed.'

    _remove_config_assert_cleared(store, cap_out)


def test_gc_config_with_missing_hook(
        tempdir_factory, store, in_git_dir, cap_out,
):
    path = make_repo(tempdir_factory, 'script_hooks_repo')
    write_config('.', make_config_from_repo(path))
    store.mark_config_used(C.CONFIG_FILE)
    # to trigger a clone

    all_hooks(load_config(C.CONFIG_FILE), store)