def write(s: str, stream: IO[bytes] = sys.stdout.buffer) -> None:
def write_line_b(
        s: bytes | None = None,
        stream: IO[bytes] = sys.stdout.buffer,
        logfile_name: str | None = None,
) -> None:
def write_line(s: str | None = None, **kwargs: Any) -> None: @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) @contextlib.contextmanager
def logging_handler(use_color: bool) -> Generator[None, None, None]:
    handler = LoggingHandler(use_color)
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    try:
        yield
    finally:
        logger.removeHandler(handler) class Store:
    get_default_directory = staticmethod(_get_default_directory)

    def __init__(self, directory: str | None = None) -> None:
        self.directory = directory or Store.get_default_directory()
        self.db_path = os.path.join(self.directory, 'db.db')
        self.readonly = (
            os.path.exists(self.directory) and
            not os.access(self.directory, os.W_OK)
        )

        if not os.path.exists(self.directory):
            os.makedirs(self.directory, exist_ok=True)
            with open(os.path.join(self.directory, 'README'), 'w') as f:
                f.write(
                    'This directory is maintained by the pre-commit project.\n'
                    'Learn more: https://github.com/pre-commit/pre-commit\n',
                )

        if os.path.exists(self.db_path):
            return
        with self.exclusive_lock():
            # Another process may have already completed this work
            if os.path.exists(self.db_path):  # pragma: no cover (race)
                return
            # To avoid a race where someone ^Cs between db creation and
            # execution of the CREATE TABLE statement
            fd, tmpfile = tempfile.mkstemp(dir=self.directory)
            # We'll be managing this file ourselves
            os.close(fd)
            with self.connect(db_path=tmpfile) as db:
                db.executescript(
                    'CREATE TABLE repos ('
                    '    repo TEXT NOT NULL,'
                    '    ref TEXT NOT NULL,'
                    '    path TEXT NOT NULL,'
                    '    PRIMARY KEY (repo, ref)'
                    ');',
                )
                self._create_config_table(db)

            # Atomic file move
            os.replace(tmpfile, self.db_path)

    @contextlib.contextmanager
    def exclusive_lock(self) -> Generator[None, None, None]:
        def blocked_cb() -> None:  # pragma: no cover (tests are in-process)
            logger.info('Locking pre-commit directory')

        with file_lock.lock(os.path.join(self.directory, '.lock'), blocked_cb):
            yield

    @contextlib.contextmanager
    def connect(
            self,
            db_path: str | None = None,
    ) -> Generator[sqlite3.Connection, None, None]:
        db_path = db_path or self.db_path
        # sqlite doesn't close its fd with its contextmanager >.<
        # contextlib.closing fixes this.
        # See: https://stackoverflow.com/a/28032829/812183
        with contextlib.closing(sqlite3.connect(db_path)) as db:
            # this creates a transaction
            with db:
                yield db

    @classmethod
    def db_repo_name(cls, repo: str, deps: Sequence[str]) -> str:
        if deps:
            return f'{repo}:{",".join(sorted(deps))}'
        else:
            return repo

    def _new_repo(
            self,
            repo: str,
            ref: str,
            deps: Sequence[str],
            make_strategy: Callable[[str], None],
    ) -> str:
        repo = self.db_repo_name(repo, deps)

        def _get_result() -> str | None:
            # Check if we already exist
            with self.connect() as db:
                result = db.execute(
                    'SELECT path FROM repos WHERE repo = ? AND ref = ?',
                    (repo, ref),
                ).fetchone()
                return result[0] if result else None

        result = _get_result()
        if result:
            return result
        with self.exclusive_lock():
            # Another process may have already completed this work
            result = _get_result()
            if result:  # pragma: no cover (race)
                return result

            logger.info(f'Initializing environment for {repo}.')

            directory = tempfile.mkdtemp(prefix='repo', dir=self.directory)
            with clean_path_on_failure(directory):
                make_strategy(directory)

            # Update our db with the created repo
            with self.connect() as db:
                db.execute(
                    'INSERT INTO repos (repo, ref, path) VALUES (?, ?, ?)',
                    [repo, ref, directory],
                )
        return directory

    def _complete_clone(self, ref: str, git_cmd: Callable[..., None]) -> None:
        """Perform a complete clone of a repository and its submodules """

        git_cmd('fetch', 'origin', '--tags')
        git_cmd('checkout', ref)
        git_cmd('submodule', 'update', '--init', '--recursive')

    def _shallow_clone(self, ref: str, git_cmd: Callable[..., None]) -> None:
        """Perform a shallow clone of a repository and its submodules """

        git_config = 'protocol.version=2'
        git_cmd('-c', git_config, 'fetch', 'origin', ref, '--depth=1')
        git_cmd('checkout', 'FETCH_HEAD')
        git_cmd(
            '-c', git_config, 'submodule', 'update', '--init', '--recursive',
            '--depth=1',
        )

    def clone(self, repo: str, ref: str, deps: Sequence[str] = ()) -> str:
        """Clone the given url and checkout the specific ref."""

        def clone_strategy(directory: str) -> None:
            git.init_repo(directory, repo)
            env = git.no_git_env()

            def _git_cmd(*args: str) -> None:
                cmd_output_b('git', *args, cwd=directory, env=env)

            try:
                self._shallow_clone(ref, _git_cmd)
            except CalledProcessError:
                self._complete_clone(ref, _git_cmd)

        return self._new_repo(repo, ref, deps, clone_strategy)

    LOCAL_RESOURCES = (
        'Cargo.toml', 'main.go', 'go.mod', 'main.rs', '.npmignore',
        'package.json', 'pre-commit-package-dev-1.rockspec',
        'pre_commit_placeholder_package.gemspec', 'setup.py',
        'environment.yml', 'Makefile.PL', 'pubspec.yaml',
        'renv.lock', 'renv/activate.R', 'renv/LICENSE.renv',
    )

    def make_local(self, deps: Sequence[str]) -> str:
        def make_local_strategy(directory: str) -> None:
            for resource in self.LOCAL_RESOURCES:
                resource_dirname, resource_basename = os.path.split(resource)
                contents = resource_text(f'empty_template_{resource_basename}')
                target_dir = os.path.join(directory, resource_dirname)
                target_file = os.path.join(target_dir, resource_basename)
                os.makedirs(target_dir, exist_ok=True)
                with open(target_file, 'w') as f:
                    f.write(contents)

            env = git.no_git_env()

            # initialize the git repository so it looks more like cloned repos
            def _git_cmd(*args: str) -> None:
                cmd_output_b('git', *args, cwd=directory, env=env)

            git.init_repo(directory, '<<unknown>>')
            _git_cmd('add', '.')
            git.commit(repo=directory)

        return self._new_repo(
            'local', C.LOCAL_REPO_VERSION, deps, make_local_strategy,
        )

    def _create_config_table(self, db: sqlite3.Connection) -> None:
        db.executescript(
            'CREATE TABLE IF NOT EXISTS configs ('
            '   path TEXT NOT NULL,'
            '   PRIMARY KEY (path)'
            ');',
        )

    def mark_config_used(self, path: str) -> None:
        if self.readonly:  # pragma: win32 no cover
            return
        path = os.path.realpath(path)
        # don't insert config files that do not exist
        if not os.path.exists(path):
            return
        with self.connect() as db:
            # TODO: eventually remove this and only create in _create
            self._create_config_table(db)
            db.execute('INSERT OR IGNORE INTO configs VALUES (?)', (path,))

    def select_all_configs(self) -> list[str]:
        with self.connect() as db:
            self._create_config_table(db)
            rows = db.execute('SELECT path FROM configs').fetchall()
            return [path for path, in rows]

    def delete_configs(self, configs: list[str]) -> None:
        with self.connect() as db:
            rows = [(path,) for path in configs]
            db.executemany('DELETE FROM configs WHERE path = ?', rows)

    def select_all_repos(self) -> list[tuple[str, str, str]]:
        with self.connect() as db:
            return db.execute('SELECT repo, ref, path from repos').fetchall()

    def delete_repo(self, db_repo_name: str, ref: str, path: str) -> None:
        with self.connect() as db:
            db.execute(
                'DELETE FROM repos WHERE repo = ? and ref = ?',
                (db_repo_name, ref),
            )
        rmtree(path) 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 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) 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)) @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')
import functools
import io
import logging
import os.path
import pytest
from unittest import mock
from pre_commit import output
from pre_commit.envcontext import envcontext
from pre_commit.logging_handler import logging_handler
from pre_commit.store import Store
from pre_commit.util import cmd_output
from pre_commit.util import make_executable
from testing.fixtures import git_dir
from testing.fixtures import make_consuming_repo
from testing.fixtures import write_config
from testing.util import cwd
from testing.util import git_commit


@pytest.fixture
def tempdir_factory(tmpdir):
    class TmpdirFactory:
        def __init__(self):
            self.tmpdir_count = 0

        def get(self):
            path = tmpdir.join(str(self.tmpdir_count)).strpath
            self.tmpdir_count += 1
            os.mkdir(path)
            return path

    yield TmpdirFactory()


@pytest.fixture
def in_tmpdir(tempdir_factory):
    path = tempdir_factory.get()
    with cwd(path):
        yield path


@pytest.fixture
def in_git_dir(tmpdir):
    repo = tmpdir.join('repo').ensure_dir()
    with repo.as_cwd():
        cmd_output('git', 'init')
        yield repo


def _make_conflict():
    cmd_output('git', 'checkout', 'origin/master', '-b', 'foo')
    with open('conflict_file', 'w') as conflict_file:
        conflict_file.write('herp\nderp\n')
    cmd_output('git', 'add', 'conflict_file')
    with open('foo_only_file', 'w') as foo_only_file:
        foo_only_file.write('foo')
    cmd_output('git', 'add', 'foo_only_file')
    git_commit(msg=_make_conflict.__name__)
    cmd_output('git', 'checkout', 'origin/master', '-b', 'bar')
    with open('conflict_file', 'w') as conflict_file:
        conflict_file.write('harp\nddrp\n')
    cmd_output('git', 'add', 'conflict_file')
    with open('bar_only_file', 'w') as bar_only_file:
        bar_only_file.write('bar')
    cmd_output('git', 'add', 'bar_only_file')
    git_commit(msg=_make_conflict.__name__)
    cmd_output('git', 'merge', 'foo', retcode=None)


@pytest.fixture
def in_merge_conflict(tempdir_factory):
    path = make_consuming_repo(tempdir_factory, 'script_hooks_repo')
    open(os.path.join(path, 'placeholder'), 'a').close()
    cmd_output('git', 'add', 'placeholder', cwd=path)
    git_commit(msg=in_merge_conflict.__name__, cwd=path)

    conflict_path = tempdir_factory.get()
    cmd_output('git', 'clone', path, conflict_path)
    with cwd(conflict_path):
        _make_conflict()
        yield os.path.join(conflict_path)


@pytest.fixture
def in_conflicting_submodule(tempdir_factory):
    git_dir_1 = git_dir(tempdir_factory)
    git_dir_2 = git_dir(tempdir_factory)
    git_commit(msg=in_conflicting_submodule.__name__, cwd=git_dir_2)
    cmd_output('git', 'submodule', 'add', git_dir_2, 'sub', cwd=git_dir_1)
    with cwd(os.path.join(git_dir_1, 'sub')):
        _make_conflict()
        yield


@pytest.fixture
def commit_msg_repo(tempdir_factory):
    path = git_dir(tempdir_factory)
    config = {
        'repo': 'local',
        'hooks': [{
            'id': 'must-have-signoff',
            'name': 'Must have "Signed off by:"',
            'entry': 'grep -q "Signed off by:"',
            'language': 'system',
            'stages': ['commit-msg'],
        }],
    }
    write_config(path, config)
    with cwd(path):
        cmd_output('git', 'add', '.')
        git_commit(msg=commit_msg_repo.__name__)
        yield path


@pytest.fixture
def prepare_commit_msg_repo(tempdir_factory):
    path = git_dir(tempdir_factory)
    script_name = 'add_sign_off.sh'
    config = {
        'repo': 'local',
        'hooks': [{
            'id': 'add-signoff',
            'name': 'Add "Signed off by:"',
            'entry': f'./{script_name}',
            'language': 'script',
            'stages': ['prepare-commit-msg'],
        }],
    }
    write_config(path, config)
    with cwd(path):
        with open(script_name, 'w') as script_file:
            script_file.write(
                '#!/usr/bin/env bash\n'
                'set -eu\n'
                'echo "\nSigned off by: " >> "$1"\n',
            )
            make_executable(script_name)

        cmd_output('git', 'add', '.')