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 from typing import Optional
   8
   9 import git_cache
  10
  11
  12 def _setenv(var: str, value: Optional[str]) -> None:
  13     if value is None:
  14         del os.environ[var]
  15     else:
  16         os.environ[var] = value
  17
  18
  19 def _git(directory: str, *args: str) -> bytes:
  20     p = subprocess.run(['git', '-C', directory] + list(args),
  21                        stdout=subprocess.PIPE, check=True)
  22     return bytes(p.stdout)
  23
  24
  25 def _commit_file(
  26         directory: str,
  27         filename: str,
  28         contents: str,
  29         commit_message: str) -> None:
  30     with open(os.path.join(directory, filename), 'w') as f:
  31         f.write(contents)
  32     _git(directory, 'add', filename)
  33     _git(directory, 'commit', '-m', commit_message)
  34
  35
  36 # pylint: disable=too-many-public-methods
  37 class TestGitCache(unittest.TestCase):
  38
  39     def setUp(self) -> None:
  40         self.xdgcache = tempfile.TemporaryDirectory(
  41             prefix='git_cache_test-cache-')
  42         self.xdgdata = tempfile.TemporaryDirectory(
  43             prefix='git_cache_test-data-')
  44         self.old_XDG_CACHE_HOME = os.environ.get('XDG_CACHE_HOME')
  45         self.old_XDG_DATA_HOME = os.environ.get('XDG_DATA_HOME')
  46         _setenv('XDG_CACHE_HOME', self.xdgcache.name)
  47         _setenv('XDG_DATA_HOME', self.xdgdata.name)
  48
  49         os.environ['GIT_AUTHOR_NAME'] = 'test_git_cache'
  50         os.environ['GIT_COMMITTER_NAME'] = 'test_git_cache'
  51         os.environ['GIT_AUTHOR_EMAIL'] = 'test_git_cache@example.com'
  52         os.environ['GIT_COMMITTER_EMAIL'] = 'test_git_cache@example.com'
  53
  54         os.environ['BACKOFF_MAX_TIME'] = '0'
  55         os.environ['FORCE_WARNING_TIME'] = '0'  # ONLY FOR TEST USE!
  56
  57         self.tempdir = tempfile.TemporaryDirectory(prefix='git_cache_test-')
  58         self.upstream = os.path.join(self.tempdir.name, 'upstream')
  59         subprocess.run(['git', 'init', self.upstream], check=True)
  60         _commit_file(self.upstream, 'file', 'Contents', 'First commit')
  61
  62     def tearDown(self) -> None:
  63         _setenv('XDG_CACHE_HOME', self.old_XDG_CACHE_HOME)
  64         _setenv('XDG_DATA_HOME', self.old_XDG_DATA_HOME)
  65
  66         self.tempdir.cleanup()
  67         self.xdgcache.cleanup()
  68
  69     def test_fetch(self) -> None:
  70         d, rev = git_cache.fetch(self.upstream, 'master')
  71         self.assertEqual(_git(d, 'show', '%s:file' % rev), b'Contents')
  72
  73     def test_fetch_twice(self) -> None:
  74         d1, rev1 = git_cache.fetch(self.upstream, 'master')
  75         self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents')
  76         d2, rev2 = git_cache.fetch(self.upstream, 'master')
  77         self.assertEqual(d1, d2)
  78         self.assertEqual(rev1, rev2)
  79         self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
  80
  81     def test_fetch_then_ensure(self) -> None:
  82         d1, rev = git_cache.fetch(self.upstream, 'master')
  83         self.assertEqual(_git(d1, 'show', '%s:file' % rev), b'Contents')
  84         d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
  85         self.assertEqual(d1, d2)
  86         self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')
  87
  88     def test_ensure_then_fetch(self) -> None:
  89         rev1 = _git(
  90             self.upstream, 'log', '--format=%H', '-n1').strip().decode()
  91         d1 = git_cache.ensure_rev_available(self.upstream, 'master', rev1)
  92         self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents')
  93         d2, rev2 = git_cache.fetch(self.upstream, 'master')
  94         self.assertEqual(d1, d2)
  95         self.assertEqual(rev1, rev2)
  96         self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
  97
  98     def test_fetch_new_file(self) -> None:
  99         d1, rev1 = git_cache.fetch(self.upstream, 'master')
 100         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 101         d2, rev2 = git_cache.fetch(self.upstream, 'master')
 102         self.assertEqual(d1, d2)
 103         self.assertNotEqual(rev1, rev2)
 104         self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')
 105
 106     def test_ensure_doesnt_fetch_new_file(self) -> None:
 107         d1, rev1 = git_cache.fetch(self.upstream, 'master')
 108         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 109         rev2 = _git(
 110             self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 111         self.assertNotEqual(rev1, rev2)
 112         d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev1)
 113         self.assertEqual(d1, d2)
 114         p = subprocess.run(
 115             ['git', '-C', d2, 'show', '%s:foofile' % rev2], check=False)
 116         self.assertNotEqual(p.returncode, 0)
 117
 118     def test_ensure_doesnt_fetch_from_deleted_upstream(self) -> None:
 119         d1, rev = git_cache.fetch(self.upstream, 'master')
 120         self.tempdir.cleanup()
 121         d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
 122         self.assertEqual(d1, d2)
 123
 124     def test_ensure_fetches_new_file(self) -> None:
 125         d1, rev1 = git_cache.fetch(self.upstream, 'master')
 126         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 127         rev2 = _git(
 128             self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 129         self.assertNotEqual(rev1, rev2)
 130         d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev2)
 131         self.assertEqual(d1, d2)
 132         self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')
 133
 134     def test_fetch_raises_on_invalid_repo(self) -> None:
 135         self.tempdir.cleanup()
 136         with self.assertRaises(Exception):
 137             git_cache.fetch(self.upstream, 'master')
 138
 139     def test_ensure_raises_on_invalid_repo(self) -> None:
 140         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 141         self.tempdir.cleanup()
 142         with self.assertRaises(Exception):
 143             git_cache.ensure_rev_available(self.upstream, 'master', rev)
 144
 145     def test_fetch_raises_on_invalid_ref(self) -> None:
 146         with self.assertRaises(Exception):
 147             git_cache.fetch(self.upstream, 'nobranch')
 148
 149     def test_ensure_raises_on_invalid_ref(self) -> None:
 150         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 151         with self.assertRaises(Exception):
 152             git_cache.ensure_rev_available(self.upstream, 'nobranch', rev)
 153
 154     def test_ensure_raises_on_invalid_rev(self) -> None:
 155         with self.assertRaises(Exception):
 156             git_cache.ensure_rev_available(
 157                 self.upstream,
 158                 'nobranch',
 159                 '1234567890abcdef01234567890abcdef1234567')
 160
 161     def test_ensure_raises_on_rev_from_other_branch(self) -> None:
 162         _git(self.upstream, 'checkout', '-b', 'otherbranch')
 163         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 164         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 165         with self.assertRaises(Exception):
 166             git_cache.ensure_rev_available(self.upstream, 'master', rev)
 167
 168     def test_ensure_other_branch(self) -> None:
 169         _git(self.upstream, 'checkout', '-b', 'otherbranch')
 170         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 171         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 172         d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev)
 173         self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
 174
 175     def test_catch_up(self) -> None:
 176         _git(self.upstream, 'checkout', '-b', 'otherbranch')
 177         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 178         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 179         d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev)
 180         self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
 181         _git(self.upstream, 'checkout', 'master')
 182         _git(self.upstream, 'merge', '--ff-only', 'otherbranch')
 183         d = git_cache.ensure_rev_available(self.upstream, 'master', rev)
 184         self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
 185
 186     def test_fetch_after_cache_deleted(self) -> None:
 187         d1, rev1 = git_cache.fetch(self.upstream, 'master')
 188         shutil.rmtree(d1)
 189         d2, rev2 = git_cache.fetch(self.upstream, 'master')
 190         self.assertEqual(d1, d2)
 191         self.assertEqual(rev1, rev2)
 192         self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
 193
 194     def test_ensure_after_cache_deleted(self) -> None:
 195         d1, rev = git_cache.fetch(self.upstream, 'master')
 196         shutil.rmtree(d1)
 197         d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
 198         self.assertEqual(d1, d2)
 199         self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')
 200
 201     def test_fetch_raises_on_amend(self) -> None:
 202         git_cache.fetch(self.upstream, 'master')
 203         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 204         with self.assertRaises(Exception):
 205             git_cache.fetch(self.upstream, 'master')
 206
 207     def test_ensure_raises_on_amend(self) -> None:
 208         git_cache.fetch(self.upstream, 'master')
 209         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 210         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 211         with self.assertRaises(Exception):
 212             git_cache.ensure_rev_available(self.upstream, 'master', rev)
 213
 214     def test_fetch_raises_on_amend_after_cache_deleted(self) -> None:
 215         d, _ = git_cache.fetch(self.upstream, 'master')
 216         shutil.rmtree(d)
 217         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 218         with self.assertRaises(Exception):
 219             git_cache.fetch(self.upstream, 'master')
 220
 221     def test_ensure_raises_on_amend_after_cache_deleted(self) -> None:
 222         d, _ = git_cache.fetch(self.upstream, 'master')
 223         shutil.rmtree(d)
 224         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 225         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 226         with self.assertRaises(Exception):
 227             git_cache.ensure_rev_available(self.upstream, 'master', rev)
 228
 229     def test_force_fetch_after_amend(self) -> None:
 230         git_cache.fetch(self.upstream, 'master')
 231         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 232         git_cache.fetch(self.upstream, 'master', force=True)
 233
 234     def test_force_ensure_after_amend(self) -> None:
 235         git_cache.fetch(self.upstream, 'master')
 236         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 237         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 238         git_cache.ensure_rev_available(
 239             self.upstream, 'master', rev, force=True)
 240
 241     def test_force_fetch_after_amend_and_cache_delete(self) -> None:
 242         d, _ = git_cache.fetch(self.upstream, 'master')
 243         shutil.rmtree(d)
 244         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 245         git_cache.fetch(self.upstream, 'master', force=True)
 246
 247     def test_force_ensure_after_amend_and_cache_delete(self) -> None:
 248         d, _ = git_cache.fetch(self.upstream, 'master')
 249         shutil.rmtree(d)
 250         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 251         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 252         git_cache.ensure_rev_available(
 253             self.upstream, 'master', rev, force=True)
 254
 255
 256 if __name__ == '__main__':
 257     unittest.main()