[scm] Mock global git config scope globally.

This will be useful for writing tests which rely on shared 'global'
config between different working directories.

This also adds support for mocking 'system' (global, immutable) and
'workspace' (local, mutable). The workspace scope I think is a bit iffy
though, given how `git cl` is actually built - currently scm.GIT doesn't
really know about clone vs. workspace, and afaik no config adjustements
actually apply to the workspace scope.

Adds tests for mocked git config implementation, including bug fixes
to the current implementation revealed by the tests.

R=ayatane, yiwzhang

Change-Id: Ia56d2a81d8df6ae75d9f8d0497be0d67bdc03651
Bug: 355505750
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/tools/depot_tools/+/5759163
Reviewed-by: Yiwei Zhang <yiwzhang@google.com>
Commit-Queue: Robbie Iannucci <iannucci@chromium.org>
changes/63/5759163/9
Robert Iannucci 1 year ago committed by LUCI CQ
parent cdf5599e67
commit a3fb9bad66

281
scm.py

@ -6,15 +6,15 @@
from __future__ import annotations
import abc
import contextlib
import os
import pathlib
import platform
import re
import threading
import typing
from collections import defaultdict
from typing import Collection, Iterable, Literal, Dict
from typing import Collection, Iterable, Iterator, Literal, Dict
from typing import Optional, Sequence, Mapping
import gclient_utils
@ -48,8 +48,8 @@ def determine_scm(root):
return 'diff'
GitConfigScope = Literal['system', 'local', 'worktree']
GitScopeOrder: list[GitConfigScope] = ['system', 'local', 'worktree']
GitConfigScope = Literal['system', 'global', 'local', 'worktree']
GitScopeOrder: list[GitConfigScope] = ['system', 'global', 'local', 'worktree']
GitFlatConfigData = Mapping[str, Sequence[str]]
@ -85,7 +85,7 @@ class GitConfigStateBase(metaclass=abc.ABCMeta):
"""
@abc.abstractmethod
def set_config_multi(self, key: str, value: str, *, append: bool,
def set_config_multi(self, key: str, value: str, *,
value_pattern: Optional[str], scope: GitConfigScope):
"""When invoked, this should replace all existing values of `key` with
`value` in the git scope `scope` in this state's underlying data.
@ -103,7 +103,10 @@ class GitConfigStateBase(metaclass=abc.ABCMeta):
"""When invoked, remove a singlar value from `key` in this state's underlying data.
If missing_ok is False and `key` is not present in the given scope, this
must raise GitConfigUnsetMissingValue with `key`.
must raise GitConfigUnsetMissingValue with `key` and `scope`.
If `key` is multi-valued in this scope, this must raise
GitConfigUnsetMultipleValues with `key` and `scope`.
"""
@abc.abstractmethod
@ -115,7 +118,7 @@ class GitConfigStateBase(metaclass=abc.ABCMeta):
be removed.
If missing_ok is False and `key` is not present in the given scope, this
must raise GitConfigUnsetMissingValue with `key`.
must raise GitConfigUnsetMissingValue with `key` and `scope`.
TODO: Make value_pattern an re.Pattern. This wasn't done at the time of
this refactor to keep the refactor small.
@ -130,6 +133,26 @@ class GitConfigUnsetMissingValue(ValueError):
)
class GitConfigUnsetMultipleValues(ValueError):
def __init__(self, key: str, scope: str) -> None:
super().__init__(
f'Cannot unset multi-value key {key!r} in scope {scope!r} with modify_all=False.'
)
class GitConfigUneditableScope(ValueError):
def __init__(self, scope: str) -> None:
super().__init__(f'Cannot edit git config in scope {scope!r}.')
class GitConfigUnknownScope(ValueError):
def __init__(self, scope: str) -> None:
super().__init__(f'Unknown git config scope {scope!r}.')
class CachedGitConfigState(object):
"""This represents the observable git configuration state for a given
repository (whose top-level path is `root`).
@ -223,7 +246,8 @@ class CachedGitConfigState(object):
key: The specific config key to affect.
value: The value to set. If this is None, `key` will be unset.
append: If True and `value` is not None, this will append
the value instead of replacing an existing one.
the value instead of replacing an existing one. Must not be
specified with value_pattern.
missing_ok: If `value` is None (i.e. this is an unset operation),
ignore retcode=5 from `git config` (meaning that the value is
not present). If `value` is not None, then this option has no
@ -231,15 +255,20 @@ class CachedGitConfigState(object):
GitConfigUnsetMissingValue.
modify_all: If True, this will change a set operation to
`--replace-all`, and will change an unset operation to
`--unset-all`.
scope: By default this is the local scope, but could be `system`,
`global`, or `worktree`, depending on which config scope you
want to affect.
`--unset-all`. Must not be specified with value_pattern.
scope: By default this is the `local` scope, but could be `global`
or `worktree`, depending on which config scope you want to affect.
Note that the `system` scope cannot be modified.
value_pattern: For use with `modify_all=True`, allows
further filtering of the set or unset operation based on
the currently configured value. Ignored for
`modify_all=False`.
"""
if scope not in GitScopeOrder:
raise GitConfigUnknownScope(scope)
if scope == 'system':
raise GitConfigUneditableScope(scope)
if value is None:
if modify_all:
self._impl.unset_config_multi(key,
@ -249,13 +278,27 @@ class CachedGitConfigState(object):
else:
self._impl.unset_config(key, scope=scope, missing_ok=missing_ok)
else:
if modify_all:
if value_pattern:
if not modify_all:
raise ValueError(
'SetConfig with (value_pattern) and (not modify_all) is invalid.'
)
if append:
raise ValueError(
'SetConfig with (value_pattern) and (append) is invalid.'
)
self._impl.set_config_multi(key,
value,
append=append,
value_pattern=value_pattern,
scope=scope)
else:
if modify_all:
self._impl.set_config_multi(key,
value,
value_pattern=None,
scope=scope)
self._impl.set_config(key, value, append=append, scope=scope)
# Once the underlying storage has set the value, we clear our cache so
@ -297,13 +340,11 @@ class GitConfigStateReal(GitConfigStateBase):
args.append('--add')
GIT.Capture(args, cwd=self.root)
def set_config_multi(self, key: str, value: str, *, append: bool,
def set_config_multi(self, key: str, value: str, *,
value_pattern: Optional[str], scope: GitConfigScope):
args = ['config', f'--{scope}', '--replace-all', key, value]
if value_pattern is not None:
args.append(value_pattern)
if append:
args.append('--add')
GIT.Capture(args, cwd=self.root)
def unset_config(self, key: str, *, scope: GitConfigScope,
@ -315,6 +356,8 @@ class GitConfigStateReal(GitConfigStateBase):
accepted_retcodes=accepted_retcodes)
except subprocess2.CalledProcessError as cpe:
if cpe.returncode == 5:
if b'multiple values' in cpe.stderr:
raise GitConfigUnsetMultipleValues(key, scope)
raise GitConfigUnsetMissingValue(key, scope)
raise
@ -335,98 +378,151 @@ class GitConfigStateReal(GitConfigStateBase):
class GitConfigStateTest(GitConfigStateBase):
"""A fake implementation of GitConfigStateBase for testing."""
"""A fake implementation of GitConfigStateBase for testing.
To properly initialize this, see tests/scm_mock.py.
"""
def __init__(self,
initial_state: Optional[Dict[GitConfigScope,
GitFlatConfigData]] = None):
self.state: Dict[GitConfigScope, Dict[str, list[str]]] = {}
if initial_state is not None:
# We want to copy initial_state to make it mutable inside our class.
for scope, data in initial_state.items():
self.state[scope] = {k: list(v) for k, v in data.items()}
global_state_lock: threading.Lock,
global_state: dict[str, list[str]],
*,
system_state: Optional[GitFlatConfigData] = None,
local_state: Optional[GitFlatConfigData] = None,
worktree_state: Optional[GitFlatConfigData] = None):
"""Initializes a new (local, worktree) config state, with a reference to
a single global `global` state and an optional immutable `system` state.
The caller must supply a single shared Lock, plus a mutable reference to
the global-state dictionary.
Optionally, the caller may supply an initial local/worktree
configuration state.
This implementation will hold global_state_lock during all read/write
operations on the 'global' scope.
"""
self.system_state: GitFlatConfigData = system_state or {}
self.global_state_lock = global_state_lock
self.global_state = global_state
self.worktree_state: dict[str, list[str]] = {}
if worktree_state is not None:
self.worktree_state = {
k: list(v)
for k, v in worktree_state.items()
}
self.local_state: dict[str, list[str]] = {}
if local_state is not None:
self.local_state = {k: list(v) for k, v in local_state.items()}
super().__init__()
def _get_scope(self, scope: GitConfigScope) -> Dict[str, list[str]]:
ret = self.state.get(scope, None)
if ret is None:
ret = {}
self.state[scope] = ret
return ret
@contextlib.contextmanager
def _editable_scope(
self, scope: GitConfigScope) -> Iterator[dict[str, list[str]]]:
if scope == 'system':
# This is also checked in CachedGitConfigState.SetConfig, but double
# check here.
raise GitConfigUneditableScope(scope)
if scope == 'global':
with self.global_state_lock:
yield self.global_state
elif scope == 'local':
yield self.local_state
elif scope == 'worktree':
yield self.worktree_state
else:
# This is also checked in CachedGitConfigState.SetConfig, but double
# check here.
raise GitConfigUnknownScope(scope)
def load_config(self) -> GitFlatConfigData:
ret = {}
ret = {k: list(v) for k, v in self.system_state.items()}
for scope in GitScopeOrder:
for key, value in self._get_scope(scope).items():
curvals = ret.get(key, None)
if curvals is None:
curvals = []
ret[key] = curvals
curvals.extend(value)
if scope == 'system':
continue
with self._editable_scope(scope) as cfg:
for key, value in cfg.items():
curvals = ret.get(key, None)
if curvals is None:
curvals = []
ret[key] = curvals
curvals.extend(value)
return ret
def set_config(self, key: str, value: str, *, append: bool,
scope: GitConfigScope):
cfg = self._get_scope(scope)
cur = cfg.get(key)
if cur is None or len(cur) == 1:
if append:
cfg[key] = (cur or []) + [value]
else:
cfg[key] = [value]
return
raise ValueError(f'GitConfigStateTest: Cannot set key {key} '
f'- current value {cur!r} is multiple.')
with self._editable_scope(scope) as cfg:
cur = cfg.get(key)
if cur is None or len(cur) == 1:
if append:
cfg[key] = (cur or []) + [value]
else:
cfg[key] = [value]
return
raise ValueError(f'GitConfigStateTest: Cannot set key {key} '
f'- current value {cur!r} is multiple.')
def set_config_multi(self, key: str, value: str, *, append: bool,
def set_config_multi(self, key: str, value: str, *,
value_pattern: Optional[str], scope: GitConfigScope):
cfg = self._get_scope(scope)
cur = cfg.get(key)
if value_pattern is None or cur is None:
if append:
cfg[key] = (cur or []) + [value]
else:
with self._editable_scope(scope) as cfg:
cur = cfg.get(key)
if value_pattern is None or cur is None:
cfg[key] = [value]
return
return
pat = re.compile(value_pattern)
newval = [v for v in cur if pat.match(v)]
newval.append(value)
cfg[key] = newval
# We want to insert `value` in place of the first pattern match - if
# multiple values match, they will all be removed.
pat = re.compile(value_pattern)
newval = []
added = False
for val in cur:
if pat.match(val):
if not added:
newval.append(value)
added = True
else:
newval.append(val)
if not added:
newval.append(value)
cfg[key] = newval
def unset_config(self, key: str, *, scope: GitConfigScope,
missing_ok: bool):
cfg = self._get_scope(scope)
cur = cfg.get(key)
if cur is None:
if missing_ok:
with self._editable_scope(scope) as cfg:
cur = cfg.get(key)
if cur is None:
if missing_ok:
return
raise GitConfigUnsetMissingValue(key, scope)
if len(cur) == 1:
del cfg[key]
return
raise GitConfigUnsetMissingValue(key, scope)
if len(cur) == 1:
del cfg[key]
return
raise ValueError(f'GitConfigStateTest: Cannot unset key {key} '
f'- current value {cur!r} is multiple.')
raise GitConfigUnsetMultipleValues(key, scope)
def unset_config_multi(self, key: str, *, value_pattern: Optional[str],
scope: GitConfigScope, missing_ok: bool):
cfg = self._get_scope(scope)
cur = cfg.get(key)
if cur is None:
if not missing_ok:
raise GitConfigUnsetMissingValue(key, scope)
return
with self._editable_scope(scope) as cfg:
cur = cfg.get(key)
if cur is None:
if not missing_ok:
raise GitConfigUnsetMissingValue(key, scope)
return
if value_pattern is None:
del cfg[key]
return
if value_pattern is None:
del cfg[key]
return
if cur is None:
del cfg[key]
return
if cur is None:
del cfg[key]
return
pat = re.compile(value_pattern)
cfg[key] = [v for v in cur if not pat.match(v)]
pat = re.compile(value_pattern)
cfg[key] = [v for v in cur if not pat.match(v)]
class GIT(object):
@ -608,17 +704,19 @@ class GIT(object):
key: The specific config key to affect.
value: The value to set. If this is None, `key` will be unset.
append: If True and `value` is not None, this will append
the value instead of replacing an existing one.
the value instead of replacing an existing one. Must not be
specified with value_pattern.
missing_ok: If `value` is None (i.e. this is an unset operation),
ignore retcode=5 from `git config` (meaning that the value is
not present). If `value` is not None, then this option has no
effect.
effect. If this is false and the key is missing, this will raise
GitConfigUnsetMissingValue.
modify_all: If True, this will change a set operation to
`--replace-all`, and will change an unset operation to
`--unset-all`.
scope: By default this is the local scope, but could be `system`,
`global`, or `worktree`, depending on which config scope you
want to affect.
`--unset-all`. Must not be specified with value_pattern.
scope: By default this is the `local` scope, but could be `global`
or `worktree`, depending on which config scope you want to affect.
Note that the `system` scope cannot be modified.
value_pattern: For use with `modify_all=True`, allows
further filtering of the set or unset operation based on
the currently configured value. Ignored for
@ -989,3 +1087,6 @@ class DIFF(object):
dirnames[:] = [d for d in dirnames if should_recurse(dirpath, d)]
return [os.path.relpath(p, cwd) for p in paths]
# vim: sts=4:ts=4:sw=4:tw=80:et:

@ -4,6 +4,7 @@
import os
import sys
import threading
from typing import Dict, List, Optional
from unittest import mock
@ -29,21 +30,24 @@ def GIT(test: unittest.TestCase,
NOTE: The dependency on git_new_branch.create_new_branch seems pretty
circular - this functionality should probably move to scm.GIT?
"""
# TODO - remove `config` - have callers just directly call SetConfig with
# whatever config state they need.
# TODO - add `system_config` - this will be configuration which exists at
# the 'system installation' level and is immutable.
_branchref = [branchref or 'refs/heads/main']
initial_state = {}
if config is not None:
initial_state['local'] = config
global_lock = threading.Lock()
global_state = {}
def _newBranch(branchref):
_branchref[0] = branchref
patches: List[mock._patch] = [
mock.patch(
'scm.GIT._new_config_state',
side_effect=lambda root: scm.GitConfigStateTest(initial_state)),
mock.patch('scm.GIT.GetBranchRef',
side_effect=lambda _root: _branchref[0]),
mock.patch('scm.GIT._new_config_state',
side_effect=lambda _: scm.GitConfigStateTest(
global_lock, global_state, local_state=config)),
mock.patch('scm.GIT.GetBranchRef', side_effect=lambda _: _branchref[0]),
mock.patch('git_new_branch.create_new_branch', side_effect=_newBranch)
]

@ -4,10 +4,13 @@
# found in the LICENSE file.
"""Unit tests for scm.py."""
from __future__ import annotations
import logging
import os
import sys
import tempfile
import threading
import unittest
from unittest import mock
@ -418,9 +421,256 @@ class DiffTestCase(unittest.TestCase):
files, ["foo/file.txt", "foo/dir/file.txt", "baz_repo"])
class GitConfigStateTestTest(unittest.TestCase):
@staticmethod
def _make(*,
global_state: dict[str, list[str]] | None = None,
system_state: dict[str, list[str]] | None = None,
local_state: dict[str, list[str]] | None = None,
worktree_state: dict[str, list[str]] | None = None):
"""_make constructs a GitConfigStateTest with an internal Lock.
If global_state is None, an empty dictionary will be constructed and
returned, otherwise the caller's provided global_state is returned,
unmodified.
Returns (GitConfigStateTest, global_state) - access to global_state must
be manually synchronized with access to GitConfigStateTest, or at least
with GitConfigStateTest.global_state_lock.
"""
global_state = global_state or {}
m = scm.GitConfigStateTest(threading.Lock(),
global_state,
system_state=system_state,
local_state=local_state,
worktree_state=worktree_state)
return m, global_state
def test_construction_empty(self):
m, gs = self._make()
self.assertDictEqual(gs, {})
self.assertDictEqual(m.load_config(), {})
gs['key'] = ['override']
self.assertDictEqual(m.load_config(), {'key': ['override']})
def test_construction_global(self):
m, gs = self._make(global_state={'key': ['global']})
self.assertDictEqual(gs, {'key': ['global']})
self.assertDictEqual(m.load_config(), {'key': ['global']})
gs['key'] = ['override']
self.assertDictEqual(m.load_config(), {'key': ['override']})
def test_construction_system(self):
m, gs = self._make(
global_state={'key': ['global']},
system_state={'key': ['system']},
)
self.assertDictEqual(gs, {'key': ['global']})
self.assertDictEqual(m.load_config(), {'key': ['system', 'global']})
gs['key'] = ['override']
self.assertDictEqual(m.load_config(), {'key': ['system', 'override']})
def test_construction_local(self):
m, gs = self._make(
global_state={'key': ['global']},
system_state={'key': ['system']},
local_state={'key': ['local']},
)
self.assertDictEqual(gs, {'key': ['global']})
self.assertDictEqual(m.load_config(), {
'key': ['system', 'global', 'local'],
})
gs['key'] = ['override']
self.assertDictEqual(m.load_config(), {
'key': ['system', 'override', 'local'],
})
def test_construction_worktree(self):
m, gs = self._make(
global_state={'key': ['global']},
system_state={'key': ['system']},
local_state={'key': ['local']},
worktree_state={'key': ['worktree']},
)
self.assertDictEqual(gs, {'key': ['global']})
self.assertDictEqual(m.load_config(), {
'key': ['system', 'global', 'local', 'worktree'],
})
gs['key'] = ['override']
self.assertDictEqual(m.load_config(), {
'key': ['system', 'override', 'local', 'worktree'],
})
def test_set_config_system(self):
m, _ = self._make()
with self.assertRaises(scm.GitConfigUneditableScope):
m.set_config('key', 'new_global', append=False, scope='system')
def test_set_config_unkown(self):
m, _ = self._make()
with self.assertRaises(scm.GitConfigUnknownScope):
m.set_config('key', 'new_global', append=False, scope='meepmorp')
def test_set_config_global(self):
m, gs = self._make()
self.assertDictEqual(gs, {})
self.assertDictEqual(m.load_config(), {})
m.set_config('key', 'new_global', append=False, scope='global')
self.assertDictEqual(m.load_config(), {
'key': ['new_global'],
})
m.set_config('key', 'new_global2', append=True, scope='global')
self.assertDictEqual(m.load_config(), {
'key': ['new_global', 'new_global2'],
})
self.assertDictEqual(gs, {
'key': ['new_global', 'new_global2'],
})
def test_set_config_multi_global(self):
m, gs = self._make(global_state={
'key': ['1', '2'],
})
m.set_config_multi('key',
'new_global',
value_pattern=None,
scope='global')
self.assertDictEqual(m.load_config(), {
'key': ['new_global'],
})
self.assertDictEqual(gs, {
'key': ['new_global'],
})
m.set_config_multi('other',
'newval',
value_pattern=None,
scope='global')
self.assertDictEqual(m.load_config(), {
'key': ['new_global'],
'other': ['newval'],
})
self.assertDictEqual(gs, {
'key': ['new_global'],
'other': ['newval'],
})
def test_set_config_multi_global_pattern(self):
m, _ = self._make(global_state={
'key': ['1', '1', '2', '2', '2', '3'],
})
m.set_config_multi('key',
'new_global',
value_pattern='2',
scope='global')
self.assertDictEqual(m.load_config(), {
'key': ['1', '1', 'new_global', '3'],
})
m.set_config_multi('key',
'additional',
value_pattern='narp',
scope='global')
self.assertDictEqual(m.load_config(), {
'key': ['1', '1', 'new_global', '3', 'additional'],
})
def test_unset_config_global(self):
m, _ = self._make(global_state={
'key': ['someval'],
})
m.unset_config('key', scope='global', missing_ok=False)
self.assertDictEqual(m.load_config(), {})
with self.assertRaises(scm.GitConfigUnsetMissingValue):
m.unset_config('key', scope='global', missing_ok=False)
self.assertDictEqual(m.load_config(), {})
m.unset_config('key', scope='global', missing_ok=True)
self.assertDictEqual(m.load_config(), {})
def test_unset_config_global_extra(self):
m, _ = self._make(global_state={
'key': ['someval'],
'extra': ['another'],
})
m.unset_config('key', scope='global', missing_ok=False)
self.assertDictEqual(m.load_config(), {
'extra': ['another'],
})
with self.assertRaises(scm.GitConfigUnsetMissingValue):
m.unset_config('key', scope='global', missing_ok=False)
self.assertDictEqual(m.load_config(), {
'extra': ['another'],
})
m.unset_config('key', scope='global', missing_ok=True)
self.assertDictEqual(m.load_config(), {
'extra': ['another'],
})
def test_unset_config_global_multi(self):
m, _ = self._make(global_state={
'key': ['1', '2'],
})
with self.assertRaises(scm.GitConfigUnsetMultipleValues):
m.unset_config('key', scope='global', missing_ok=True)
def test_unset_config_multi_global(self):
m, _ = self._make(global_state={
'key': ['1', '2'],
})
m.unset_config_multi('key',
value_pattern=None,
scope='global',
missing_ok=False)
self.assertDictEqual(m.load_config(), {})
with self.assertRaises(scm.GitConfigUnsetMissingValue):
m.unset_config_multi('key',
value_pattern=None,
scope='global',
missing_ok=False)
def test_unset_config_multi_global_pattern(self):
m, _ = self._make(global_state={
'key': ['1', '2', '3', '1', '2'],
})
m.unset_config_multi('key',
value_pattern='2',
scope='global',
missing_ok=False)
self.assertDictEqual(m.load_config(), {
'key': ['1', '3', '1'],
})
if __name__ == '__main__':
if '-v' in sys.argv:
logging.basicConfig(level=logging.DEBUG)
unittest.main()
# vim: ts=2:sw=2:tw=80:et:
# vim: ts=4:sw=4:tw=80:et:

Loading…
Cancel
Save