import subprocess
import unittest
+from typing import Optional
+
import git_cache
+def _setenv(var: str, value: Optional[str]) -> None:
+ if value is None:
+ del os.environ[var]
+ else:
+ os.environ[var] = value
+
+
def _git(directory: str, *args: str) -> bytes:
p = subprocess.run(['git', '-C', directory] + list(args),
stdout=subprocess.PIPE, check=True)
class TestGitCache(unittest.TestCase):
def setUp(self) -> None:
- self.xdgcache = tempfile.TemporaryDirectory(prefix='git_cache_test-')
+ self.xdgcache = tempfile.TemporaryDirectory(
+ prefix='git_cache_test-cache-')
+ self.xdgdata = tempfile.TemporaryDirectory(
+ prefix='git_cache_test-data-')
self.old_XDG_CACHE_HOME = os.environ.get('XDG_CACHE_HOME')
- os.environ['XDG_CACHE_HOME'] = self.xdgcache.name
+ self.old_XDG_DATA_HOME = os.environ.get('XDG_DATA_HOME')
+ _setenv('XDG_CACHE_HOME', self.xdgcache.name)
+ _setenv('XDG_DATA_HOME', self.xdgdata.name)
os.environ['GIT_AUTHOR_NAME'] = 'test_git_cache'
os.environ['GIT_COMMITTER_NAME'] = 'test_git_cache'
os.environ['GIT_COMMITTER_EMAIL'] = 'test_git_cache@example.com'
os.environ['BACKOFF_MAX_TIME'] = '0'
+ os.environ['FORCE_WARNING_TIME'] = '0' # ONLY FOR TEST USE!
self.tempdir = tempfile.TemporaryDirectory(prefix='git_cache_test-')
self.upstream = os.path.join(self.tempdir.name, 'upstream')
_commit_file(self.upstream, 'file', 'Contents', 'First commit')
def tearDown(self) -> None:
- if self.old_XDG_CACHE_HOME is None:
- del os.environ['XDG_CACHE_HOME']
- else:
- os.environ['XDG_CACHE_HOME'] = self.old_XDG_CACHE_HOME
+ _setenv('XDG_CACHE_HOME', self.old_XDG_CACHE_HOME)
+ _setenv('XDG_DATA_HOME', self.old_XDG_DATA_HOME)
self.tempdir.cleanup()
self.xdgcache.cleanup()
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)
with self.assertRaises(Exception):
git_cache.ensure_rev_available(self.upstream, 'master', rev)
+ def test_fetch_raises_on_amend_after_cache_deleted(self) -> None:
+ d, _ = git_cache.fetch(self.upstream, 'master')
+ shutil.rmtree(d)
+ _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
+ with self.assertRaises(Exception):
+ git_cache.fetch(self.upstream, 'master')
+
+ def test_ensure_raises_on_amend_after_cache_deleted(self) -> None:
+ d, _ = git_cache.fetch(self.upstream, 'master')
+ shutil.rmtree(d)
+ _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
+ rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
+ with self.assertRaises(Exception):
+ git_cache.ensure_rev_available(self.upstream, 'master', rev)
+
+ def test_force_fetch_after_amend(self) -> None:
+ git_cache.fetch(self.upstream, 'master')
+ _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
+ git_cache.fetch(self.upstream, 'master', force=True)
+
+ def test_force_ensure_after_amend(self) -> None:
+ git_cache.fetch(self.upstream, 'master')
+ _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
+ rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
+ git_cache.ensure_rev_available(
+ self.upstream, 'master', rev, force=True)
+
+ def test_force_fetch_after_amend_and_cache_delete(self) -> None:
+ d, _ = git_cache.fetch(self.upstream, 'master')
+ shutil.rmtree(d)
+ _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
+ git_cache.fetch(self.upstream, 'master', force=True)
+
+ def test_force_ensure_after_amend_and_cache_delete(self) -> None:
+ d, _ = git_cache.fetch(self.upstream, 'master')
+ shutil.rmtree(d)
+ _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
+ rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
+ git_cache.ensure_rev_available(
+ self.upstream, 'master', rev, force=True)
+
if __name__ == '__main__':
unittest.main()