# 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 glob
import io
import os
import os.path as osp
import subprocess
import sys
import time
from unittest import mock
import pytest
from git import GitConfigParser, Repo
from git.compat import defenc
from git.config import _OMD, cp
from git.util import cwd, rmfile
from test.lib import SkipTest, TestCase, fixture_path, with_rw_directory
_tc_lock_fpaths = osp.join(osp.dirname(__file__), "fixtures/*.lock")
def _rm_lock_files():
for lfp in glob.glob(_tc_lock_fpaths):
rmfile(lfp)
class TestBase(TestCase):
def setUp(self):
_rm_lock_files()
def tearDown(self):
for lfp in glob.glob(_tc_lock_fpaths):
if osp.isfile(lfp):
raise AssertionError("Previous TC left hanging git-lock file: {}".format(lfp))
def _to_memcache(self, file_path):
with open(file_path, "rb") as fp:
sio = io.BytesIO(fp.read())
sio.name = file_path
return sio
def test_read_write(self):
# The writer must create the exact same file as the one read before.
for filename in ("git_config", "git_config_global"):
file_obj = self._to_memcache(fixture_path(filename))
with GitConfigParser(file_obj, read_only=False) as w_config:
w_config.read() # Enforce reading.
assert w_config._sections
w_config.write() # Enforce writing.
# We stripped lines when reading, so the results differ.
assert file_obj.getvalue()
self.assertEqual(
file_obj.getvalue(),
self._to_memcache(fixture_path(filename)).getvalue(),
)
# Creating an additional config writer must fail due to exclusive
# access.
with self.assertRaises(IOError):
GitConfigParser(file_obj, read_only=False)
# Should still have a lock and be able to make changes.
assert w_config._lock._has_lock()
# Changes should be written right away.
sname = "my_section"
val = "myvalue"
w_config.add_section(sname)
assert w_config.has_section(sname)
w_config.set(sname, oname, val)
assert w_config.has_option(sname, oname)
assert w_config.get(sname, oname) == val
sname_new = "new_section"
ival = 10
w_config.set_value(sname_new, oname_new, ival)
assert w_config.get_value(sname_new, oname_new) == ival
file_obj.seek(0)
r_config = GitConfigParser(file_obj, read_only=True)
assert r_config.has_section(sname)
assert r_config.has_option(sname, oname)
assert r_config.get(sname, oname) == val
# END for each filename
def test_includes_order(self):
with GitConfigParser(list(map(fixture_path, ("git_config", "git_config_global")))) as r_config:
r_config.read() # Enforce reading.
# Simple inclusions, again checking them taking precedence.
assert r_config.get_value("sec", "var0") == "value0_included"
# This one should take the git_config_global value since included values
# must be considered as soon as they get them.
assert r_config.get_value("diff", "tool") == "meld"
try:
# FIXME: Split this assertion out somehow and mark it xfail (or fix it).
assert r_config.get_value("sec", "var1") == "value1_main"
except AssertionError as e:
raise SkipTest("Known failure -- included values are not in effect right away") from e
@with_rw_directory
def test_case_insensitive_names(self, rw_dir):
config_path = osp.join(rw_dir, "config")
with open(config_path, "wb") as config_file:
config_file.write(
b"[core]\n\tBigName = 1\n"
b"[CoRe]\n\tbigname = 2\n\tFlag\n"
b'[REMOTE "Origin"]\n\tUrl = upper\n'
b'[remote "origin"]\n\tURL = lower\n'
)
with GitConfigParser(config_path) as config:
for section in ("core", "CORE", "CoRe"):
for option in ("BigName", "bigname", "BIGNAME"):
self.assertTrue(config.has_section(section))
self.assertTrue(config.has_option(section, option))
self.assertEqual(config.get(section, option), "2")
self.assertEqual(config.getint(section, option), 2)
self.assertEqual(config.get_value(section, option), 2)
self.assertEqual(config.get_values(section, option), [1, 2])
self.assertIs(config.getboolean(section, "FLAG"), True)
self.assertEqual(config.sections(), ["core", 'REMOTE "Origin"', 'remote "origin"'])
self.assertEqual(config.items("CORE"), [("BigName", "2"), ("Flag", None)])
self.assertEqual(config.items_all("CORE"), [("BigName", ["1", "2"]), ("Flag", [None])])
self.assertIn("BigName", config.options("CORE"))
self.assertEqual(config.get('remote "Origin"', "URL"), "upper")
self.assertEqual(config.get('REMOTE "origin"', "url"), "lower")
self.assertFalse(config.has_section('remote "ORIGIN"'))
with self.assertRaises(cp.NoSectionError):
config.get('remote "ORIGIN"', "url")
git_config = ["git", "config", "--file", config_path]
self.assertEqual(subprocess.check_output(git_config + ["--get-all", "CORE.BIGNAME"]), b"1\n2\n")
self.assertEqual(subprocess.check_output(git_config + ["--get", "remote.Origin.URL"]), b"upper\n")
self.assertEqual(subprocess.check_output(git_config + ["--get", "REMOTE.origin.url"]), b"lower\n")
@with_rw_directory
def test_case_insensitive_writes_preserve_spelling(self, rw_dir):
config_path = osp.join(rw_dir, "config")
content = b'[CoRe]\n\tBigName = 1\n[REMOTE "Origin"]\n\tUrl = upper\n'
with open(config_path, "wb") as config_file:
config_file.write(content)
with GitConfigParser(config_path, read_only=False) as config:
config.set_value("core", "bigname", 1)
with open(config_path, "rb") as config_file:
self.assertEqual(config_file.read(), content)
with self.assertRaises(cp.DuplicateSectionError):
config.add_section("CORE")
config.set("CORE", "BIGNAME", "3")
config.add_value("core", "bigname", 4)
config.set_value("CORE", "NewKey", "new")
self.assertEqual(config.items_all("core"), [("BigName", ["3", "4"]), ("NewKey", ["new"])])
self.assertEqual(config.get_values("CORE", "BIGNAME"), [3, 4])
self.assertTrue(config.remove_option("CORE", "NEWKEY"))
self.assertFalse(config.has_option("core", "newkey"))
config.set_value("core", "newkey", "again")
self.assertIn(("newkey", "again"), config.items("CORE"))
self.assertTrue(config.remove_option("CORE", "NEWKEY"))
config.rename_section('remote "Origin"', 'Remote "Other"')
self.assertEqual(config.get('REMOTE "Other"', "URL"), "upper")
self.assertTrue(config.remove_section('REMOTE "Other"'))
with open(config_path, "rb") as config_file:
self.assertEqual(config_file.read(), b"[CoRe]\n\tBigName = 3\n\tBigName = 4\n")
with GitConfigParser(config_path) as config:
self.assertEqual(config.get_values("CORE", "bigname"), [3, 4])
@with_rw_directory
def test_case_insensitive_includes_and_remotes(self, rw_dir):
with Repo.init(rw_dir) as repo:
config_path = osp.join(repo.git_dir, "config")
with open(config_path, "ab") as config_file:
config_file.write(
b'[REMOTE "Origin"]\n\tURL = upper\n'
b'[Remote "origin"]\n\tUrl = lower\n'
b"[core]\n\tBigName = 1\n"
b"[INCLUDE]\n\tPaTh = included\n"
b'[INCLUDEIF "onbranch:*"]\n\tPATH = conditional\n'
b'[INCLUDEIF "ONBRANCH:*"]\n\tpath = wrong-case\n'
)
for filename, content in (
("included", b"[CORE]\n\tBIGNAME = 2\n"),
("conditional", b"[core]\n\tBranchName = 3\n"),
("wrong-case", b"[core]\n\tbigname = 4\n"),
):
with open(osp.join(repo.git_dir, filename), "wb") as config_file:
config_file.write(content)
with repo.config_reader("repository") as config:
self.assertEqual(config.get_values("CORE", "bigname"), [1, 2])
self.assertEqual(config.get_value("CORE", "branchname"), 3)
self.assertEqual([remote.name for remote in repo.remotes], ["Origin", "origin"])
self.assertEqual(repo.remote("Origin").config_reader.get("url"), "upper")
self.assertEqual(repo.remote("origin").config_reader.get("URL"), "lower")
@with_rw_directory
def test_lock_reentry(self, rw_dir):
fpl = osp.join(rw_dir, "l")
gcp = GitConfigParser(fpl, read_only=False)
with gcp as cw:
cw.set_value("include", "some_value", "a")
# Entering again locks the file again...
with gcp as cw:
cw.set_value("include", "some_other_value", "b")
# ...so creating an additional config writer must fail due to exclusive
# access.
with self.assertRaises(IOError):
GitConfigParser(fpl, read_only=False)
# but work when the lock is removed
with GitConfigParser(fpl, read_only=False):
assert osp.exists(fpl)
# Reentering with an existing lock must fail due to exclusive access.
with self.assertRaises(IOError):
gcp.__enter__()
def test_multi_line_config(self):
file_obj = self._to_memcache(fixture_path("git_config_with_comments"))
with GitConfigParser(file_obj, read_only=False) as config:
ev = "ruby -e '\n"
ev += " system %(git), %(merge-file), %(--marker-size=%L), %(%A), %(%O), %(%B)\n"
ev += " b = File.read(%(%A))\n"
ev += " b.sub!(/^ (\\d+). do\\n=+\\nActiveRecord::Schema\\." # noqa: E501
ev += "define.:version => (\\d+). do\\n>+ .*/) do\n"
ev += " %(ActiveRecord::Schema.define(:version => #{[$1, $2].max}) do)\n"
ev += " end\n"
ev += " File.open(%(%A), %(w)) {|f| f.write(b)}\n"
ev += " exit 1 if b.include?(%(