iframe-proxy | Sunbelt Computer Software

960 lines (813 loc) · 36.6 KB
# -*- coding: utf-8 -*-
# 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 BSD License: https://opensource.org/license/bsd-3-clause/
from io import BytesIO
import os.path as osp
from pathlib import Path
from stat import S_ISLNK, ST_MODE
import shutil
import tempfile
import pytest
from git import (
    IndexFile,
    BlobFilter,
    UnmergedEntriesError,
    Object,
    GitCommandError,
    CheckoutError,
from git.compat import is_win
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"
def _found_in(cmd, directory):
    """Check if a command is resolved in a directory (without following symlinks)."""
    path = shutil.which(cmd)
    return path and Path(path).parent == Path(directory)
is_win_without_bash = is_win and not shutil.which("bash.exe")
is_win_with_wsl_bash = is_win and _found_in(
    cmd="bash.exe",
    directory=Path(os.getenv("WINDIR")) / "System32",
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(TestIndex, self).__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:
            ## 1st 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)
        ## 2nd 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)
    @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 - its 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))
    @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
        # must operate on the same index for this ! Its 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 deos 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])
    @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 - its 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"])
        # checkout 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)
        # checkout 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):
        """
        Returns 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
    @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 not is_win:
            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))
        # assure 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
        # checkout the fakelink, 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 is_win:
            # simlinks 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"])
        # file onto existing file
        files = ["AUTHORS", "LICENSE"]
        self.assertRaises(GitCommandError, index.move, files)
        # again, with force
        assert_mv_rval(index.move(files, f=True))
        # files into 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)
        # 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():
            # two existing ones, one new one
            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)
    @with_rw_repo("HEAD")
    def test_compare_write_tree(self, rw_repo):
        # write all trees and compare them
        # its 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(object):
            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))
    @with_rw_repo("HEAD", bare=True)
    def test_pre_commit_hook_success(self, rw_repo):
        index = rw_repo.index
        _make_hook(index.repo.git_dir, "pre-commit", "exit 0")
        index.commit("This should not fail")
    @with_rw_repo("HEAD", bare=True)
    def test_pre_commit_hook_fail(self, rw_repo):
        index = rw_repo.index
        hp = _make_hook(index.repo.git_dir, "pre-commit", "echo stdout; echo stderr 1>&2; exit 1")
        try:
            index.commit("This should fail")
        except HookExecutionError as err:
            if is_win_without_bash:
                self.assertIsInstance(err.status, OSError)
                self.assertEqual(err.command, [hp])
                self.assertEqual(err.stdout, "")
                self.assertEqual(err.stderr, "")
                assert str(err)
            else:
                self.assertEqual(err.status, 1)
                self.assertEqual(err.command, [hp])
                self.assertEqual(err.stdout, "\n  stdout: 'stdout\n'")
                self.assertEqual(err.stderr, "\n  stderr: 'stderr\n'")
                assert str(err)
        else:
            raise AssertionError("Should have caught a HookExecutionError")
    @pytest.mark.xfail(
        is_win_without_bash or is_win_with_wsl_bash,
        reason="Specifically seems to fail on WSL bash (in spite of #1399)",
        raises=AssertionError,
    @with_rw_repo("HEAD", bare=True)
    def test_commit_msg_hook_success(self, rw_repo):
        commit_message = "commit default head by Frèderic Çaufl€"
        from_hook_message = "from commit-msg"
        index = rw_repo.index
        _make_hook(
            index.repo.git_dir,
            "commit-msg",
            'printf " {}" >> "$1"'.format(from_hook_message),
        new_commit = index.commit(commit_message)
        self.assertEqual(new_commit.message, "{} {}".format(commit_message, from_hook_message))
    @with_rw_repo("HEAD", bare=True)
    def test_commit_msg_hook_fail(self, rw_repo):
        index = rw_repo.index
        hp = _make_hook(index.repo.git_dir, "commit-msg", "echo stdout; echo stderr 1>&2; exit 1")
        try:
            index.commit("This should fail")
        except HookExecutionError as err:
            if is_win_without_bash:
                self.assertIsInstance(err.status, OSError)
                self.assertEqual(err.command, [hp])
                self.assertEqual(err.stdout, "")
                self.assertEqual(err.stderr, "")
                assert str(err)
            else:
                self.assertEqual(err.status, 1)
                self.assertEqual(err.command, [hp])
                self.assertEqual(err.stdout, "\n  stdout: 'stdout\n'")
                self.assertEqual(err.stderr, "\n  stderr: 'stderr\n'")
                assert str(err)
        else:
            raise AssertionError("Should have caught a HookExecutionError")
    @with_rw_repo("HEAD")
    def test_index_add_pathlike(self, rw_repo):
        git_dir = Path(rw_repo.git_dir)
        file = git_dir / "file.txt"
        file.touch()
        rw_repo.index.add(file)
Sunbelt Computer Software

PL/B Language Development and Support

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

FilesExpand file tree

test_index.py

Latest commit

History

test_index.py

File metadata and controls
