Read conditional include

git/config.py

@@ -13,6 +13,7 @@
 import logging
 import os
 import re
+import fnmatch
 from collections import OrderedDict
 
 from git.compat import (
@@ -38,6 +39,10 @@
 # represents the configuration level of a configuration file
 CONFIG_LEVELS = ("system", "user", "global", "repository")
 
+# Section pattern to detect conditional includes.
+# https://git-scm.com/docs/git-config#_conditional_includes
+CONDITIONAL_INCLUDE_REGEXP = re.compile(r"(?<=includeIf )\"(gitdir|gitdir/i|onbranch):(.+)\"")
+
 
 class MetaParserBuilder(abc.ABCMeta):
 
@@ -247,7 +252,7 @@ class GitConfigParser(with_metaclass(MetaParserBuilder, cp.RawConfigParser, obje
  # list of RawConfigParser methods able to change the instance
  _mutating_methods_ = ("add_section", "remove_section", "remove_option", "set")
 
- def __init__(self, file_or_files=None, read_only=True, merge_includes=True, config_level=None):
+ def __init__(self, file_or_files=None, read_only=True, merge_includes=True, config_level=None, repo=None):
  """Initialize a configuration reader to read the given file_or_files and to
  possibly allow changes to it by setting read_only False
 
@@ -262,7 +267,10 @@ def __init__(self, file_or_files=None, read_only=True, merge_includes=True, conf
  :param merge_includes: if True, we will read files mentioned in [include] sections and merge their
  contents into ours. This makes it impossible to write back an individual configuration file.
  Thus, if you want to modify a single configuration file, turn this off to leave the original
- dataset unaltered when reading it."""
+ dataset unaltered when reading it.
+ :param repo: Reference to repository to use if [includeIf] sections are found in configuration files.
+
+ """
  cp.RawConfigParser.__init__(self, dict_type=_OMD)
 
  # Used in python 3, needs to stay in sync with sections for underlying implementation to work
@@ -284,6 +292,7 @@ def __init__(self, file_or_files=None, read_only=True, merge_includes=True, conf
  self._dirty = False
  self._is_initialized = False
  self._merge_includes = merge_includes
+ self._repo = repo
  self._lock = None
  self._acquire_lock()
 
@@ -443,7 +452,57 @@ def string_decode(v):
  raise e
 
  def _has_includes(self):
- return self._merge_includes and self.has_section('include')
+ return self._merge_includes and len(self._included_paths())
+
+ def _included_paths(self):
+ """Return all paths that must be included to configuration.
+ """
+ paths = []
+
+ for section in self.sections():
+ if section == "include":
+ paths += self.items(section)
+
+ match = CONDITIONAL_INCLUDE_REGEXP.search(section)
+ if match is None or self._repo is None:
+ continue
+
+ keyword = match.group(1)
+ value = match.group(2).strip()
+
+ if keyword in ["gitdir", "gitdir/i"]:
+ value = osp.expanduser(value)
+
+ if not any(value.startswith(s) for s in ["./", "/"]):
+ value = "**/" + value
+ if value.endswith("/"):
+ value += "**"
+
+ # Ensure that glob is always case insensitive if required.
+ if keyword.endswith("/i"):
+ value = re.sub(
+ r"[a-zA-Z]",
+ lambda m: "[{}{}]".format(
+ m.group().lower(),
+ m.group().upper()
+ ),
+ value
+ )
+
+ if fnmatch.fnmatchcase(self._repo.git_dir, value):
+ paths += self.items(section)
+
+ elif keyword == "onbranch":
+ try:
+ branch_name = self._repo.active_branch.name
+ except TypeError:
+ # Ignore section if active branch cannot be retrieved.
+ continue
+
+ if fnmatch.fnmatchcase(branch_name, value):
+ paths += self.items(section)
+
+ return paths
 
  def read(self):
  """Reads the data stored in the files we have been initialized with. It will
@@ -482,7 +541,7 @@ def read(self):
  # Read includes and append those that we didn't handle yet
  # We expect all paths to be normalized and absolute (and will assure that is the case)
  if self._has_includes():
- for _, include_path in self.items('include'):
+ for _, include_path in self._included_paths():
  if include_path.startswith('~'):
  include_path = osp.expanduser(include_path)
  if not osp.isabs(include_path):

git/repo/base.py

@@ -452,7 +452,7 @@ def config_reader(self, config_level=None):
  files = [self._get_config_path(f) for f in self.config_level]
  else:
  files = [self._get_config_path(config_level)]
- return GitConfigParser(files, read_only=True)
+ return GitConfigParser(files, read_only=True, repo=self)
 
  def config_writer(self, config_level="repository"):
  """
@@ -467,7 +467,7 @@ def config_writer(self, config_level="repository"):
  system = system wide configuration file
  global = user level configuration file
  repository = configuration file for this repostory only"""
- return GitConfigParser(self._get_config_path(config_level), read_only=False)
+ return GitConfigParser(self._get_config_path(config_level), read_only=False, repo=self)
 
  def commit(self, rev=None):
  """The Commit object for the specified revision

test/test_config.py

@@ -6,6 +6,8 @@
 
 import glob
 import io
+import os
+from unittest import mock
 
 from git import (
  GitConfigParser
@@ -238,6 +240,128 @@ def check_test_value(cr, value):
  with GitConfigParser(fpa, read_only=True) as cr:
  check_test_value(cr, tv)
 
+ @with_rw_directory
+ def test_conditional_includes_from_git_dir(self, rw_dir):
+ # Initiate repository path
+ git_dir = osp.join(rw_dir, "target1", "repo1")
+ os.makedirs(git_dir)
+
+ # Initiate mocked repository
+ repo = mock.Mock(git_dir=git_dir)
+
+ # Initiate config files.
+ path1 = osp.join(rw_dir, "config1")
+ path2 = osp.join(rw_dir, "config2")
+ template = "[includeIf \"{}:{}\"]\n path={}\n"
+
+ with open(path1, "w") as stream:
+ stream.write(template.format("gitdir", git_dir, path2))
+
+ # Ensure that config is ignored if no repo is set.
+ with GitConfigParser(path1) as config:
+ assert not config._has_includes()
+ assert config._included_paths() == []
+
+ # Ensure that config is included if path is matching git_dir.
+ with GitConfigParser(path1, repo=repo) as config:
+ assert config._has_includes()
+ assert config._included_paths() == [("path", path2)]
+
+ # Ensure that config is ignored if case is incorrect.
+ with open(path1, "w") as stream:
+ stream.write(template.format("gitdir", git_dir.upper(), path2))
+
+ with GitConfigParser(path1, repo=repo) as config:
+ assert not config._has_includes()
+ assert config._included_paths() == []
+
+ # Ensure that config is included if case is ignored.
+ with open(path1, "w") as stream:
+ stream.write(template.format("gitdir/i", git_dir.upper(), path2))
+
+ with GitConfigParser(path1, repo=repo) as config:
+ assert config._has_includes()
+ assert config._included_paths() == [("path", path2)]
+
+ # Ensure that config is included with path using glob pattern.
+ with open(path1, "w") as stream:
+ stream.write(template.format("gitdir", "**/repo1", path2))
+
+ with GitConfigParser(path1, repo=repo) as config:
+ assert config._has_includes()
+ assert config._included_paths() == [("path", path2)]
+
+ # Ensure that config is ignored if path is not matching git_dir.
+ with open(path1, "w") as stream:
+ stream.write(template.format("gitdir", "incorrect", path2))
+
+ with GitConfigParser(path1, repo=repo) as config:
+ assert not config._has_includes()
+ assert config._included_paths() == []
+
+ # Ensure that config is included if path in hierarchy.
+ with open(path1, "w") as stream:
+ stream.write(template.format("gitdir", "target1/", path2))
+
+ with GitConfigParser(path1, repo=repo) as config:
+ assert config._has_includes()
+ assert config._included_paths() == [("path", path2)]
+
+ @with_rw_directory
+ def test_conditional_includes_from_branch_name(self, rw_dir):
+ # Initiate mocked branch
+ branch = mock.Mock()
+ type(branch).name = mock.PropertyMock(return_value="/foo/branch")
+
+ # Initiate mocked repository
+ repo = mock.Mock(active_branch=branch)
+
+ # Initiate config files.
+ path1 = osp.join(rw_dir, "config1")
+ path2 = osp.join(rw_dir, "config2")
+ template = "[includeIf \"onbranch:{}\"]\n path={}\n"
+
+ # Ensure that config is included is branch is correct.
+ with open(path1, "w") as stream:
+ stream.write(template.format("/foo/branch", path2))
+
+ with GitConfigParser(path1, repo=repo) as config:
+ assert config._has_includes()
+ assert config._included_paths() == [("path", path2)]
+
+ # Ensure that config is included is branch is incorrect.
+ with open(path1, "w") as stream:
+ stream.write(template.format("incorrect", path2))
+
+ with GitConfigParser(path1, repo=repo) as config:
+ assert not config._has_includes()
+ assert config._included_paths() == []
+
+ # Ensure that config is included with branch using glob pattern.
+ with open(path1, "w") as stream:
+ stream.write(template.format("/foo/**", path2))
+
+ with GitConfigParser(path1, repo=repo) as config:
+ assert config._has_includes()
+ assert config._included_paths() == [("path", path2)]
+
+ @with_rw_directory
+ def test_conditional_includes_from_branch_name_error(self, rw_dir):
+ # Initiate mocked repository to raise an error if HEAD is detached.
+ repo = mock.Mock()
+ type(repo).active_branch = mock.PropertyMock(side_effect=TypeError)
+
+ # Initiate config file.
+ path1 = osp.join(rw_dir, "config1")
+
+ # Ensure that config is ignored when active branch cannot be found.
+ with open(path1, "w") as stream:
+ stream.write("[includeIf \"onbranch:foo\"]\n path=/path\n")
+
+ with GitConfigParser(path1, repo=repo) as config:
+ assert not config._has_includes()
+ assert config._included_paths() == []
+
  def test_rename(self):
  file_obj = self._to_memcache(fixture_path('git_config'))
  with GitConfigParser(file_obj, read_only=False, merge_includes=False) as cw: