def _git(directory: str, *args: str) -> bytes:
p = subprocess.run(['git', '-C', directory] + list(args),
stdout=subprocess.PIPE, check=True)
- return p.stdout
+ return bytes(p.stdout)
def _commit_file(
os.environ['GIT_AUTHOR_EMAIL'] = 'test_git_cache@example.com'
os.environ['GIT_COMMITTER_EMAIL'] = 'test_git_cache@example.com'
+ os.environ['BACKOFF_MAX_TIME'] = '0'
+
self.tempdir = tempfile.TemporaryDirectory(prefix='git_cache_test-')
self.upstream = os.path.join(self.tempdir.name, 'upstream')
subprocess.run(['git', 'init', self.upstream], check=True)
d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev)
self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
+ def test_catch_up(self) -> None:
+ _git(self.upstream, 'checkout', '-b', 'otherbranch')
+ _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
+ rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
+ d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev)
+ self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
+ _git(self.upstream, 'checkout', 'master')
+ _git(self.upstream, 'merge', '--ff-only', 'otherbranch')
+ d = git_cache.ensure_rev_available(self.upstream, 'master', rev)
+ self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
+
def test_fetch_after_cache_deleted(self) -> None:
d1, rev1 = git_cache.fetch(self.upstream, 'master')
shutil.rmtree(d1)