# implementation details for my comfort. So we re-implement here half of
# nix's builtins.fetchGit. :(
+import functools
import hashlib
import logging
import os
import subprocess
import sys
+import time
-from typing import Tuple, Union
+from typing import Iterator, NamedTuple, Optional, TypeVar, Tuple, Union
import backoff
-Path = str # eg: "/home/user/.cache/git-cache/v1"
-Repo = str # eg: "https://github.com/NixOS/nixpkgs.git"
-Ref = str # eg: "master" or "v1.0.0"
-Rev = str # eg: "53a27350551844e1ed1a9257690294767389ef0d"
+Path = str # eg: "/home/user/.cache/git-cache/v1"
+Repo = str # eg: "https://github.com/NixOS/nixpkgs.git"
+Ref = str # eg: "master" or "v1.0.0"
+Rev = str # eg: "53a27350551844e1ed1a9257690294767389ef0d"
RefOrRev = Union[Ref, Rev]
+class _LogEntry(NamedTuple):
+ ref: Ref
+ rev: Rev
+
+
+T = TypeVar('T')
+
+
+def _repo_hashname(repo: Repo) -> str:
+ return hashlib.sha256(repo.encode()).hexdigest()
+
+
def git_cachedir(repo: Repo) -> Path:
# Use xdg module when it's less painful to have as a dependency
XDG_CACHE_HOME = Path(
return Path(os.path.join(
XDG_CACHE_HOME,
'git-cache/v1',
- hashlib.sha256(repo.encode()).hexdigest()))
+ _repo_hashname(repo)))
+
+
+def _log_filename(repo: Repo) -> Path:
+ # Use xdg module when it's less painful to have as a dependency
+ XDG_DATA_HOME = Path(
+ os.environ.get('XDG_DATA_HOME', os.path.expanduser('~/.local/share')))
+
+ return Path(os.path.join(
+ XDG_DATA_HOME,
+ 'git-cache/v1',
+ _repo_hashname(repo)))
def is_ancestor(repo: Repo, descendant: RefOrRev, ancestor: RefOrRev) -> bool:
raise Exception('%s is not an ancestor of %s' % (ancestor, descendant))
+def _read_fetch_log(repo: Repo) -> Iterator[_LogEntry]:
+ filename = _log_filename(repo)
+ if not os.path.exists(filename):
+ return
+ with open(filename, 'r') as f:
+ for line in f:
+ _, _, rev, ref = line.strip().split(maxsplit=3)
+ yield _LogEntry(ref, rev)
+
+
+def _last(it: Iterator[T]) -> Optional[T]:
+ return functools.reduce(lambda a, b: b, it, None)
+
+
+def _previous_fetched_rev(repo: Repo, ref: Ref) -> Optional[Rev]:
+ return _last(entry.rev for entry in _read_fetch_log(
+ repo) if entry.ref == ref)
+
+
+def _log_fetch(repo: Repo, ref: Ref, rev: Rev) -> None:
+ prev_rev = _previous_fetched_rev(repo, ref)
+ if prev_rev is not None:
+ verify_ancestry(repo, rev, prev_rev)
+ filename = _log_filename(repo)
+ os.makedirs(os.path.dirname(filename), exist_ok=True)
+ with open(filename, 'a') as f:
+ f.write('%s fetch %s %s\n' %
+ (time.strftime('%Y-%m%d-%H:%M:%S%z'), rev, ref))
+
+
@backoff.on_exception(
backoff.expo,
subprocess.CalledProcessError,
with open(os.path.join(cachedir, 'refs', 'heads', ref)) as rev_file:
rev = Rev(rev_file.read(999).strip())
verify_ancestry(repo, ref, rev)
+ _log_fetch(repo, ref, rev)
return cachedir, rev
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')
+ 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'
def tearDown(self) -> None:
_setenv('XDG_CACHE_HOME', self.old_XDG_CACHE_HOME)
+ _setenv('XDG_DATA_HOME', self.old_XDG_DATA_HOME)
self.tempdir.cleanup()
self.xdgcache.cleanup()
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)
+
if __name__ == '__main__':
unittest.main()