summaryrefslogtreecommitdiff
path: root/git/test/test_index.py
diff options
context:
space:
mode:
authorKonrad Weihmann <kweihmann@outlook.com>2020-07-10 18:41:02 +0200
committerKonrad Weihmann <kweihmann@outlook.com>2020-07-12 15:01:21 +0200
commit0374d7cf84ecd8182b74a639fcfdb9eafddcfd15 (patch)
treef462fb9fc434f197b39213b53f33f4e09fc0329e /git/test/test_index.py
parent9cb7ae8d9721e1269f5bacd6dbc33ecdec4659c0 (diff)
downloadgitpython-0374d7cf84ecd8182b74a639fcfdb9eafddcfd15.tar.gz
tests: move to root dir
This should ensure that tests are NOT packaged into release package by setuptools, as tests are development only + fixtures after moving Signed-off-by: Konrad Weihmann <kweihmann@outlook.com>
Diffstat (limited to 'git/test/test_index.py')
-rw-r--r--git/test/test_index.py934
1 files changed, 0 insertions, 934 deletions
diff --git a/git/test/test_index.py b/git/test/test_index.py
deleted file mode 100644
index ce14537a..00000000
--- a/git/test/test_index.py
+++ /dev/null
@@ -1,934 +0,0 @@
-# -*- 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: http://www.opensource.org/licenses/bsd-license.php
-
-from io import BytesIO
-import os
-from stat import (
- S_ISLNK,
- ST_MODE
-)
-import tempfile
-from unittest import skipIf
-
-from git import (
- IndexFile,
- Repo,
- BlobFilter,
- UnmergedEntriesError,
- Tree,
- Object,
- Diff,
- 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 git.test.lib import (
- TestBase,
- fixture_path,
- fixture,
- with_rw_repo
-)
-from git.test.lib import with_rw_directory
-from git.util import Actor, rmtree
-from git.util import HIDE_WINDOWS_KNOWN_ERRORS, hex_to_bin
-from gitdb.base import IStream
-
-import os.path as osp
-from git.cmd import Git
-
-HOOKS_SHEBANG = "#!/usr/bin/env sh\n"
-
-
-@skipIf(HIDE_WINDOWS_KNOWN_ERRORS, "TODO: fix hooks execution on Windows: #703")
-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:
- 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
-
- @skipIf(HIDE_WINDOWS_KNOWN_ERRORS and Git.is_cygwin(),
- """FIXME: File "C:\\projects\\gitpython\\git\\test\\test_index.py", line 642, in test_index_mutation
- self.assertEqual(fd.read(), link_target)
- AssertionError: '!<symlink>\xff\xfe/\x00e\x00t\x00c\x00/\x00t\x00h\x00a\x00t\x00\x00\x00'
- != '/etc/that'
- """)
- @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:
- 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', 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:
- 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 cought a HookExecutionError")