]> git.scottworley.com Git - git-cache/blobdiff - git_cache.py
Allow ancestry checks beween two revs
[git-cache] / git_cache.py
index 48430033e7afa805d307cf208b46c52a8e23bd0f..077406c63275d97d701e1e9498e1efca989045c1 100644 (file)
@@ -10,7 +10,7 @@ import os
 import subprocess
 import sys
 
-from typing import Tuple
+from typing import Tuple, Union
 
 import backoff
 
@@ -18,6 +18,7 @@ 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]
 
 
 def git_cachedir(repo: Repo) -> Path:
@@ -31,17 +32,26 @@ def git_cachedir(repo: Repo) -> Path:
         hashlib.sha256(repo.encode()).hexdigest()))
 
 
-def is_ancestor(repo: Repo, ref: Ref, rev: Rev) -> bool:
+def is_ancestor(repo: Repo, descendant: RefOrRev, ancestor: RefOrRev) -> bool:
     cachedir = git_cachedir(repo)
-    logging.debug('Checking if rev %s is an ancestor of ref "%s"', rev, ref)
-    process = subprocess.run(
-        ['git', '-C', cachedir, 'merge-base', '--is-ancestor', rev, ref], check=False)
+    logging.debug('Checking if %s is an ancestor of %s', ancestor, descendant)
+    process = subprocess.run(['git',
+                              '-C',
+                              cachedir,
+                              'merge-base',
+                              '--is-ancestor',
+                              ancestor,
+                              descendant],
+                             check=False)
     return process.returncode == 0
 
 
-def verify_ancestry(repo: Repo, ref: Ref, rev: Rev) -> None:
-    if not is_ancestor(repo, ref, rev):
-        raise Exception('Rev %s is not an ancestor of ref "%s"' % (rev, ref))
+def verify_ancestry(
+        repo: Repo,
+        descendant: RefOrRev,
+        ancestor: RefOrRev) -> None:
+    if not is_ancestor(repo, descendant, ancestor):
+        raise Exception('%s is not an ancestor of %s' % (ancestor, descendant))
 
 
 @backoff.on_exception(