# 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: http://www.opensource.org/licenses/bsd-license.php from git.test.lib import * from git import * import inspect import os import sys import tempfile import glob import shutil from stat import * class TestIndex(TestBase): def __init__(self, *args): super(TestIndex, self).__init__(*args) self._reset_progress() def _assert_fprogress(self, entries): assert len(entries) == len(self._fprogress_map) for path, call_count in self._fprogress_map.iteritems(): assert 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 = dict() def _assert_entries(self, entries): for entry in entries: assert isinstance(entry, BaseIndexEntry) assert not os.path.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 last_val = None entry = index.entries.itervalues().next() for attr in ("path","ctime","mtime","dev","inode","mode","uid", "gid","size","binsha", "hexsha", "stage"): val = 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")) assert len(index_merge.entries) == 106 assert len(list(e for e in index_merge.entries.itervalues() if e.stage != 0 )) # write the data - it must match the original tmpfile = tempfile.mktemp() index_merge.write(tmpfile) fp = open(tmpfile, 'rb') assert fp.read() == fixture("index_merge") fp.close() 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 num_blobs = 0 blist = list() 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 = set(k[0] for k in index.entries.keys()) bset = set(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_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(list(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.failUnlessRaises(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( l[0][1] for l in unmerged_blob_map.itervalues() ) 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 assert 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: assert self.rorepo != rw_repo and not (self.rorepo == rw_repo) assert len(set((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) assert 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], "\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() assert 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.failUnlessRaises(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.failUnlessRaises(GitCommandError, index.write_tree) # if missing objects are okay, this would work though ( they are always okay now ) 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() assert len(unmerged_blobs) == 1 and 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 ref = rw_repo.head.reset('HEAD~6', index=True, working_tree=False) # diff against same index is 0 diff = index.diff() assert len(diff) == 0 # against HEAD as string, must be the same as it matches index diff = index.diff('HEAD') assert 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 assert 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.failUnlessRaises(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 assert cur_branch == rw_repo.active_branch assert 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 = "will be reverted" file_path = os.path.join(rw_repo.working_tree_dir, "CHANGES") fp = open(file_path, "wb") fp.write(new_data) fp.close() index.reset(rev_head_parent, working_tree=True) assert not index.diff(None) assert cur_branch == rw_repo.active_branch assert cur_commit == rw_repo.head.commit fp = open(file_path,'rb') try: assert fp.read() != new_data finally: fp.close() # test full checkout test_file = os.path.join(rw_repo.working_tree_dir, "CHANGES") open(test_file, 'ab').write("some data") rval = index.checkout(None, force=True, fprogress=self._fprogress) assert 'CHANGES' in list(rval) self._assert_fprogress([None]) assert os.path.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 os.path.isfile(test_file) # individual file os.remove(test_file) rval = index.checkout(test_file, fprogress=self._fprogress) assert list(rval)[0] == 'CHANGES' self._assert_fprogress([test_file]) assert os.path.exists(test_file) # checking out non-existing file throws self.failUnlessRaises(CheckoutError, index.checkout, "doesnt_exist_ever.txt.that") self.failUnlessRaises(CheckoutError, index.checkout, paths=["doesnt/exist"]) # checkout file with modifications append_data = "hello" fp = open(test_file, "ab") fp.write(append_data) fp.close() try: index.checkout(test_file) except CheckoutError, e: assert len(e.failed_files) == 1 and e.failed_files[0] == os.path.basename(test_file) assert (len(e.failed_files) == len(e.failed_reasons)) and isinstance(e.failed_reasons[0], basestring) assert len(e.valid_files) == 0 assert open(test_file).read().endswith(append_data) 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).read().endswith(append_data) # checkout directory shutil.rmtree(os.path.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 += os.path.isfile(os.path.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 = "Some Developer" umail = "sd@company.com" rw_repo.config_writer().set_value("user", "name", uname) rw_repo.config_writer().set_value("user", "email", umail) # remove all of the files, provide a wild mix of paths, BaseIndexEntries, # IndexEntries def mixed_iterator(): count = 0 for entry in index.entries.itervalues(): 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 assert self._count_existing(rw_repo, deleted_files) == len(deleted_files) assert len(index.entries) == 0 # reset the index to undo our changes index.reset() assert len(index.entries) == num_entries # remove with working copy deleted_files = index.remove(mixed_iterator(), working_tree=True) assert deleted_files assert self._count_existing(rw_repo, deleted_files) == 0 # reset everything index.reset(working_tree=True) assert self._count_existing(rw_repo, deleted_files) == len(deleted_files) # invalid type self.failUnlessRaises(TypeError, index.remove, [1]) # absolute path deleted_files = index.remove([os.path.join(rw_repo.working_tree_dir,"lib")], r=True) assert len(deleted_files) > 1 self.failUnlessRaises(ValueError, index.remove, ["/doesnt/exists"]) # TEST COMMITTING # commit changed index cur_commit = cur_head.commit commit_message = "commit default head" new_commit = index.commit(commit_message, head=False) assert cur_commit != new_commit assert new_commit.author.name == uname assert new_commit.author.email == umail assert new_commit.committer.name == uname assert new_commit.committer.email == umail assert new_commit.message == commit_message assert new_commit.parents[0] == cur_commit assert len(new_commit.parents) == 1 assert cur_head.commit == cur_commit # same index, no parents commit_message = "index without parents" commit_no_parents = index.commit(commit_message, parent_commits=list(), head=True) assert commit_no_parents.message == commit_message assert len(commit_no_parents.parents) == 0 assert 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)) assert commit_multi_parent.message == commit_message assert len(commit_multi_parent.parents) == 2 assert commit_multi_parent.parents[0] == commit_no_parents assert commit_multi_parent.parents[1] == new_commit assert 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 = os.path.join("lib", "git", "__init__.py") assert (lib_file_path, 0) not in index.entries assert os.path.isfile(os.path.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([os.path.join('lib', 'git', '*.py')], fprogress=self._fprogress_add) self._assert_entries(entries) self._assert_fprogress(entries) assert len(entries) == 14 # same file entries = index.reset(new_commit).add([os.path.abspath(os.path.join('lib', 'git', 'head.py'))]*2, fprogress=self._fprogress_add) self._assert_entries(entries) assert entries[0].mode & 0644 == 0644 # would fail, test is too primitive to handle this case # self._assert_fprogress(entries) self._reset_progress() assert len(entries) == 2 # missing path self.failUnlessRaises(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) assert index.entries[(old_blob.path,0)].hexsha == old_blob.hexsha and len(entries) == 1 # mode 0 not allowed null_hex_sha = Diff.NULL_HEX_SHA null_bin_sha = "\0" * 20 self.failUnlessRaises(ValueError, index.reset(new_commit).add, [BaseIndexEntry((0, null_bin_sha,0,"doesntmatter"))]) # add new file new_file_relapath = "my_new_file" new_file_path = self._make_file(new_file_relapath, "hello world", rw_repo) entries = index.reset(new_commit).add([BaseIndexEntry((010644, null_bin_sha, 0, new_file_relapath))], fprogress=self._fprogress_add) self._assert_entries(entries) self._assert_fprogress(entries) assert len(entries) == 1 and entries[0].hexsha != null_hex_sha # add symlink if sys.platform != "win32": basename = "my_real_symlink" target = "/etc/that" link_file = os.path.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) assert len(entries) == 1 and S_ISLNK(entries[0].mode) assert S_ISLNK(index.entries[index.entry_key("my_real_symlink", 0)].mode) # we expect only the target to be written assert index.repo.odb.stream(entries[0].binsha).read() == 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((0120000, 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 assert len(entries) == 1 and S_ISLNK(entries[0].mode) # assure this also works with an alternate method full_index_entry = IndexEntry.from_base(BaseIndexEntry((0120000, 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 os.name == 'nt': # simlinks should contain the link as text ( which is what a # symlink actually is ) open(fake_symlink_path,'rb').read() == link_target else: assert S_ISLNK(os.lstat(fake_symlink_path)[ST_MODE]) # TEST RENAMING def assert_mv_rval(rval): for source, dest in rval: assert not os.path.exists(source) and os.path.exists(dest) # END for each renamed item # END move assertion utility self.failUnlessRaises(ValueError, index.move, ['just_one_path']) # file onto existing file files = ['AUTHORS', 'LICENSE'] self.failUnlessRaises(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) assert len(rval) == 2 assert os.path.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 open(fname, 'wb').write("abcd") yield Blob(rw_repo, Blob.NULL_BIN_SHA, 0100644, 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 not fkey 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 not akey 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 os.path.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 assert index.write_tree() == orig_tree # END for each commit 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