test_index.py

# Copyright (C) 2008, 2009 Michael Trier (mtrier@gmail.com) and contributors
#
# This module is part of GitPython and is released under the
# 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/

from io import BytesIO
import logging
import os
import os.path as osp
from pathlib import Path
from stat import S_ISLNK, ST_MODE
import subprocess
import tempfile

import pytest
from sumtypes import constructor, sumtype

from git import (
    IndexFile,
    Repo,
    BlobFilter,
    UnmergedEntriesError,
    Tree,
    Object,
    Diff,
    GitCommandError,
    CheckoutError,
)
from git.exc import HookExecutionError, InvalidGitRepositoryError
from git.index.fun import hook_path
from git.index.typ import BaseIndexEntry, IndexEntry
from git.objects import Blob
from test.lib import TestBase, fixture, fixture_path, with_rw_directory, with_rw_repo
from git.util import Actor, hex_to_bin, rmtree
from gitdb.base import IStream

HOOKS_SHEBANG = "#!/usr/bin/env sh\n"

log = logging.getLogger(__name__)


@sumtype
class _WinBashStatus:
    """Status of bash.exe for native Windows. Affects which commit hook tests can pass.

    Call :meth:`check` to check the status.
    """

    Inapplicable = constructor()
    """This system is not native Windows: either not Windows at all, or Cygwin."""

    Absent = constructor()
    """No command for ``bash.exe`` is found on the system."""

    Native = constructor()
    """Running ``bash.exe`` operates outside any WSL distribution (as with Git Bash)."""

    Wsl = constructor()
    """Running ``bash.exe`` calls ``bash`` in a WSL distribution."""

    WslNoDistro = constructor()
    """Running ``bash.exe` tries to run bash on a WSL distribution, but none exists."""

    ErrorWhileChecking = constructor("error_or_process")
    """Could not determine the status.

    This should not trigger a skip or xfail, as it typically indicates either a fixable
    problem on the test machine, such as an "Insufficient system resources exist to
    complete the requested service" error starting WSL, or a bug in this detection code.
    """

    @classmethod
    def check(cls):
        """Check the status of the ``bash.exe`` :func:`index.fun.run_commit_hook` uses.

        This uses EAFP, attempting to run a command via ``bash.exe``. Which ``bash.exe``
        is used can't be reliably discovered by :func:`shutil.which`, which approximates
        how a shell is expected to search for an executable. On Windows, there are major
        differences between how executables are found by a shell and otherwise. (This is
        the cmd.exe Windows shell, and shouldn't be confused with bash.exe itself. That
        the command being looked up also happens to be an interpreter is not relevant.)

        :func:`index.fun.run_commit_hook` uses :class:`subprocess.Popen`, including when
        it runs ``bash.exe`` on Windows. It doesn't pass ``shell=True`` (and shouldn't).
        On Windows, `Popen` calls ``CreateProcessW``, which searches several locations
        prior to using the ``PATH`` environment variable. It is expected to search the
        ``System32`` directory, even if another directory containing the executable
        precedes it in ``PATH``. (Other differences are less relevant here.) When WSL is
        installed, even with no distributions, ``bash.exe`` exists in ``System32``, and
        `Popen` finds it even if another ``bash.exe`` precedes it in ``PATH``, as on CI.
        If WSL is absent, ``System32`` may still have ``bash.exe``, as Windows users and
        administrators occasionally put executables there in lieu of extending ``PATH``.
        """
        if os.name != "nt":
            return cls.Inapplicable()

        no_distro_message = "Windows Subsystem for Linux has no installed distributions."

        def error_running_bash(error):
            log.error("Error running bash.exe to check WSL status: %s", error)
            return cls.ErrorWhileChecking(error)

        try:
            # Output rather than forwarding the test command's exit status so that if a
            # failure occurs before we even get to this point, we will detect it. For
            # information on ways to check for WSL, see https://superuser.com/a/1749811.
            script = 'test -e /proc/sys/fs/binfmt_misc/WSLInterop; echo "$?"'
            command = ["bash.exe", "-c", script]
            proc = subprocess.run(command, capture_output=True, check=True, text=True)
        except FileNotFoundError:
            return cls.Absent()
        except OSError as error:
            return error_running_bash(error)
        except subprocess.CalledProcessError as error:
            if error.returncode == 1 and error.stdout.startswith(no_distro_message):
                return cls.WslNoDistro()
            return error_running_bash(error)

        status = proc.stdout.rstrip()
        if status == "0":
            return cls.Wsl()
        if status == "1":
            return cls.Native()
        log.error("Strange output checking WSL status: %s", proc.stdout)
        return cls.ErrorWhileChecking(proc)


_win_bash_status = _WinBashStatus.check()


def _make_hook(git_dir, name, content, make_exec=True):
    """A helper to create a hook"""
    hp = hook_path(name, git_dir)
    hpd = osp.dirname(hp)
    if not osp.isdir(hpd):
        os.mkdir(hpd)
    with open(hp, "wt") as fp:
        fp.write(HOOKS_SHEBANG + content)
    if make_exec:
        os.chmod(hp, 0o744)
    return hp


class TestIndex(TestBase):
    def __init__(self, *args):
        super().__init__(*args)
        self._reset_progress()

    def _assert_fprogress(self, entries):
        self.assertEqual(len(entries), len(self._fprogress_map))
        for _path, call_count in self._fprogress_map.items():
            self.assertEqual(call_count, 2)
        # END for each item in progress map
        self._reset_progress()

    def _fprogress(self, path, done, item):
        self._fprogress_map.setdefault(path, 0)
        curval = self._fprogress_map[path]
        if curval == 0:
            assert not done
        if curval == 1:
            assert done
        self._fprogress_map[path] = curval + 1

    def _fprogress_add(self, path, done, item):
        """Called as progress func - we keep track of the proper
        call order"""
        assert item is not None
        self._fprogress(path, done, item)

    def _reset_progress(self):
        # Maps paths to the count of calls.
        self._fprogress_map = {}

    def _assert_entries(self, entries):
        for entry in entries:
            assert isinstance(entry, BaseIndexEntry)
            assert not osp.isabs(entry.path)
            assert "\\" not in entry.path
        # END for each entry

    def test_index_file_base(self):
        # Read from file.
        index = IndexFile(self.rorepo, fixture_path("index"))
        assert index.entries
        assert index.version > 0

        # Test entry.
        entry = next(iter(index.entries.values()))
        for attr in (
            "path",
            "ctime",
            "mtime",
            "dev",
            "inode",
            "mode",
            "uid",
            "gid",
            "size",
            "binsha",
            "hexsha",
            "stage",
        ):
            getattr(entry, attr)
        # END for each method

        # Test update.
        entries = index.entries
        assert isinstance(index.update(), IndexFile)
        assert entries is not index.entries

        # Test stage.
        index_merge = IndexFile(self.rorepo, fixture_path("index_merge"))
        self.assertEqual(len(index_merge.entries), 106)
        assert len([e for e in index_merge.entries.values() if e.stage != 0])

        # Write the data - it must match the original.
        tmpfile = tempfile.mktemp()
        index_merge.write(tmpfile)
        with open(tmpfile, "rb") as fp:
            self.assertEqual(fp.read(), fixture("index_merge"))
        os.remove(tmpfile)

    def _cmp_tree_index(self, tree, index):
        # Fail unless both objects contain the same paths and blobs.
        if isinstance(tree, str):
            tree = self.rorepo.commit(tree).tree

        blist = []
        for blob in tree.traverse(predicate=lambda e, d: e.type == "blob", branch_first=False):
            assert (blob.path, 0) in index.entries
            blist.append(blob)
        # END for each blob in tree
        if len(blist) != len(index.entries):
            iset = {k[0] for k in index.entries.keys()}
            bset = {b.path for b in blist}
            raise AssertionError(
                "CMP Failed: Missing entries in index: %s, missing in tree: %s" % (bset - iset, iset - bset)
            )
        # END assertion message

    @with_rw_repo("0.1.6")
    def test_index_lock_handling(self, rw_repo):
        def add_bad_blob():
            rw_repo.index.add([Blob(rw_repo, b"f" * 20, "bad-permissions", "foo")])

        try:
            ## First, fail on purpose adding into index.
            add_bad_blob()
        except Exception as ex:
            msg_py3 = "required argument is not an integer"
            msg_py2 = "cannot convert argument to integer"
            assert msg_py2 in str(ex) or msg_py3 in str(ex)

        ## The second time should not fail due to stray lock file.
        try:
            add_bad_blob()
        except Exception as ex:
            assert "index.lock' could not be obtained" not in str(ex)

    @pytest.mark.xfail(
        os.name == "nt",
        reason=(
            "IndexFile.from_tree is broken on Windows (related to NamedTemporaryFile), see #1630.\n"
            "'git read-tree --index-output=...' fails with 'fatal: unable to write new index file'."
        ),
        raises=GitCommandError,
    )
    @with_rw_repo("0.1.6")
    def test_index_file_from_tree(self, rw_repo):
        common_ancestor_sha = "5117c9c8a4d3af19a9958677e45cda9269de1541"
        cur_sha = "4b43ca7ff72d5f535134241e7c797ddc9c7a3573"
        other_sha = "39f85c4358b7346fee22169da9cad93901ea9eb9"

        # Simple index from tree.
        base_index = IndexFile.from_tree(rw_repo, common_ancestor_sha)
        assert base_index.entries
        self._cmp_tree_index(common_ancestor_sha, base_index)

        # Merge two trees - it's like a fast-forward.
        two_way_index = IndexFile.from_tree(rw_repo, common_ancestor_sha, cur_sha)
        assert two_way_index.entries
        self._cmp_tree_index(cur_sha, two_way_index)

        # Merge three trees - here we have a merge conflict.
        three_way_index = IndexFile.from_tree(rw_repo, common_ancestor_sha, cur_sha, other_sha)
        assert len([e for e in three_way_index.entries.values() if e.stage != 0])

        # ITERATE BLOBS
        merge_required = lambda t: t[0] != 0
        merge_blobs = list(three_way_index.iter_blobs(merge_required))
        assert merge_blobs
        assert merge_blobs[0][0] in (1, 2, 3)
        assert isinstance(merge_blobs[0][1], Blob)

        # Test BlobFilter.
        prefix = "lib/git"
        for _stage, blob in base_index.iter_blobs(BlobFilter([prefix])):
            assert blob.path.startswith(prefix)

        # Writing a tree should fail with an unmerged index.
        self.assertRaises(UnmergedEntriesError, three_way_index.write_tree)

        # Removed unmerged entries.
        unmerged_blob_map = three_way_index.unmerged_blobs()
        assert unmerged_blob_map

        # Pick the first blob at the first stage we find and use it as resolved version.
        three_way_index.resolve_blobs(line[0][1] for line in unmerged_blob_map.values())
        tree = three_way_index.write_tree()
        assert isinstance(tree, Tree)
        num_blobs = 0
        for blob in tree.traverse(predicate=lambda item, d: item.type == "blob"):
            assert (blob.path, 0) in three_way_index.entries
            num_blobs += 1
        # END for each blob
        self.assertEqual(num_blobs, len(three_way_index.entries))

    @pytest.mark.xfail(
        os.name == "nt",
        reason=(
            "IndexFile.from_tree is broken on Windows (related to NamedTemporaryFile), see #1630.\n"
            "'git read-tree --index-output=...' fails with 'fatal: unable to write new index file'."
        ),
        raises=GitCommandError,
    )
    @with_rw_repo("0.1.6")
    def test_index_merge_tree(self, rw_repo):
        # A bit out of place, but we need a different repo for this:
        self.assertNotEqual(self.rorepo, rw_repo)
        self.assertEqual(len({self.rorepo, self.rorepo, rw_repo, rw_repo}), 2)

        # SINGLE TREE MERGE
        # Current index is at the (virtual) cur_commit.
        next_commit = "4c39f9da792792d4e73fc3a5effde66576ae128c"
        parent_commit = rw_repo.head.commit.parents[0]
        manifest_key = IndexFile.entry_key("MANIFEST.in", 0)
        manifest_entry = rw_repo.index.entries[manifest_key]
        rw_repo.index.merge_tree(next_commit)
        # Only one change should be recorded.
        assert manifest_entry.binsha != rw_repo.index.entries[manifest_key].binsha

        rw_repo.index.reset(rw_repo.head)
        self.assertEqual(rw_repo.index.entries[manifest_key].binsha, manifest_entry.binsha)

        # FAKE MERGE
        #############
        # Add a change with a NULL sha that should conflict with next_commit. We
        # pretend there was a change, but we do not even bother adding a proper
        # sha for it (which makes things faster of course).
        manifest_fake_entry = BaseIndexEntry((manifest_entry[0], b"\0" * 20, 0, manifest_entry[3]))
        # Try write flag.
        self._assert_entries(rw_repo.index.add([manifest_fake_entry], write=False))
        # Add actually resolves the null-hex-sha for us as a feature, but we can
        # edit the index manually.
        assert rw_repo.index.entries[manifest_key].binsha != Object.NULL_BIN_SHA
        # We must operate on the same index for this! It's a bit problematic as
        # it might confuse people.
        index = rw_repo.index
        index.entries[manifest_key] = IndexEntry.from_base(manifest_fake_entry)
        index.write()
        self.assertEqual(rw_repo.index.entries[manifest_key].hexsha, Diff.NULL_HEX_SHA)

        # Write an unchanged index (just for the fun of it).
        rw_repo.index.write()

        # A three way merge would result in a conflict and fails as the command will
        # not overwrite any entries in our index and hence leave them unmerged. This is
        # mainly a protection feature as the current index is not yet in a tree.
        self.assertRaises(GitCommandError, index.merge_tree, next_commit, base=parent_commit)

        # The only way to get the merged entries is to safe the current index away into a tree,
        # which is like a temporary commit for us. This fails as well as the NULL sha does not
        # have a corresponding object.
        # NOTE: missing_ok is not a kwarg anymore, missing_ok is always true.
        # self.assertRaises(GitCommandError, index.write_tree)

        # If missing objects are okay, this would work though (they are always okay now).
        # As we can't read back the tree with NULL_SHA, we rather set it to something else.
        index.entries[manifest_key] = IndexEntry(manifest_entry[:1] + (hex_to_bin("f" * 40),) + manifest_entry[2:])
        tree = index.write_tree()

        # Now make a proper three way merge with unmerged entries.
        unmerged_tree = IndexFile.from_tree(rw_repo, parent_commit, tree, next_commit)
        unmerged_blobs = unmerged_tree.unmerged_blobs()
        self.assertEqual(len(unmerged_blobs), 1)
        self.assertEqual(list(unmerged_blobs.keys())[0], manifest_key[0])

    @pytest.mark.xfail(
        os.name == "nt",
        reason=(
            "IndexFile.from_tree is broken on Windows (related to NamedTemporaryFile), see #1630.\n"
            "'git read-tree --index-output=...' fails with 'fatal: unable to write new index file'."
        ),
        raises=GitCommandError,
    )
    @with_rw_repo("0.1.6")
    def test_index_file_diffing(self, rw_repo):
        # Default Index instance points to our index.
        index = IndexFile(rw_repo)
        assert index.path is not None
        assert len(index.entries)

        # Write the file back.
        index.write()

        # Could sha it, or check stats.

        # Test diff.
        # Resetting the head will leave the index in a different state, and the
        # diff will yield a few changes.
        cur_head_commit = rw_repo.head.reference.commit
        rw_repo.head.reset("HEAD~6", index=True, working_tree=False)

        # Diff against same index is 0.
        diff = index.diff()
        self.assertEqual(len(diff), 0)

        # Against HEAD as string, must be the same as it matches index.
        diff = index.diff("HEAD")
        self.assertEqual(len(diff), 0)

        # Against previous head, there must be a difference.
        diff = index.diff(cur_head_commit)
        assert len(diff)

        # We reverse the result.
        adiff = index.diff(str(cur_head_commit), R=True)
        odiff = index.diff(cur_head_commit, R=False)  # Now its not reversed anymore.
        assert adiff != odiff
        self.assertEqual(odiff, diff)  # Both unreversed diffs against HEAD.

        # Against working copy - it's still at cur_commit.
        wdiff = index.diff(None)
        assert wdiff != adiff
        assert wdiff != odiff

        # Against something unusual.
        self.assertRaises(ValueError, index.diff, int)

        # Adjust the index to match an old revision.
        cur_branch = rw_repo.active_branch
        cur_commit = cur_branch.commit
        rev_head_parent = "HEAD~1"
        assert index.reset(rev_head_parent) is index

        self.assertEqual(cur_branch, rw_repo.active_branch)
        self.assertEqual(cur_commit, rw_repo.head.commit)

        # There must be differences towards the working tree which is in the 'future'.
        assert index.diff(None)

        # Reset the working copy as well to current head, to pull 'back' as well.
        new_data = b"will be reverted"
        file_path = osp.join(rw_repo.working_tree_dir, "CHANGES")
        with open(file_path, "wb") as fp:
            fp.write(new_data)
        index.reset(rev_head_parent, working_tree=True)
        assert not index.diff(None)
        self.assertEqual(cur_branch, rw_repo.active_branch)
        self.assertEqual(cur_commit, rw_repo.head.commit)
        with open(file_path, "rb") as fp:
            assert fp.read() != new_data

        # Test full checkout.
        test_file = osp.join(rw_repo.working_tree_dir, "CHANGES")
        with open(test_file, "ab") as fd:
            fd.write(b"some data")
        rval = index.checkout(None, force=True, fprogress=self._fprogress)
        assert "CHANGES" in list(rval)
        self._assert_fprogress([None])
        assert osp.isfile(test_file)

        os.remove(test_file)
        rval = index.checkout(None, force=False, fprogress=self._fprogress)
        assert "CHANGES" in list(rval)
        self._assert_fprogress([None])
        assert osp.isfile(test_file)

        # Individual file.
        os.remove(test_file)
        rval = index.checkout(test_file, fprogress=self._fprogress)
        self.assertEqual(list(rval)[0], "CHANGES")
        self._assert_fprogress([test_file])
        assert osp.exists(test_file)

        # Checking out non-existing file throws.
        self.assertRaises(CheckoutError, index.checkout, "doesnt_exist_ever.txt.that")
        self.assertRaises(CheckoutError, index.checkout, paths=["doesnt/exist"])

        # Check out file with modifications.
        append_data = b"hello"
        with open(test_file, "ab") as fp:
            fp.write(append_data)
        try:
            index.checkout(test_file)
        except CheckoutError as e:
            # Detailed exceptions are only possible in older git versions.
            if rw_repo.git._version_info < (2, 29):
                self.assertEqual(len(e.failed_files), 1)
                self.assertEqual(e.failed_files[0], osp.basename(test_file))
                self.assertEqual(len(e.failed_files), len(e.failed_reasons))
                self.assertIsInstance(e.failed_reasons[0], str)
                self.assertEqual(len(e.valid_files), 0)
                with open(test_file, "rb") as fd:
                    s = fd.read()
                self.assertTrue(s.endswith(append_data), s)
        else:
            raise AssertionError("Exception CheckoutError not thrown")

        # If we force it, it should work.
        index.checkout(test_file, force=True)
        assert not open(test_file, "rb").read().endswith(append_data)

        # Check out directory.
        rmtree(osp.join(rw_repo.working_tree_dir, "lib"))
        rval = index.checkout("lib")
        assert len(list(rval)) > 1

    def _count_existing(self, repo, files):
        """Return count of files that actually exist in the repository directory."""
        existing = 0
        basedir = repo.working_tree_dir
        for f in files:
            existing += osp.isfile(osp.join(basedir, f))
        # END for each deleted file
        return existing

    # END num existing helper

    @pytest.mark.xfail(
        os.name == "nt",
        reason=(
            "IndexFile.from_tree is broken on Windows (related to NamedTemporaryFile), see #1630.\n"
            "'git read-tree --index-output=...' fails with 'fatal: unable to write new index file'."
        ),
        raises=GitCommandError,
    )
    @with_rw_repo("0.1.6")
    def test_index_mutation(self, rw_repo):
        index = rw_repo.index
        num_entries = len(index.entries)
        cur_head = rw_repo.head

        uname = "Thomas Müller"
        umail = "sd@company.com"
        with rw_repo.config_writer() as writer:
            writer.set_value("user", "name", uname)
            writer.set_value("user", "email", umail)
        self.assertEqual(writer.get_value("user", "name"), uname)

        # Remove all of the files, provide a wild mix of paths, BaseIndexEntries,
        # IndexEntries.
        def mixed_iterator():
            count = 0
            for entry in index.entries.values():
                type_id = count % 4
                if type_id == 0:  # path
                    yield entry.path
                elif type_id == 1:  # blob
                    yield Blob(rw_repo, entry.binsha, entry.mode, entry.path)
                elif type_id == 2:  # BaseIndexEntry
                    yield BaseIndexEntry(entry[:4])
                elif type_id == 3:  # IndexEntry
                    yield entry
                else:
                    raise AssertionError("Invalid Type")
                count += 1
            # END for each entry

        # END mixed iterator
        deleted_files = index.remove(mixed_iterator(), working_tree=False)
        assert deleted_files
        self.assertEqual(self._count_existing(rw_repo, deleted_files), len(deleted_files))
        self.assertEqual(len(index.entries), 0)

        # Reset the index to undo our changes.
        index.reset()
        self.assertEqual(len(index.entries), num_entries)

        # Remove with working copy.
        deleted_files = index.remove(mixed_iterator(), working_tree=True)
        assert deleted_files
        self.assertEqual(self._count_existing(rw_repo, deleted_files), 0)

        # Reset everything.
        index.reset(working_tree=True)
        self.assertEqual(self._count_existing(rw_repo, deleted_files), len(deleted_files))

        # Invalid type.
        self.assertRaises(TypeError, index.remove, [1])

        # Absolute path.
        deleted_files = index.remove([osp.join(rw_repo.working_tree_dir, "lib")], r=True)
        assert len(deleted_files) > 1
        self.assertRaises(ValueError, index.remove, ["/doesnt/exists"])

        # TEST COMMITTING
        # Commit changed index.
        cur_commit = cur_head.commit
        commit_message = "commit default head by Frèderic Çaufl€"

        new_commit = index.commit(commit_message, head=False)
        assert cur_commit != new_commit
        self.assertEqual(new_commit.author.name, uname)
        self.assertEqual(new_commit.author.email, umail)
        self.assertEqual(new_commit.committer.name, uname)
        self.assertEqual(new_commit.committer.email, umail)
        self.assertEqual(new_commit.message, commit_message)
        self.assertEqual(new_commit.parents[0], cur_commit)
        self.assertEqual(len(new_commit.parents), 1)
        self.assertEqual(cur_head.commit, cur_commit)

        # Commit with other actor.
        cur_commit = cur_head.commit

        my_author = Actor("Frèderic Çaufl€", "author@example.com")
        my_committer = Actor("Committing Frèderic Çaufl€", "committer@example.com")
        commit_actor = index.commit(commit_message, author=my_author, committer=my_committer)
        assert cur_commit != commit_actor
        self.assertEqual(commit_actor.author.name, "Frèderic Çaufl€")
        self.assertEqual(commit_actor.author.email, "author@example.com")
        self.assertEqual(commit_actor.committer.name, "Committing Frèderic Çaufl€")
        self.assertEqual(commit_actor.committer.email, "committer@example.com")
        self.assertEqual(commit_actor.message, commit_message)
        self.assertEqual(commit_actor.parents[0], cur_commit)
        self.assertEqual(len(new_commit.parents), 1)
        self.assertEqual(cur_head.commit, commit_actor)
        self.assertEqual(cur_head.log()[-1].actor, my_committer)

        # Commit with author_date and commit_date.
        cur_commit = cur_head.commit
        commit_message = "commit with dates by Avinash Sajjanshetty"

        new_commit = index.commit(
            commit_message,
            author_date="2006-04-07T22:13:13",
            commit_date="2005-04-07T22:13:13",
        )
        assert cur_commit != new_commit
        print(new_commit.authored_date, new_commit.committed_date)
        self.assertEqual(new_commit.message, commit_message)
        self.assertEqual(new_commit.authored_date, 1144447993)
        self.assertEqual(new_commit.committed_date, 1112911993)

        # Same index, no parents.
        commit_message = "index without parents"
        commit_no_parents = index.commit(commit_message, parent_commits=[], head=True)
        self.assertEqual(commit_no_parents.message, commit_message)
        self.assertEqual(len(commit_no_parents.parents), 0)
        self.assertEqual(cur_head.commit, commit_no_parents)

        # same index, multiple parents.
        commit_message = "Index with multiple parents\n    commit with another line"
        commit_multi_parent = index.commit(commit_message, parent_commits=(commit_no_parents, new_commit))
        self.assertEqual(commit_multi_parent.message, commit_message)
        self.assertEqual(len(commit_multi_parent.parents), 2)
        self.assertEqual(commit_multi_parent.parents[0], commit_no_parents)
        self.assertEqual(commit_multi_parent.parents[1], new_commit)
        self.assertEqual(cur_head.commit, commit_multi_parent)

        # Re-add all files in lib.
        # Get the lib folder back on disk, but get an index without it.
        index.reset(new_commit.parents[0], working_tree=True).reset(new_commit, working_tree=False)
        lib_file_path = osp.join("lib", "git", "__init__.py")
        assert (lib_file_path, 0) not in index.entries
        assert osp.isfile(osp.join(rw_repo.working_tree_dir, lib_file_path))

        # Directory.
        entries = index.add(["lib"], fprogress=self._fprogress_add)
        self._assert_entries(entries)
        self._assert_fprogress(entries)
        assert len(entries) > 1

        # Glob.
        entries = index.reset(new_commit).add([osp.join("lib", "git", "*.py")], fprogress=self._fprogress_add)
        self._assert_entries(entries)
        self._assert_fprogress(entries)
        self.assertEqual(len(entries), 14)

        # Same file.
        entries = index.reset(new_commit).add(
            [osp.join(rw_repo.working_tree_dir, "lib", "git", "head.py")] * 2,
            fprogress=self._fprogress_add,
        )
        self._assert_entries(entries)
        self.assertEqual(entries[0].mode & 0o644, 0o644)
        # Would fail, test is too primitive to handle this case.
        # self._assert_fprogress(entries)
        self._reset_progress()
        self.assertEqual(len(entries), 2)

        # Missing path.
        self.assertRaises(OSError, index.reset(new_commit).add, ["doesnt/exist/must/raise"])

        # Blob from older revision overrides current index revision.
        old_blob = new_commit.parents[0].tree.blobs[0]
        entries = index.reset(new_commit).add([old_blob], fprogress=self._fprogress_add)
        self._assert_entries(entries)
        self._assert_fprogress(entries)
        self.assertEqual(index.entries[(old_blob.path, 0)].hexsha, old_blob.hexsha)
        self.assertEqual(len(entries), 1)

        # Mode 0 not allowed.
        null_hex_sha = Diff.NULL_HEX_SHA
        null_bin_sha = b"\0" * 20
        self.assertRaises(
            ValueError,
            index.reset(new_commit).add,
            [BaseIndexEntry((0, null_bin_sha, 0, "doesntmatter"))],
        )

        # Add new file.
        new_file_relapath = "my_new_file"
        self._make_file(new_file_relapath, "hello world", rw_repo)
        entries = index.reset(new_commit).add(
            [BaseIndexEntry((0o10644, null_bin_sha, 0, new_file_relapath))],
            fprogress=self._fprogress_add,
        )
        self._assert_entries(entries)
        self._assert_fprogress(entries)
        self.assertEqual(len(entries), 1)
        self.assertNotEqual(entries[0].hexsha, null_hex_sha)

        # Add symlink.
        if os.name != "nt":
            for target in ("/etc/nonexisting", "/etc/passwd", "/etc"):
                basename = "my_real_symlink"

                link_file = osp.join(rw_repo.working_tree_dir, basename)
                os.symlink(target, link_file)
                entries = index.reset(new_commit).add([link_file], fprogress=self._fprogress_add)
                self._assert_entries(entries)
                self._assert_fprogress(entries)
                self.assertEqual(len(entries), 1)
                self.assertTrue(S_ISLNK(entries[0].mode))
                self.assertTrue(S_ISLNK(index.entries[index.entry_key("my_real_symlink", 0)].mode))

                # We expect only the target to be written.
                self.assertEqual(
                    index.repo.odb.stream(entries[0].binsha).read().decode("ascii"),
                    target,
                )

                os.remove(link_file)
            # END for each target
        # END real symlink test

        # Add fake symlink and assure it checks-our as symlink.
        fake_symlink_relapath = "my_fake_symlink"
        link_target = "/etc/that"
        fake_symlink_path = self._make_file(fake_symlink_relapath, link_target, rw_repo)
        fake_entry = BaseIndexEntry((0o120000, null_bin_sha, 0, fake_symlink_relapath))
        entries = index.reset(new_commit).add([fake_entry], fprogress=self._fprogress_add)
        self._assert_entries(entries)
        self._assert_fprogress(entries)
        assert entries[0].hexsha != null_hex_sha
        self.assertEqual(len(entries), 1)
        self.assertTrue(S_ISLNK(entries[0].mode))

        # Check that this also works with an alternate method.
        full_index_entry = IndexEntry.from_base(BaseIndexEntry((0o120000, entries[0].binsha, 0, entries[0].path)))
        entry_key = index.entry_key(full_index_entry)
        index.reset(new_commit)

        assert entry_key not in index.entries
        index.entries[entry_key] = full_index_entry
        index.write()
        index.update()  # Force reread of entries.
        new_entry = index.entries[entry_key]
        assert S_ISLNK(new_entry.mode)

        # A tree created from this should contain the symlink.
        tree = index.write_tree()
        assert fake_symlink_relapath in tree
        index.write()  # Flush our changes for the checkout.

        # Check out the fake link, should be a link then.
        assert not S_ISLNK(os.stat(fake_symlink_path)[ST_MODE])
        os.remove(fake_symlink_path)
        index.checkout(fake_symlink_path)

        # On Windows, we will never get symlinks.
        if os.name == "nt":
            # Symlinks should contain the link as text (which is what a
            # symlink actually is).
            with open(fake_symlink_path, "rt") as fd:
                self.assertEqual(fd.read(), link_target)
        else:
            self.assertTrue(S_ISLNK(os.lstat(fake_symlink_path)[ST_MODE]))

        # TEST RENAMING
        def assert_mv_rval(rval):
            for source, dest in rval:
                assert not osp.exists(source) and osp.exists(dest)
            # END for each renamed item

        # END move assertion utility

        self.assertRaises(ValueError, index.move, ["just_one_path"])
        # Try to move a file onto an existing file.
        files = ["AUTHORS", "LICENSE"]
        self.assertRaises(GitCommandError, index.move, files)

        # Again, with force.
        assert_mv_rval(index.move(files, f=True))

        # Move files into a directory - dry run.
        paths = ["LICENSE", "VERSION", "doc"]
        rval = index.move(paths, dry_run=True)
        self.assertEqual(len(rval), 2)
        assert osp.exists(paths[0])

        # Again, no dry run.
        rval = index.move(paths)
        assert_mv_rval(rval)

        # Move dir into dir.
        rval = index.move(["doc", "test"])
        assert_mv_rval(rval)

        # TEST PATH REWRITING
        ######################
        count = [0]

        def rewriter(entry):
            rval = str(count[0])
            count[0] += 1
            return rval

        # END rewriter

        def make_paths():
            """Help out the test by yielding two existing paths and one new path."""
            yield "CHANGES"
            yield "ez_setup.py"
            yield index.entries[index.entry_key("README", 0)]
            yield index.entries[index.entry_key(".gitignore", 0)]

            for fid in range(3):
                fname = "newfile%i" % fid
                with open(fname, "wb") as fd:
                    fd.write(b"abcd")
                yield Blob(rw_repo, Blob.NULL_BIN_SHA, 0o100644, fname)
            # END for each new file

        # END path producer
        paths = list(make_paths())
        self._assert_entries(index.add(paths, path_rewriter=rewriter))

        for filenum in range(len(paths)):
            assert index.entry_key(str(filenum), 0) in index.entries

        # TEST RESET ON PATHS
        ######################
        arela = "aa"
        brela = "bb"
        afile = self._make_file(arela, "adata", rw_repo)
        bfile = self._make_file(brela, "bdata", rw_repo)
        akey = index.entry_key(arela, 0)
        bkey = index.entry_key(brela, 0)
        keys = (akey, bkey)
        absfiles = (afile, bfile)
        files = (arela, brela)

        for fkey in keys:
            assert fkey not in index.entries

        index.add(files, write=True)
        nc = index.commit("2 files committed", head=False)

        for fkey in keys:
            assert fkey in index.entries

        # Just the index.
        index.reset(paths=(arela, afile))
        assert akey not in index.entries
        assert bkey in index.entries

        # Now with working tree - files on disk as well as entries must be recreated.
        rw_repo.head.commit = nc
        for absfile in absfiles:
            os.remove(absfile)

        index.reset(working_tree=True, paths=files)

        for fkey in keys:
            assert fkey in index.entries
        for absfile in absfiles:
            assert osp.isfile(absfile)

    @pytest.mark.xfail(
        os.name == "nt",
        reason=(
            "IndexFile.from_tree is broken on Windows (related to NamedTemporaryFile), see #1630.\n"
            "'git read-tree --index-output=...' fails with 'fatal: unable to write new index file'."
        ),
        raises=GitCommandError,
    )
    @with_rw_repo("HEAD")
    def test_compare_write_tree(self, rw_repo):
        """Test writing all trees, comparing them for equality."""
        # It's important to have a few submodules in there too.
        max_count = 25
        count = 0
        for commit in rw_repo.head.commit.traverse():
            if count >= max_count:
                break
            count += 1
            index = rw_repo.index.reset(commit)
            orig_tree = commit.tree
            self.assertEqual(index.write_tree(), orig_tree)
        # END for each commit

    @with_rw_repo("HEAD", bare=False)
    def test_index_single_addremove(self, rw_repo):
        fp = osp.join(rw_repo.working_dir, "testfile.txt")
        with open(fp, "w") as fs:
            fs.write("content of testfile")
        self._assert_entries(rw_repo.index.add(fp))
        deleted_files = rw_repo.index.remove(fp)
        assert deleted_files

    def test_index_new(self):
        B = self.rorepo.tree("6d9b1f4f9fa8c9f030e3207e7deacc5d5f8bba4e")
        H = self.rorepo.tree("25dca42bac17d511b7e2ebdd9d1d679e7626db5f")
        M = self.rorepo.tree("e746f96bcc29238b79118123028ca170adc4ff0f")

        for args in ((B,), (B, H), (B, H, M)):
            index = IndexFile.new(self.rorepo, *args)
            assert isinstance(index, IndexFile)
        # END for each arg tuple

    @with_rw_repo("HEAD", bare=True)
    def test_index_bare_add(self, rw_bare_repo):
        # Something is wrong after cloning to a bare repo, reading the
        # property rw_bare_repo.working_tree_dir will return '/tmp'
        # instead of throwing the Exception we are expecting. This is
        # a quick hack to make this test fail when expected.
        assert rw_bare_repo.working_tree_dir is None
        assert rw_bare_repo.bare
        contents = b"This is a BytesIO file"
        filesize = len(contents)
        fileobj = BytesIO(contents)
        filename = "my-imaginary-file"
        istream = rw_bare_repo.odb.store(IStream(Blob.type, filesize, fileobj))
        entry = BaseIndexEntry((0o100644, istream.binsha, 0, filename))
        try:
            rw_bare_repo.index.add([entry])
        except AssertionError:
            self.fail("Adding to the index of a bare repo is not allowed.")

        # Adding using a path should still require a non-bare repository.
        asserted = False
        path = osp.join("git", "test", "test_index.py")
        try:
            rw_bare_repo.index.add([path])
        except InvalidGitRepositoryError:
            asserted = True
        assert asserted, "Adding using a filename is not correctly asserted."

    @with_rw_directory
    def test_add_utf8P_path(self, rw_dir):
        # NOTE: fp is not a Unicode object in Python 2 (which is the source of the problem).
        fp = osp.join(rw_dir, "ø.txt")
        with open(fp, "wb") as fs:
            fs.write("content of ø".encode("utf-8"))

        r = Repo.init(rw_dir)
        r.index.add([fp])
        r.index.commit("Added orig and prestable")

    @with_rw_directory
    def test_add_a_file_with_wildcard_chars(self, rw_dir):
        # See issue #407.
        fp = osp.join(rw_dir, "[.exe")
        with open(fp, "wb") as f:
            f.write(b"something")

        r = Repo.init(rw_dir)
        r.index.add([fp])
        r.index.commit("Added [.exe")

    def test__to_relative_path_at_root(self):
        root = osp.abspath(os.sep)

        class Mocked:
            bare = False
            git_dir = root
            working_tree_dir = root

        repo = Mocked()
        path = os.path.join(root, "file")
        index = IndexFile(repo)

        rel = index._to_relative_path(path)
        self.assertEqual(rel, os.path.relpath(path, root))

    @pytest.mark.xfail(
        type(_win_bash_status) is _WinBashStatus.WslNoDistro,