Release 1.4.1

[git-cache] / test_git_cache.py

diff --git a/test_git_cache.py b/test_git_cache.py
index 7408cad13e340992c21f95b612f971bc77dbb0c0..ce80fcbfaca911589ae13cee4eb61dfb67c30cd1 100644 (file)
--- a/test_git_cache.py
+++ b/test_git_cache.py
@@ -4,13 +4,22 @@ import shutil
 import subprocess
 import unittest
 
 import subprocess
 import unittest
 

+from typing import Optional
+

 import git_cache
 
 
 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)
 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(
 
 
 def _commit_file(


@@ -28,104 +37,111 @@ 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')

-        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_AUTHOR_EMAIL'] = 'test_git_cache@example.com'
         os.environ['GIT_COMMITTER_EMAIL'] = 'test_git_cache@example.com'
 
 
         os.environ['GIT_AUTHOR_NAME'] = 'test_git_cache'
         os.environ['GIT_COMMITTER_NAME'] = 'test_git_cache'
         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'
+        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')
         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)
+        subprocess.run(['git', '-c', 'init.defaultBranch=main',
+                       'init', self.upstream], check=True)

         _commit_file(self.upstream, 'file', 'Contents', 'First commit')
 
     def tearDown(self) -> None:
         _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()
 
     def test_fetch(self) -> None:
 
         self.tempdir.cleanup()
         self.xdgcache.cleanup()
 
     def test_fetch(self) -> None:

-        d, rev = git_cache.fetch(self.upstream, 'master')
+        d, rev = git_cache.fetch(self.upstream, 'main')

         self.assertEqual(_git(d, 'show', '%s:file' % rev), b'Contents')
 
     def test_fetch_twice(self) -> None:
         self.assertEqual(_git(d, 'show', '%s:file' % rev), b'Contents')
 
     def test_fetch_twice(self) -> None:

-        d1, rev1 = git_cache.fetch(self.upstream, 'master')
+        d1, rev1 = git_cache.fetch(self.upstream, 'main')

         self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents')
         self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents')

-        d2, rev2 = git_cache.fetch(self.upstream, 'master')
+        d2, rev2 = git_cache.fetch(self.upstream, 'main')

         self.assertEqual(d1, d2)
         self.assertEqual(rev1, rev2)
         self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
 
     def test_fetch_then_ensure(self) -> None:
         self.assertEqual(d1, d2)
         self.assertEqual(rev1, rev2)
         self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
 
     def test_fetch_then_ensure(self) -> None:

-        d1, rev = git_cache.fetch(self.upstream, 'master')
+        d1, rev = git_cache.fetch(self.upstream, 'main')

         self.assertEqual(_git(d1, 'show', '%s:file' % rev), b'Contents')
         self.assertEqual(_git(d1, 'show', '%s:file' % rev), b'Contents')

-        d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
+        d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev)

         self.assertEqual(d1, d2)
         self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')
 
     def test_ensure_then_fetch(self) -> None:
         rev1 = _git(
             self.upstream, 'log', '--format=%H', '-n1').strip().decode()
         self.assertEqual(d1, d2)
         self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')
 
     def test_ensure_then_fetch(self) -> None:
         rev1 = _git(
             self.upstream, 'log', '--format=%H', '-n1').strip().decode()

-        d1 = git_cache.ensure_rev_available(self.upstream, 'master', rev1)
+        d1 = git_cache.ensure_rev_available(self.upstream, 'main', rev1)

         self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents')
         self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents')

-        d2, rev2 = git_cache.fetch(self.upstream, 'master')
+        d2, rev2 = git_cache.fetch(self.upstream, 'main')

         self.assertEqual(d1, d2)
         self.assertEqual(rev1, rev2)
         self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
 
     def test_fetch_new_file(self) -> None:
         self.assertEqual(d1, d2)
         self.assertEqual(rev1, rev2)
         self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
 
     def test_fetch_new_file(self) -> None:

-        d1, rev1 = git_cache.fetch(self.upstream, 'master')
+        d1, rev1 = git_cache.fetch(self.upstream, 'main')

         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')

-        d2, rev2 = git_cache.fetch(self.upstream, 'master')
+        d2, rev2 = git_cache.fetch(self.upstream, 'main')

         self.assertEqual(d1, d2)
         self.assertNotEqual(rev1, rev2)
         self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')
 
     def test_ensure_doesnt_fetch_new_file(self) -> None:
         self.assertEqual(d1, d2)
         self.assertNotEqual(rev1, rev2)
         self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')
 
     def test_ensure_doesnt_fetch_new_file(self) -> None:

-        d1, rev1 = git_cache.fetch(self.upstream, 'master')
+        d1, rev1 = git_cache.fetch(self.upstream, 'main')

         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
         rev2 = _git(
             self.upstream, 'log', '--format=%H', '-n1').strip().decode()
         self.assertNotEqual(rev1, rev2)
         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
         rev2 = _git(
             self.upstream, 'log', '--format=%H', '-n1').strip().decode()
         self.assertNotEqual(rev1, rev2)

-        d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev1)
+        d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev1)

         self.assertEqual(d1, d2)
         p = subprocess.run(
             ['git', '-C', d2, 'show', '%s:foofile' % rev2], check=False)
         self.assertNotEqual(p.returncode, 0)
 
     def test_ensure_doesnt_fetch_from_deleted_upstream(self) -> None:
         self.assertEqual(d1, d2)
         p = subprocess.run(
             ['git', '-C', d2, 'show', '%s:foofile' % rev2], check=False)
         self.assertNotEqual(p.returncode, 0)
 
     def test_ensure_doesnt_fetch_from_deleted_upstream(self) -> None:

-        d1, rev = git_cache.fetch(self.upstream, 'master')
+        d1, rev = git_cache.fetch(self.upstream, 'main')

         self.tempdir.cleanup()
         self.tempdir.cleanup()

-        d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
+        d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev)

         self.assertEqual(d1, d2)
 
     def test_ensure_fetches_new_file(self) -> None:
         self.assertEqual(d1, d2)
 
     def test_ensure_fetches_new_file(self) -> None:

-        d1, rev1 = git_cache.fetch(self.upstream, 'master')
+        d1, rev1 = git_cache.fetch(self.upstream, 'main')

         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
         rev2 = _git(
             self.upstream, 'log', '--format=%H', '-n1').strip().decode()
         self.assertNotEqual(rev1, rev2)
         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
         rev2 = _git(
             self.upstream, 'log', '--format=%H', '-n1').strip().decode()
         self.assertNotEqual(rev1, rev2)

-        d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev2)
+        d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev2)

         self.assertEqual(d1, d2)
         self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')
 
     def test_fetch_raises_on_invalid_repo(self) -> None:
         self.tempdir.cleanup()
         with self.assertRaises(Exception):
         self.assertEqual(d1, d2)
         self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')
 
     def test_fetch_raises_on_invalid_repo(self) -> None:
         self.tempdir.cleanup()
         with self.assertRaises(Exception):

-            git_cache.fetch(self.upstream, 'master')
+            git_cache.fetch(self.upstream, 'main')

 
     def test_ensure_raises_on_invalid_repo(self) -> None:
         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
         self.tempdir.cleanup()
         with self.assertRaises(Exception):
 
     def test_ensure_raises_on_invalid_repo(self) -> None:
         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
         self.tempdir.cleanup()
         with self.assertRaises(Exception):

-            git_cache.ensure_rev_available(self.upstream, 'master', rev)
+            git_cache.ensure_rev_available(self.upstream, 'main', rev)

 
     def test_fetch_raises_on_invalid_ref(self) -> None:
         with self.assertRaises(Exception):
 
     def test_fetch_raises_on_invalid_ref(self) -> None:
         with self.assertRaises(Exception):


@@ -148,7 +164,7 @@ class TestGitCache(unittest.TestCase):
         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
         with self.assertRaises(Exception):
         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
         with self.assertRaises(Exception):

-            git_cache.ensure_rev_available(self.upstream, 'master', rev)
+            git_cache.ensure_rev_available(self.upstream, 'main', rev)

 
     def test_ensure_other_branch(self) -> None:
         _git(self.upstream, 'checkout', '-b', 'otherbranch')
 
     def test_ensure_other_branch(self) -> None:
         _git(self.upstream, 'checkout', '-b', 'otherbranch')


@@ -157,33 +173,85 @@ class TestGitCache(unittest.TestCase):
         d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev)
         self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
 
         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', 'main')
+        _git(self.upstream, 'merge', '--ff-only', 'otherbranch')
+        d = git_cache.ensure_rev_available(self.upstream, 'main', rev)
+        self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
+

     def test_fetch_after_cache_deleted(self) -> None:
     def test_fetch_after_cache_deleted(self) -> None:

-        d1, rev1 = git_cache.fetch(self.upstream, 'master')
+        d1, rev1 = git_cache.fetch(self.upstream, 'main')

         shutil.rmtree(d1)
         shutil.rmtree(d1)

-        d2, rev2 = git_cache.fetch(self.upstream, 'master')
+        d2, rev2 = git_cache.fetch(self.upstream, 'main')

         self.assertEqual(d1, d2)
         self.assertEqual(rev1, rev2)
         self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
 
     def test_ensure_after_cache_deleted(self) -> None:
         self.assertEqual(d1, d2)
         self.assertEqual(rev1, rev2)
         self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
 
     def test_ensure_after_cache_deleted(self) -> None:

-        d1, rev = git_cache.fetch(self.upstream, 'master')
+        d1, rev = git_cache.fetch(self.upstream, 'main')

         shutil.rmtree(d1)
         shutil.rmtree(d1)

-        d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
+        d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev)

         self.assertEqual(d1, d2)
         self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')
 
     def test_fetch_raises_on_amend(self) -> None:
         self.assertEqual(d1, d2)
         self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')
 
     def test_fetch_raises_on_amend(self) -> None:

-        git_cache.fetch(self.upstream, 'master')
+        git_cache.fetch(self.upstream, 'main')

         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
         with self.assertRaises(Exception):
         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
         with self.assertRaises(Exception):

-            git_cache.fetch(self.upstream, 'master')
+            git_cache.fetch(self.upstream, 'main')

 
     def test_ensure_raises_on_amend(self) -> None:
 
     def test_ensure_raises_on_amend(self) -> None:

-        git_cache.fetch(self.upstream, 'master')
+        git_cache.fetch(self.upstream, 'main')
+        _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, 'main', rev)
+
+    def test_fetch_raises_on_amend_after_cache_deleted(self) -> None:
+        d, _ = git_cache.fetch(self.upstream, 'main')
+        shutil.rmtree(d)
+        _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
+        with self.assertRaises(Exception):
+            git_cache.fetch(self.upstream, 'main')
+
+    def test_ensure_raises_on_amend_after_cache_deleted(self) -> None:
+        d, _ = git_cache.fetch(self.upstream, 'main')
+        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(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)
+            git_cache.ensure_rev_available(self.upstream, 'main', rev)
+
+    def test_force_fetch_after_amend(self) -> None:
+        git_cache.fetch(self.upstream, 'main')
+        _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
+        git_cache.fetch(self.upstream, 'main', force=True)
+
+    def test_force_ensure_after_amend(self) -> None:
+        git_cache.fetch(self.upstream, 'main')
+        _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
+        rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
+        git_cache.ensure_rev_available(
+            self.upstream, 'main', rev, force=True)
+
+    def test_force_fetch_after_amend_and_cache_delete(self) -> None:
+        d, _ = git_cache.fetch(self.upstream, 'main')
+        shutil.rmtree(d)
+        _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
+        git_cache.fetch(self.upstream, 'main', force=True)
+
+    def test_force_ensure_after_amend_and_cache_delete(self) -> None:
+        d, _ = git_cache.fetch(self.upstream, 'main')
+        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, 'main', rev, force=True)

 
 
 if __name__ == '__main__':
 
 
 if __name__ == '__main__':