git.scottworley.com Git - git-cache/blob - test_git_cache.py

   1 import os.path
   2 import tempfile
   3 import shutil
   4 import subprocess
   5 import unittest
   6
   7 import git_cache
   8
   9
  10 def _git(directory: str, *args: str) -> bytes:
  11     p = subprocess.run(['git', '-C', directory] + list(args),
  12                        stdout=subprocess.PIPE, check=True)
  13     return bytes(p.stdout)
  14
  15
  16 def _commit_file(
  17         directory: str,
  18         filename: str,
  19         contents: str,
  20         commit_message: str) -> None:
  21     with open(os.path.join(directory, filename), 'w') as f:
  22         f.write(contents)
  23     _git(directory, 'add', filename)
  24     _git(directory, 'commit', '-m', commit_message)
  25
  26
  27 # pylint: disable=too-many-public-methods
  28 class TestGitCache(unittest.TestCase):
  29
  30     def setUp(self) -> None:
  31         self.xdgcache = tempfile.TemporaryDirectory(prefix='git_cache_test-')
  32         self.old_XDG_CACHE_HOME = os.environ.get('XDG_CACHE_HOME')
  33         os.environ['XDG_CACHE_HOME'] = self.xdgcache.name
  34
  35         os.environ['GIT_AUTHOR_NAME'] = 'test_git_cache'
  36         os.environ['GIT_COMMITTER_NAME'] = 'test_git_cache'
  37         os.environ['GIT_AUTHOR_EMAIL'] = 'test_git_cache@example.com'
  38         os.environ['GIT_COMMITTER_EMAIL'] = 'test_git_cache@example.com'
  39
  40         os.environ['BACKOFF_MAX_TIME'] = '0'
  41
  42         self.tempdir = tempfile.TemporaryDirectory(prefix='git_cache_test-')
  43         self.upstream = os.path.join(self.tempdir.name, 'upstream')
  44         subprocess.run(['git', 'init', self.upstream], check=True)
  45         _commit_file(self.upstream, 'file', 'Contents', 'First commit')
  46
  47     def tearDown(self) -> None:
  48         if self.old_XDG_CACHE_HOME is None:
  49             del os.environ['XDG_CACHE_HOME']
  50         else:
  51             os.environ['XDG_CACHE_HOME'] = self.old_XDG_CACHE_HOME
  52
  53         self.tempdir.cleanup()
  54         self.xdgcache.cleanup()
  55
  56     def test_fetch(self) -> None:
  57         d, rev = git_cache.fetch(self.upstream, 'master')
  58         self.assertEqual(_git(d, 'show', '%s:file' % rev), b'Contents')
  59
  60     def test_fetch_twice(self) -> None:
  61         d1, rev1 = git_cache.fetch(self.upstream, 'master')
  62         self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents')
  63         d2, rev2 = git_cache.fetch(self.upstream, 'master')
  64         self.assertEqual(d1, d2)
  65         self.assertEqual(rev1, rev2)
  66         self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
  67
  68     def test_fetch_then_ensure(self) -> None:
  69         d1, rev = git_cache.fetch(self.upstream, 'master')
  70         self.assertEqual(_git(d1, 'show', '%s:file' % rev), b'Contents')
  71         d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
  72         self.assertEqual(d1, d2)
  73         self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')
  74
  75     def test_ensure_then_fetch(self) -> None:
  76         rev1 = _git(
  77             self.upstream, 'log', '--format=%H', '-n1').strip().decode()
  78         d1 = git_cache.ensure_rev_available(self.upstream, 'master', rev1)
  79         self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents')
  80         d2, rev2 = git_cache.fetch(self.upstream, 'master')
  81         self.assertEqual(d1, d2)
  82         self.assertEqual(rev1, rev2)
  83         self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
  84
  85     def test_fetch_new_file(self) -> None:
  86         d1, rev1 = git_cache.fetch(self.upstream, 'master')
  87         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
  88         d2, rev2 = git_cache.fetch(self.upstream, 'master')
  89         self.assertEqual(d1, d2)
  90         self.assertNotEqual(rev1, rev2)
  91         self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')
  92
  93     def test_ensure_doesnt_fetch_new_file(self) -> None:
  94         d1, rev1 = git_cache.fetch(self.upstream, 'master')
  95         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
  96         rev2 = _git(
  97             self.upstream, 'log', '--format=%H', '-n1').strip().decode()
  98         self.assertNotEqual(rev1, rev2)
  99         d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev1)
 100         self.assertEqual(d1, d2)
 101         p = subprocess.run(
 102             ['git', '-C', d2, 'show', '%s:foofile' % rev2], check=False)
 103         self.assertNotEqual(p.returncode, 0)
 104
 105     def test_ensure_doesnt_fetch_from_deleted_upstream(self) -> None:
 106         d1, rev = git_cache.fetch(self.upstream, 'master')
 107         self.tempdir.cleanup()
 108         d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
 109         self.assertEqual(d1, d2)
 110
 111     def test_ensure_fetches_new_file(self) -> None:
 112         d1, rev1 = git_cache.fetch(self.upstream, 'master')
 113         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 114         rev2 = _git(
 115             self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 116         self.assertNotEqual(rev1, rev2)
 117         d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev2)
 118         self.assertEqual(d1, d2)
 119         self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')
 120
 121     def test_fetch_raises_on_invalid_repo(self) -> None:
 122         self.tempdir.cleanup()
 123         with self.assertRaises(Exception):
 124             git_cache.fetch(self.upstream, 'master')
 125
 126     def test_ensure_raises_on_invalid_repo(self) -> None:
 127         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 128         self.tempdir.cleanup()
 129         with self.assertRaises(Exception):
 130             git_cache.ensure_rev_available(self.upstream, 'master', rev)
 131
 132     def test_fetch_raises_on_invalid_ref(self) -> None:
 133         with self.assertRaises(Exception):
 134             git_cache.fetch(self.upstream, 'nobranch')
 135
 136     def test_ensure_raises_on_invalid_ref(self) -> None:
 137         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 138         with self.assertRaises(Exception):
 139             git_cache.ensure_rev_available(self.upstream, 'nobranch', rev)
 140
 141     def test_ensure_raises_on_invalid_rev(self) -> None:
 142         with self.assertRaises(Exception):
 143             git_cache.ensure_rev_available(
 144                 self.upstream,
 145                 'nobranch',
 146                 '1234567890abcdef01234567890abcdef1234567')
 147
 148     def test_ensure_raises_on_rev_from_other_branch(self) -> None:
 149         _git(self.upstream, 'checkout', '-b', 'otherbranch')
 150         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 151         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 152         with self.assertRaises(Exception):
 153             git_cache.ensure_rev_available(self.upstream, 'master', rev)
 154
 155     def test_ensure_other_branch(self) -> None:
 156         _git(self.upstream, 'checkout', '-b', 'otherbranch')
 157         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 158         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 159         d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev)
 160         self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
 161
 162     def test_catch_up(self) -> None:
 163         _git(self.upstream, 'checkout', '-b', 'otherbranch')
 164         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 165         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 166         d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev)
 167         self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
 168         _git(self.upstream, 'checkout', 'master')
 169         _git(self.upstream, 'merge', '--ff-only', 'otherbranch')
 170         d = git_cache.ensure_rev_available(self.upstream, 'master', rev)
 171         self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
 172
 173     def test_fetch_after_cache_deleted(self) -> None:
 174         d1, rev1 = git_cache.fetch(self.upstream, 'master')
 175         shutil.rmtree(d1)
 176         d2, rev2 = git_cache.fetch(self.upstream, 'master')
 177         self.assertEqual(d1, d2)
 178         self.assertEqual(rev1, rev2)
 179         self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
 180
 181     def test_ensure_after_cache_deleted(self) -> None:
 182         d1, rev = git_cache.fetch(self.upstream, 'master')
 183         shutil.rmtree(d1)
 184         d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
 185         self.assertEqual(d1, d2)
 186         self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')
 187
 188     def test_fetch_raises_on_amend(self) -> None:
 189         git_cache.fetch(self.upstream, 'master')
 190         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 191         with self.assertRaises(Exception):
 192             git_cache.fetch(self.upstream, 'master')
 193
 194     def test_ensure_raises_on_amend(self) -> None:
 195         git_cache.fetch(self.upstream, 'master')
 196         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 197         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 198         with self.assertRaises(Exception):
 199             git_cache.ensure_rev_available(self.upstream, 'master', rev)
 200
 201
 202 if __name__ == '__main__':
 203     unittest.main()