Refuse force-pushes even if cache is cleared

diff --git a/Changelog b/Changelog
index 1f6a1bedef8fc6cb533b8647c5dd02fe91084298..3e6b10866a9cfff5d2aa01e22b6bf29074173e6a 100644 (file)
--- a/Changelog
+++ b/Changelog
@@ -1,4 +1,6 @@
 ## [Unreleased]
 ## [Unreleased]

+### Changed
+- Refuse force-push updates even if XDG_CACHE_HOME is cleared

 
 
 ## [1.3.1] - 2021-06-07
 
 
 ## [1.3.1] - 2021-06-07

diff --git a/git_cache.py b/git_cache.py
index 077406c63275d97d701e1e9498e1efca989045c1..1363f119b484555b1c4c48ae9983e7175a068608 100644 (file)
--- a/git_cache.py
+++ b/git_cache.py
@@ -4,23 +4,37 @@
 # implementation details for my comfort.  So we re-implement here half of
 # nix's builtins.fetchGit.  :(
 
 # 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 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
 
 
 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]
 
 
 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(
 def git_cachedir(repo: Repo) -> Path:
     # Use xdg module when it's less painful to have as a dependency
     XDG_CACHE_HOME = Path(


@@ -29,7 +43,18 @@ def git_cachedir(repo: Repo) -> Path:
     return Path(os.path.join(
         XDG_CACHE_HOME,
         'git-cache/v1',
     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:
 
 
 def is_ancestor(repo: Repo, descendant: RefOrRev, ancestor: RefOrRev) -> bool:


@@ -54,6 +79,36 @@ def verify_ancestry(
         raise Exception('%s is not an ancestor of %s' % (ancestor, descendant))
 
 
         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,
 @backoff.on_exception(
     backoff.expo,
     subprocess.CalledProcessError,


@@ -78,6 +133,7 @@ def fetch(repo: Repo, ref: Ref) -> Tuple[Path, Rev]:
     with open(os.path.join(cachedir, 'refs', 'heads', ref)) as rev_file:
         rev = Rev(rev_file.read(999).strip())
     verify_ancestry(repo, ref, rev)
     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
 
 
     return cachedir, rev
 

diff --git a/test_git_cache.py b/test_git_cache.py
index bbe71188272c9d13254c5b4f4674bb6a4e43001f..b7e0edeb149ff4d478d85f235f0ff7ed1ae17e3f 100644 (file)
--- a/test_git_cache.py
+++ b/test_git_cache.py
@@ -37,9 +37,14 @@ def _commit_file(
 class TestGitCache(unittest.TestCase):
 
     def setUp(self) -> None:
 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_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_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_AUTHOR_NAME'] = 'test_git_cache'
         os.environ['GIT_COMMITTER_NAME'] = 'test_git_cache'


@@ -55,6 +60,7 @@ class TestGitCache(unittest.TestCase):
 
     def tearDown(self) -> None:
         _setenv('XDG_CACHE_HOME', self.old_XDG_CACHE_HOME)
 
     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()
 
         self.tempdir.cleanup()
         self.xdgcache.cleanup()


@@ -204,6 +210,21 @@ class TestGitCache(unittest.TestCase):
         with self.assertRaises(Exception):
             git_cache.ensure_rev_available(self.upstream, 'master', rev)
 
         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()
 
 if __name__ == '__main__':
     unittest.main()