# 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/
import contextlib
import logging
import os
import os.path as osp
import re
import shutil
import struct
import subprocess
import sys
import tempfile
from dataclasses import dataclass
from hashlib import sha1
from io import BytesIO
from itertools import product
from pathlib import Path
from stat import S_ISLNK, ST_MODE
from unittest import mock
import ddt
import pytest
from gitdb.base import IStream
from git import NULL_TREE, BlobFilter, Diff, Git, IndexFile, Object, Repo, Tree
from git.diff import NULL_TREE_SHA
from git.exc import (
CheckoutError,
GitCommandError,
HookExecutionError,
InvalidGitRepositoryError,
UnmergedEntriesError,
UnsafeOptionError,
)
from git.index.fun import _git_for_windows_bash, _which_from_path, hook_path, read_cache, run_commit_hook, write_cache
from git.index.typ import BaseIndexEntry, IndexEntry
from git.index.util import TemporaryFileSwap
from git.objects import Blob
from git.util import Actor, cwd, hex_to_bin, rmtree
from test.lib import PathLikeMock, TestBase, VirtualEnvironment, fixture, fixture_path, with_rw_directory, with_rw_repo
from test.lib.helper import symlinks_supported, xfail_if_raises
HOOKS_SHEBANG = "#!/usr/bin/env sh\n"
_logger = logging.getLogger(__name__)
def _get_windows_ansi_encoding():
"""Get the encoding specified by the Windows system-wide ANSI active code page."""
# locale.getencoding may work but is only in Python 3.11+. Use the registry instead.
import winreg
hklm_path = R"SYSTEM\CurrentControlSet\Control\Nls\CodePage"
with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, hklm_path) as key:
value, _ = winreg.QueryValueEx(key, "ACP")
return f"cp{value}"
class WinBashStatus:
"""Namespace of native-Windows bash.exe statuses. Affects what hook tests can pass.
Call check() to check the status. (CheckError and WinError should not typically be
used to trigger skip or xfail, because they represent unexpected situations.)
"""
@dataclass
class Inapplicable:
"""This system is not native Windows: either not Windows at all, or Cygwin."""
@dataclass
class Absent:
"""No command for bash.exe is found on the system."""
@dataclass
class Native:
"""Running bash.exe operates outside any WSL distribution (as with Git Bash)."""
@dataclass
class Wsl:
"""Running bash.exe calls bash in a WSL distribution."""
@dataclass
class WslNoDistro:
"""Running bash.exe tries to run bash on a WSL distribution, but none exists."""
process: "subprocess.CompletedProcess[bytes]"
message: str
@dataclass
class CheckError:
"""Running bash.exe fails in an unexpected error or gives unexpected output."""
process: "subprocess.CompletedProcess[bytes]"
message: str
@dataclass
class WinError:
"""bash.exe may exist but can't run. CreateProcessW fails unexpectedly."""
exception: OSError
@classmethod
def check(cls):
"""Check the status of the bash.exe that run_commit_hook will try to use.
This runs a command with bash.exe and checks the result. On Windows, shell and
non-shell executable search differ; shutil.which often finds the wrong bash.exe.
run_commit_hook uses Popen, including to run bash.exe on Windows. It doesn't
pass shell=True (and shouldn't). On Windows, Popen calls CreateProcessW, which
checks some locations before using the PATH environment variable. It is expected
to try System32, even if another directory with the executable precedes it in
PATH. When WSL is present, even with no distributions, bash.exe usually exists
in System32; Popen finds it even if a shell would run another one, as on CI.
(Without WSL, System32 may still have bash.exe; users sometimes put it there.)
"""
if sys.platform != "win32":
return cls.Inapplicable()
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]
process = subprocess.run(command, capture_output=True)
except FileNotFoundError:
return cls.Absent()
except OSError as error:
return cls.WinError(error)
text = cls._decode(process.stdout).rstrip() # stdout includes WSL's own errors.
if process.returncode == 1 and re.search(r"\bhttps://aka.ms/wslstore\b", text):
return cls.WslNoDistro(process, text)
if process.returncode != 0:
_logger.error("Error running bash.exe to check WSL status: %s", text)
return cls.CheckError(process, text)
if text == "0":
return cls.Wsl()
if text == "1":
return cls.Native()
_logger.error("Strange output checking WSL status: %s", text)
return cls.CheckError(process, text)
@staticmethod
def _decode(stdout):
"""Decode bash.exe output as best we can."""
# When bash.exe is the WSL wrapper but the output is from WSL itself rather than
# code running in a distribution, the output is often in UTF-16LE, which Windows
# uses internally. The UTF-16LE representation of a Windows-style line ending is
# rarely seen otherwise, so use it to detect this situation.
if b"\r\0\n\0" in stdout:
return stdout.decode("utf-16le")
# At this point, the output is either blank or probably not UTF-16LE. It's often
# UTF-8 from inside a WSL distro or non-WSL bash shell. Our test command only
# uses the ASCII subset, so we can safely guess a wrong code page for it. Errors
# from such an environment can contain any text, but unlike WSL's own messages,
# they go to stderr, not stdout. So we can try the system ANSI code page first.
acp = _get_windows_ansi_encoding()
try:
return stdout.decode(acp)
except UnicodeDecodeError:
pass
except LookupError as error:
_logger.warning(str(error)) # Message already says "Unknown encoding:".
# Assume UTF-8. If invalid, substitute Unicode replacement characters.
return stdout.decode("utf-8", errors="replace")
_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
def _raw_index(path):
"""Build an index without using the writer under test."""
name = path.encode("utf-8")
entry = struct.pack(">10L20sH", 0, 0, 0, 0, 0, 0, 0o100644, 0, 0, 0, b"a" * 20, min(len(name), 0xFFF)) + name
entry += b"\0" * (8 - len(entry) % 8)
data = b"DIRC" + struct.pack(">LL", 2, 1) + entry
return data + sha1(data).digest()
@ddt.ddt
class TestIndex(TestBase):
@with_rw_repo("HEAD")
def test_checkout_rejects_unsafe_prefix(self, rw_repo):
with tempfile.TemporaryDirectory() as target:
with self.assertRaises(UnsafeOptionError):
rw_repo.index.checkout(prefix=f"{target}/")
rw_repo.index.checkout(prefix=f"{target}/", allow_unsafe_options=True)
self.assertTrue(osp.isfile(osp.join(target, "CHANGES")))
@with_rw_repo("HEAD")
def test_remove_rejects_pathspec_from_file(self, rw_repo):
with tempfile.TemporaryDirectory() as tdir:
pathspecs = Path(tdir) / "pathspecs"
pathspecs.write_bytes(b"unmatched-path-one\nunmatched-path-two")
for option_name in ("pathspec_from_file", "pathspec_from"):
with self.assertRaises(UnsafeOptionError):
rw_repo.index.remove(
[],
pathspec_file_nul=True,
**{option_name: str(pathspecs)},
)
@with_rw_repo("HEAD")
def test_remove_allows_explicit_pathspec_from_file(self, rw_repo):
with tempfile.TemporaryDirectory() as tdir:
pathspecs = Path(tdir) / "pathspecs"
pathspecs.write_bytes(b"CHANGES\0")
removed = rw_repo.index.remove(
[],
pathspec_from_file=str(pathspecs),
pathspec_file_nul=True,
allow_unsafe_options=True,
)
assert "CHANGES" in removed
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)
@ddt.data(
"",
".",
"..",
"../outside",
"a/../outside",
"/absolute",
"C:relative",
"a//b",
"a/./b",
"a/",
"nul\0name",
".git/config",
"a/.GiT/hooks/hook",
"git~1/config",
".git. /config",
".git:stream",
".g\u200cit/config",
"a\\.git\\config",
)
def test_index_reader_and_writer_reject_unsafe_paths(self, path):
with pytest.raises(ValueError):
read_cache(BytesIO(_raw_index(path)))
entry = IndexEntry((0o100644, b"a" * 20, 0, path))
with pytest.raises(ValueError):
write_cache([entry], BytesIO())
def test_valid_unusual_index_names_round_trip(self):
names = ["a b", "a\nb", "a\tb", "name:value", "dir/.gitignore", "caf"]
if os.name != "nt":
names.append("a\\b")
entries = [IndexEntry((0o100644, b"a" * 20, 0, name)) for name in sorted(names)]
stream = BytesIO()
write_cache(entries, stream)
stream.seek(0)
assert [entry.path for entry in read_cache(stream)[1].values()] == sorted(names)
def test_long_index_names_are_fully_validated(self):
prefix = "a/" + "nested/" * 650
with pytest.raises(ValueError):
read_cache(BytesIO(_raw_index(prefix + "../outside")))
name = prefix + "file"
assert next(iter(read_cache(BytesIO(_raw_index(name)))[1])) == (name, 0)
stream = BytesIO()
write_cache([IndexEntry((0o100644, b"a" * 20, 0, name))], stream)
stream.seek(0)
assert next(iter(read_cache(stream)[1])) == (name, 0)
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:
assert "required argument is not an integer" 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)
@with_rw_repo("0.1.6")
def test_read_tree_methods_reject_index_output(self, rw_repo):
output_path = (Path(rw_repo.working_tree_dir) / "alternate-index").as_posix()
unsafe_option = f"--index-output={output_path}"
with pytest.raises(UnsafeOptionError):
IndexFile.from_tree(rw_repo, unsafe_option)
with pytest.raises(UnsafeOptionError):
IndexFile.from_tree(rw_repo, "HEAD", index_output=output_path)
with pytest.raises(UnsafeOptionError):
rw_repo.index.reset(unsafe_option)
with pytest.raises(UnsafeOptionError):
rw_repo.index.merge_tree(unsafe_option)
with pytest.raises(UnsafeOptionError):
rw_repo.index.merge_tree("HEAD", base=unsafe_option)
@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
def merge_required(t):
return 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
# 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])
@with_rw_repo("0.1.6")
def test_index_file_diffing(self, rw_repo):
# Default IndexFile 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
@with_rw_directory
def test_index_file_diff_null_tree_with_initial_index(self, rw_dir):
repo = Repo.init(rw_dir)
filename = ".gitkeep"
file_path = osp.join(repo.working_tree_dir, filename)
with open(file_path, "w") as fp:
fp.write("# Initial file\n")
index = repo.index
index.add([filename])
index.write()
index = IndexFile(repo)
self.assertEqual(len(index.diff(None)), 0)
diff = index.diff(NULL_TREE)
self.assertEqual(len(diff), 1)
self.assertEqual(diff[0].change_type, "A")
assert diff[0].new_file
self.assertEqual(diff[0].b_path, filename)
self.assertEqual(len(index.diff(NULL_TREE, paths=filename)), 1)
self.assertEqual(len(index.diff(NULL_TREE_SHA, paths=filename)), 1)
self.assertEqual(len(index.diff(NULL_TREE, paths="missing")), 0)
patch = index.diff(NULL_TREE, create_patch=True)
self.assertEqual(len(patch), 1)
self.assertIn(b"+# Initial file", patch[0].diff)
with self.assertRaises(GitCommandError) as exc_info:
index.diff(NULL_TREE, bogus_option=True)
self.assertIn("usage: git diff", exc_info.exception.stderr)
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
@ddt.data("write", "write_tree", "checkout", "cached_checkout")
@with_rw_directory
def test_index_boundaries_reject_injected_entries_before_side_effects(self, rw_dir, operation):
tmp_path = Path(rw_dir) / "repo"
with Repo.init(tmp_path) as repo:
index_path = Path(repo.index.path)
if operation == "checkout":
index_path.write_bytes(_raw_index("../outside"))
else:
repo.index.write()
before = index_path.read_bytes()
index = repo.index
if operation == "cached_checkout":
assert not index.entries
index_path.write_bytes(_raw_index("../outside"))
before = index_path.read_bytes()
operation = "checkout"
elif operation != "checkout":
index.entries[("../outside", 0)] = IndexEntry((0o100644, b"a" * 20, 0, "../outside"))
with pytest.raises(ValueError):
getattr(index, operation)()
assert index_path.read_bytes() == before
assert not (tmp_path.parent / "outside").exists()
@with_rw_repo("0.1.6")
def test_index_mutation(self, rw_repo):
with xfail_if_raises(
sys.platform == "win32" and (Git().config("core.symlinks") == "true" or symlinks_supported()),
raises=(FileNotFoundError, GitCommandError),
reason="Assumes symlinks are not created on Windows and opens a symlink to a nonexistent target.",
):
index = rw_repo.index
num_entries = len(index.entries)
cur_head = rw_repo.head
uname = "Thomas Mller"
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 % 5
if type_id == 0: # path (str)
yield entry.path
elif type_id == 1: # path (PathLike)
yield Path(entry.path)
elif type_id == 2: # path mock (PathLike)
yield PathLikeMock(entry.path)
elif type_id == 3: # path mock in a blob
yield Blob(rw_repo, entry.binsha, entry.mode, entry.path)
elif type_id == 4: # blob
yield Blob(rw_repo, entry.binsha, entry.mode, entry.path)
elif type_id == 5: # BaseIndexEntry
yield BaseIndexEntry(entry[:4])
elif type_id == 6: # 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 Frderic 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("Frderic aufl", "author@example.com")
my_committer = Actor("Committing Frderic 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, "Frderic aufl")
self.assertEqual(commit_actor.author.email, "author@example.com")
self.assertEqual(commit_actor.committer.name, "Committing Frderic 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 sys.platform != "win32":
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 out as a 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 currently assume we will never get symlinks.
if sys.platform == "win32":
# 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)
@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.binsha,), (B.hexsha, H), (B, H.binsha, M.hexsha.encode("ascii"))):
index = IndexFile.new(self.rorepo, *args)
assert isinstance(index, IndexFile)
# END for each arg tuple
@ddt.data(*product(("../outside", ".git/hooks/pre-commit", "a/.GIT/config"), (1, 2, 3)))
@ddt.unpack
@with_rw_directory
def test_native_tree_merge_rejects_unsafe_paths(self, rw_dir, path, tree_count):
tmp_path = Path(rw_dir)
with Repo.init(tmp_path) as repo:
blob = repo.odb.store(IStream("blob", 4, BytesIO(b"data"))).binsha
data = b"100755 " + path.encode() + b"\0" + blob
tree = repo.odb.store(IStream("tree", len(data), BytesIO(data))).binsha
empty = repo.odb.store(IStream("tree", 0, BytesIO())).binsha
with pytest.raises(ValueError):
IndexFile.new(repo, *([empty] * (tree_count - 1) + [tree]))
assert not (tmp_path / ".git" / "index").exists()
@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):
fp = osp.join(rw_dir, ".txt")
with open(fp, "wb") as fs:
fs.write("content of ".encode())
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")
@ddt.data(*product(("path", "glob", "blob", "entry"), (False, True)))
@ddt.unpack
@with_rw_directory
def test_staging_never_reads_through_a_directory_symlink(self, rw_dir, kind, outside):
tmp_path = Path(rw_dir)
with Repo.init(tmp_path / "repo") as repo:
root = tmp_path / "repo"
target = (tmp_path if outside else root) / "target"
target.mkdir()
(target / "secret").write_text("private data")
try:
(root / "link").symlink_to(target, target_is_directory=True)
except OSError:
pytest.skip("Symlinks unavailable")
item = "link/secret"
if kind == "glob":
item = "link/*"
elif kind == "blob":
item = Blob(repo, Blob.NULL_BIN_SHA, 0o100644, item)
elif kind == "entry":
item = BaseIndexEntry((0o100644, Blob.NULL_BIN_SHA, 0, item))
with mock.patch.object(repo.odb, "store", wraps=repo.odb.store) as store:
with pytest.raises(ValueError, match="symbolic link"):
repo.index.add([item], write=False)
store.assert_not_called()
@ddt.data("blob", "entry", "rewriter")
@with_rw_directory
def test_staging_rejects_unsafe_object_paths_even_without_writing(self, rw_dir, kind):
with Repo.init(rw_dir) as repo:
entry = BaseIndexEntry((0o100644, b"a" * 20, 0, "../outside"))
item = Blob(repo, entry.binsha, entry.mode, entry.path) if kind == "blob" else entry
kwargs = {}
if kind == "rewriter":
item = BaseIndexEntry((0o100644, b"a" * 20, 0, "safe"))
kwargs["path_rewriter"] = lambda entry: "../outside"
index = repo.index
with pytest.raises(ValueError):
index.add([item], write=False, **kwargs)
assert not index.entries
@with_rw_directory
def test_staging_root_preserves_symlinks_and_skips_git_metadata(self, rw_dir):
tmp_path = Path(rw_dir)
root = tmp_path / "repo"
(tmp_path / "outside").mkdir()
with Repo.init(root) as repo:
(root / "file").write_text("contents")
try:
(root / "link").symlink_to("../outside", target_is_directory=True)
except OSError:
pytest.skip("Symlinks unavailable")
entries = repo.index.add(["."])
assert {entry.path for entry in entries} == {"file", "link"}
link = repo.index.entries[("link", 0)]
assert link.mode == 0o120000
assert repo.odb.stream(link.binsha).read() == os.fsencode(os.readlink(root / "link"))
@with_rw_directory
def test_staging_symlink_measures_encoded_target_instead_of_stat_size(self, rw_dir):
tmp_path = Path(rw_dir)
with Repo.init(tmp_path) as repo:
link = tmp_path / "link"
try:
link.symlink_to("../caf", target_is_directory=True)
except OSError:
pytest.skip("Symlinks unavailable")
target = os.fsencode(os.readlink(link))
original_lstat = os.lstat
def zero_size_for_symlinks(*args, **kwargs):
result = original_lstat(*args, **kwargs)
if S_ISLNK(result.st_mode):
fields = list(result)
fields[6] = 0
return os.stat_result(fields)
return result
with mock.patch("os.lstat", side_effect=zero_size_for_symlinks):
(entry,) = repo.index.add(["link"])
assert entry.mode == 0o120000
assert repo.odb.stream(entry.binsha).read() == target
@pytest.mark.skipif(os.name == "nt", reason="Colons are not valid Windows filenames")
@with_rw_directory
def test_staging_nested_colon_directory(self, rw_dir):
tmp_path = Path(rw_dir)
with Repo.init(tmp_path) as repo:
directory = tmp_path / "nested" / "a:b"
directory.mkdir(parents=True)
(directory / "file").write_text("contents")
assert [entry.path for entry in repo.index.add(["nested"])] == ["nested/a:b/file"]
assert repo.index.write_tree()["nested/a:b/file"].data_stream.read() == b"contents"
@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="FIFOs unavailable")
@with_rw_directory
def test_staging_special_files_fails_before_opening(self, rw_dir):
tmp_path = Path(rw_dir)
with Repo.init(tmp_path) as repo:
os.mkfifo(tmp_path / "fifo")
with mock.patch("builtins.open", side_effect=AssertionError("must not open a FIFO")):
with pytest.raises(ValueError, match="regular file"):
repo.index.add(["fifo"], write=False)
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))
def test__to_relative_path_absolute_trailing_slash(self):
repo_root = os.path.join(osp.abspath(os.sep), "directory1", "repo_root")
class Mocked:
bare = False
git_dir = repo_root
working_tree_dir = repo_root
repo = Mocked()
path = os.path.join(repo_root, f"directory2{os.sep}")
index = IndexFile(repo)
expected_path = "directory2/"
actual_path = index._to_relative_path(path)
self.assertEqual(expected_path, actual_path)
@pytest.mark.skipif(sys.platform != "win32", reason="Specifically for Windows.")
def test__to_relative_path_windows_unc_share_root(self):
for repo_root in [R"\\server\share", R"\\?\UNC\server\share"]:
with self.subTest(repo_root=repo_root):
repo = mock.Mock(bare=False, git_dir=repo_root, working_tree_dir=repo_root)
index = IndexFile(repo)
self.assertEqual(index._to_relative_path(repo_root), ".")
@pytest.mark.skipif(sys.platform != "win32", reason="Specifically for Windows.")
@with_rw_directory
def test__to_relative_path_windows_path_kinds(self, rw_dir):
repo_root = osp.join(rw_dir, "repo")
os.makedirs(osp.join(repo_root, "nested"))
class Mocked:
bare = False
git_dir = osp.join(repo_root, ".git")
working_tree_dir = repo_root
index = IndexFile(Mocked())
inside_path = osp.join(repo_root, "nested", "file")
_drive, rooted_inside_path = osp.splitdrive(inside_path)
self.assertEqual(index._to_relative_path(inside_path), "nested/file")
self.assertEqual(index._to_relative_path(rooted_inside_path), "nested/file")
self.assertEqual(index._to_relative_path(rooted_inside_path.replace("\\", "/")), "nested/file")
self.assertEqual(index._to_relative_path(PathLikeMock(inside_path)), "nested/file")
self.assertEqual(index._to_relative_path(inside_path.upper()), "NESTED/FILE")
self.assertRaises(ValueError, index._to_relative_path, osp.join(repo_root + "-other", "file"))
self.assertRaises(ValueError, index._to_relative_path, osp.join("..", "outside"))
self.assertRaises(ValueError, index._to_relative_path, f"{osp.splitdrive(repo_root)[0]}relative")
self.assertRaises(ValueError, index._to_relative_path, R"Z:\outside")
self.assertRaises(ValueError, index._to_relative_path, R"\\server\share\outside")
self.assertRaises(ValueError, index._to_relative_path, R"\\?\C:\outside")
Mocked.bare = True
bare_index = IndexFile(Mocked())
self.assertRaises(InvalidGitRepositoryError, bare_index._to_relative_path, rooted_inside_path)
self.assertRaises(
InvalidGitRepositoryError, bare_index._to_relative_path, f"{osp.splitdrive(repo_root)[0]}relative"
)
@pytest.mark.xfail(
type(_win_bash_status) is WinBashStatus.Absent,
reason="Can't run a hook on Windows without bash.exe.",
raises=HookExecutionError,
)
@pytest.mark.xfail(
type(_win_bash_status) is WinBashStatus.WslNoDistro,
reason="Currently uses the bash.exe of WSL, even with no WSL distro installed",
raises=HookExecutionError,
)
@with_rw_repo("HEAD", bare=True)
def test_run_commit_hook(self, rw_repo):
index = rw_repo.index
_make_hook(index.repo.git_dir, "fake-hook", "echo 'ran fake hook' >output.txt")
output = Path(rw_repo.git_dir, "output.txt")
with mock.patch.object(Repo, "config_level", ("repository",)):
with mock.patch.object(Git, "execute", side_effect=AssertionError("hook lookup must not run git")):
run_commit_hook("fake-hook", index)
self.assertEqual(output.read_text(encoding="utf-8"), "ran fake hook\n")
output.unlink()
with index.repo.config_writer() as writer:
writer.set_value("core", "hooksPath", "")
run_commit_hook("fake-hook", index)
self.assertEqual(output.read_text(encoding="utf-8"), "ran fake hook\n")
@with_rw_directory
def test_run_commit_hook_outside_worktree_on_windows(self, rw_dir):
root = Path(rw_dir).resolve()
repo = Repo.init(root / "repo")
hooks_dir = root / "hooks"
_make_hook(root, "fake-hook", "exit 0")
system_root = root / "Windows"
system_bash = system_root / "System32" / "bash.exe"
git_executable = root / "Git" / "cmd" / "git.exe"
git_bash = root / "Git" / "bin" / "bash.exe"
for executable in (system_bash, git_executable, git_bash):
executable.parent.mkdir(parents=True)
executable.touch()
executable.chmod(0o755)
with repo.config_writer() as writer:
writer.set_value("core", "hooksPath", str(hooks_dir))
# Model a normal Windows PATH: System32 (containing the WSL launcher) comes
# before Git's cmd directory, while Git's Bash is not itself on PATH. This
# exercises both Git-installation discovery and shell selection without
# mocking either resolver's answer.
with mock.patch("git.index.fun.sys.platform", "win32"), mock.patch.object(
Git, "GIT_PYTHON_GIT_EXECUTABLE", "git"
), mock.patch.dict(os.environ, {"SystemRoot": str(system_root)}), mock.patch(
"git.index.fun.os.get_exec_path", return_value=["", str(system_bash.parent), str(git_executable.parent)]
):
with mock.patch("git.index.fun.safer_popen") as popen, mock.patch("git.index.fun.handle_process_output"):
popen.return_value.returncode = 0
run_commit_hook("fake-hook", repo.index)
command = popen.call_args[0][0]
self.assertEqual(command, [str(git_bash), "../hooks/fake-hook"])
@with_rw_directory
def test_windows_bash_lookup_respects_explicit_current_directory_in_path(self, rw_dir):
root = Path(rw_dir).resolve()
bash = root / "bash.exe"
bash.touch()
bash.chmod(0o755)
# An explicitly listed directory is trusted PATH configuration, even when
# it happens to be the current directory. This differs from an empty entry,
# which Windows requires PATH lookup to ignore.
with cwd(root), mock.patch("git.index.fun.os.get_exec_path", return_value=[str(root)]):
self.assertEqual(_which_from_path("bash.exe"), str(bash))
@with_rw_directory
def test_windows_bash_lookup_from_explicit_git_bin(self, rw_dir):
git_root = Path(rw_dir).resolve() / "Git"
git_executable = git_root / "bin" / "git.exe"
bash = git_root / "bin" / "bash.exe"
git_executable.parent.mkdir(parents=True)
for executable in (git_executable, bash):
executable.touch()
executable.chmod(0o755)
# A relative executable containing a directory is resolved by CreateProcess
# from the parent process cwd, not the separately supplied child cwd. Enter the
# temporary root first because Windows cannot express a relative path between
# drives, and CI may keep the checkout and its temporary directory on different
# drives.
with cwd(Path(rw_dir).resolve()):
relative_git = osp.relpath(git_executable, os.curdir)
with mock.patch.object(Git, "GIT_PYTHON_GIT_EXECUTABLE", relative_git):
self.assertEqual(_git_for_windows_bash(), str(bash))
@with_rw_directory
def test_windows_bash_lookup_ignores_custom_git_executable(self, rw_dir):
root = Path(rw_dir).resolve()
for directory_name in ("cmd", "bin"):
executable = root / directory_name / "mygit.exe"
bash = root / "bin" / "bash.exe"
executable.parent.mkdir(parents=True, exist_ok=True)
bash.parent.mkdir(parents=True, exist_ok=True)
executable.touch()
bash.touch()
executable.chmod(0o755)
bash.chmod(0o755)
with mock.patch.object(Git, "GIT_PYTHON_GIT_EXECUTABLE", str(executable)):
self.assertIsNone(_git_for_windows_bash())
@ddt.data((False,), (True,))
@with_rw_directory
def test_hook_uses_shell_not_from_cwd(self, rw_dir, case):
(chdir_to_repo,) = case
shell_name = "bash.exe" if sys.platform == "win32" else "sh"
maybe_chdir = cwd(rw_dir) if chdir_to_repo else contextlib.nullcontext()
repo = Repo.init(rw_dir)
# We need an impostor shell that works on Windows and that the test can
# distinguish from the real bash.exe. But even if the real bash.exe is absent or
# unusable, we should verify the impostor is not run. So the impostor needs a
# clear side effect (unlike in TestGit.test_it_executes_git_not_from_cwd). Popen
# on Windows uses CreateProcessW, which disregards PATHEXT; the impostor may
# need to be a binary executable to ensure the vulnerability is found if
# present. No compiler need exist, shipping a binary in the test suite may
# target the wrong architecture, and generating one in a bespoke way may trigger
# false positive virus scans. So we use a Bash/Python polyglot for the hook and
# use the Python interpreter itself as the bash.exe impostor. But an interpreter
# from a venv may not run when copied outside of it, and a global interpreter
# won't run when copied to a different location if it was installed from the
# Microsoft Store. So we make a new venv in rw_dir and use its interpreter.
venv = VirtualEnvironment(rw_dir, with_pip=False)
shutil.copy(venv.python, Path(rw_dir, shell_name))
shutil.copy(fixture_path("polyglot"), hook_path("polyglot", repo.git_dir))
payload = Path(rw_dir, "payload.txt")
if type(_win_bash_status) in {WinBashStatus.Absent, WinBashStatus.WslNoDistro}:
# The real shell can't run, but the impostor should still not be used.
with self.assertRaises(HookExecutionError):
with maybe_chdir:
run_commit_hook("polyglot", repo.index)
self.assertFalse(payload.exists())
else:
# The real shell should run, and not the impostor.
with maybe_chdir:
run_commit_hook("polyglot", repo.index)
self.assertFalse(payload.exists())
output = Path(rw_dir, "output.txt").read_text(encoding="utf-8")
self.assertEqual(output, "Ran intended hook.\n")
@pytest.mark.xfail(
type(_win_bash_status) is WinBashStatus.Absent,
reason="Can't run a hook on Windows without bash.exe.",
raises=HookExecutionError,
)
@pytest.mark.xfail(
type(_win_bash_status) is WinBashStatus.WslNoDistro,
reason="Currently uses the bash.exe of WSL, even with no WSL distro installed",
raises=HookExecutionError,
)
@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")
@pytest.mark.xfail(
type(_win_bash_status) is WinBashStatus.Absent,
reason="Can't run a hook on Windows without bash.exe.",
raises=HookExecutionError,
)
@pytest.mark.xfail(
type(_win_bash_status) is WinBashStatus.WslNoDistro,
reason="Currently uses the bash.exe of WSL, even with no WSL distro installed",
raises=HookExecutionError,
)
@with_rw_repo("HEAD")
def test_pre_commit_hook_respects_core_hooks_path(self, rw_repo):
index = rw_repo.index
hooks_dir = Path(index.repo.working_dir, "custom-hooks")
hooks_dir.mkdir()
hp = hooks_dir / "pre-commit"
hp.write_text(HOOKS_SHEBANG + "echo 'ran custom hook' >custom-hook-output.txt", encoding="utf-8")
os.chmod(hp, 0o744)
with index.repo.config_writer() as writer:
writer.set_value("core", "hooksPath", "custom-hooks")
index.commit("This should run the custom hook")
output = Path(rw_repo.working_dir, "custom-hook-output.txt").read_text(encoding="utf-8")
self.assertEqual(output, "ran custom hook\n")
@pytest.mark.xfail(
type(_win_bash_status) is WinBashStatus.WslNoDistro,
reason="Currently uses the bash.exe of WSL, even with no WSL distro installed",
raises=AssertionError,
)
@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 type(_win_bash_status) is WinBashStatus.Absent:
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(
type(_win_bash_status) is WinBashStatus.Absent,
reason="Can't run a hook on Windows without bash.exe.",
raises=HookExecutionError,
)
@pytest.mark.xfail(
type(_win_bash_status) is WinBashStatus.Wsl,
reason="Specifically seems to fail on WSL bash (in spite of #1399)",
raises=AssertionError,
)
@pytest.mark.xfail(
type(_win_bash_status) is WinBashStatus.WslNoDistro,
reason="Currently uses the bash.exe of WSL, even with no WSL distro installed",
raises=HookExecutionError,
)
@with_rw_repo("HEAD", bare=True)
def test_commit_msg_hook_success(self, rw_repo):
commit_message = "commit default head by Frderic 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))
@pytest.mark.xfail(
type(_win_bash_status) is WinBashStatus.WslNoDistro,
reason="Currently uses the bash.exe of WSL, even with no WSL distro installed",
raises=AssertionError,
)
@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 type(_win_bash_status) is WinBashStatus.Absent:
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_pathlib(self, rw_repo):
worktree = Path(rw_repo.working_tree_dir)
file = worktree / "file.txt"
file.touch()
rw_repo.index.add(file)
@with_rw_repo("HEAD")
def test_index_add_pathlike(self, rw_repo):
worktree = Path(rw_repo.working_tree_dir)
file = worktree / "file.txt"
file.touch()
rw_repo.index.add(PathLikeMock(str(file)))
@with_rw_repo("HEAD")
def test_index_add_non_normalized_path(self, rw_repo):
worktree = Path(rw_repo.working_tree_dir)
file = worktree / "file.txt"
file.touch()
non_normalized_path = file.as_posix()
if os.name != "nt":
non_normalized_path = "/" + non_normalized_path[1:].replace("/", "//")
rw_repo.index.add(non_normalized_path)
@ddt.data(0, 4, 5)
def test_unsupported_index_versions_fail_even_with_optimization(self, version):
data = b"DIRC" + struct.pack(">LL", version, 0)
data += sha1(data).digest()
with pytest.raises(AssertionError, match="Unsupported git index version"):
read_cache(BytesIO(data))
code = """
from io import BytesIO
import sys
from git.index.fun import read_cache
try:
read_cache(BytesIO(sys.stdin.buffer.read()))
except AssertionError as error:
if "Unsupported git index version" not in str(error):
raise
else:
raise SystemExit("Unsupported index version was accepted")
"""
result = subprocess.run([sys.executable, "-O", "-c", code], input=data, capture_output=True, timeout=10)
assert result.returncode == 0, result.stderr.decode()
@ddt.data(b"link", b"sdir")
def test_unsupported_mandatory_index_extensions_fail_closed(self, signature):
data = b"DIRC" + struct.pack(">LL", 2, 0) + signature + struct.pack(">L", 0)
with pytest.raises(ValueError, match="extension"):
read_cache(BytesIO(data + sha1(data).digest()))
def test_index_file_v3(self):
index = IndexFile(self.rorepo, fixture_path("index_extended_flags"))
assert index.entries
assert index.version == 3
assert len(index.entries) == 4
assert index.entries[("init.t", 0)].skip_worktree
# Write the data - it must match the original.
with tempfile.NamedTemporaryFile() as tmpfile:
index.write(tmpfile.name)
assert Path(tmpfile.name).read_bytes() == Path(fixture_path("index_extended_flags")).read_bytes()
@with_rw_directory
def test_index_file_v3_with_git_command(self, tmp_dir):
tmp_dir = Path(tmp_dir)
with cwd(tmp_dir):
git = Git(tmp_dir)
git.init()
file = tmp_dir / "file.txt"
file.write_text("hello")
git.add("--intent-to-add", "file.txt") # intent-to-add sets extended flag
repo = Repo(tmp_dir)
index = repo.index
assert len(index.entries) == 1
assert index.version == 3
entry = list(index.entries.values())[0]
assert entry.path == "file.txt"
assert entry.intent_to_add
file2 = tmp_dir / "file2.txt"
file2.write_text("world")
index.add(["file2.txt"])
index.write()
status_str = git.status(porcelain=True)
status_lines = status_str.splitlines()
assert " A file.txt" in status_lines
assert "A file2.txt" in status_lines
class TestIndexUtils:
@pytest.mark.parametrize("file_path_type", [str, Path])
def test_temporary_file_swap(self, tmp_path, file_path_type):
file_path = tmp_path / "foo"
file_path.write_bytes(b"some data")
with TemporaryFileSwap(file_path_type(file_path)) as ctx:
assert Path(ctx.file_path) == file_path
assert not file_path.exists()
# Recreate it with new data, so we can observe that they're really separate.
file_path.write_bytes(b"other data")
temp_file_path = Path(ctx.tmp_file_path)
assert temp_file_path.parent == file_path.parent
assert temp_file_path.name.startswith(file_path.name)
assert temp_file_path.read_bytes() == b"some data"
assert not temp_file_path.exists()
assert file_path.read_bytes() == b"some data" # Not b"other data".