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', encoding='utf-8') 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(  # pylint: disable=consider-using-with
  41             prefix='git_cache_test-cache-')
  42         self.xdgdata = tempfile.TemporaryDirectory(  # pylint: disable=consider-using-with
  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(  # pylint: disable=consider-using-with
  58             prefix='git_cache_test-')
  59         self.upstream = os.path.join(self.tempdir.name, 'upstream')
  60         subprocess.run(['git', '-c', 'init.defaultBranch=main',
  61                        'init', self.upstream], check=True)
  62         _commit_file(self.upstream, 'file', 'Contents', 'First commit')
  63
  64     def tearDown(self) -> None:
  65         _setenv('XDG_CACHE_HOME', self.old_XDG_CACHE_HOME)
  66         _setenv('XDG_DATA_HOME', self.old_XDG_DATA_HOME)
  67
  68         self.tempdir.cleanup()
  69         self.xdgcache.cleanup()
  70
  71     def test_fetch(self) -> None:
  72         d, rev = git_cache.fetch(self.upstream, 'main')
  73         self.assertEqual(_git(d, 'show', f'{rev}:file'), b'Contents')
  74
  75     def test_fetch_twice(self) -> None:
  76         d1, rev1 = git_cache.fetch(self.upstream, 'main')
  77         self.assertEqual(_git(d1, 'show', f'{rev1}:file'), b'Contents')
  78         d2, rev2 = git_cache.fetch(self.upstream, 'main')
  79         self.assertEqual(d1, d2)
  80         self.assertEqual(rev1, rev2)
  81         self.assertEqual(_git(d2, 'show', f'{rev2}:file'), b'Contents')
  82
  83     def test_fetch_then_ensure(self) -> None:
  84         d1, rev = git_cache.fetch(self.upstream, 'main')
  85         self.assertEqual(_git(d1, 'show', f'{rev}:file'), b'Contents')
  86         d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev)
  87         self.assertEqual(d1, d2)
  88         self.assertEqual(_git(d2, 'show', f'{rev}:file'), b'Contents')
  89
  90     def test_ensure_then_fetch(self) -> None:
  91         rev1 = _git(
  92             self.upstream, 'log', '--format=%H', '-n1').strip().decode()
  93         d1 = git_cache.ensure_rev_available(self.upstream, 'main', rev1)
  94         self.assertEqual(_git(d1, 'show', f'{rev1}:file'), b'Contents')
  95         d2, rev2 = git_cache.fetch(self.upstream, 'main')
  96         self.assertEqual(d1, d2)
  97         self.assertEqual(rev1, rev2)
  98         self.assertEqual(_git(d2, 'show', f'{rev2}:file'), b'Contents')
  99
 100     def test_fetch_new_file(self) -> None:
 101         d1, rev1 = git_cache.fetch(self.upstream, 'main')
 102         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 103         d2, rev2 = git_cache.fetch(self.upstream, 'main')
 104         self.assertEqual(d1, d2)
 105         self.assertNotEqual(rev1, rev2)
 106         self.assertEqual(_git(d2, 'show', f'{rev2}:foofile'), b'foo')
 107
 108     def test_ensure_doesnt_fetch_new_file(self) -> None:
 109         d1, rev1 = git_cache.fetch(self.upstream, 'main')
 110         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 111         rev2 = _git(
 112             self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 113         self.assertNotEqual(rev1, rev2)
 114         d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev1)
 115         self.assertEqual(d1, d2)
 116         p = subprocess.run(
 117             ['git', '-C', d2, 'show', f'{rev2}:foofile'], check=False)
 118         self.assertNotEqual(p.returncode, 0)
 119
 120     def test_ensure_doesnt_fetch_from_deleted_upstream(self) -> None:
 121         d1, rev = git_cache.fetch(self.upstream, 'main')
 122         self.tempdir.cleanup()
 123         d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev)
 124         self.assertEqual(d1, d2)
 125
 126     def test_ensure_fetches_new_file(self) -> None:
 127         d1, rev1 = git_cache.fetch(self.upstream, 'main')
 128         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 129         rev2 = _git(
 130             self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 131         self.assertNotEqual(rev1, rev2)
 132         d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev2)
 133         self.assertEqual(d1, d2)
 134         self.assertEqual(_git(d2, 'show', f'{rev2}:foofile'), b'foo')
 135
 136     def test_fetch_raises_on_invalid_repo(self) -> None:
 137         self.tempdir.cleanup()
 138         with self.assertRaises(Exception):
 139             git_cache.fetch(self.upstream, 'main')
 140
 141     def test_ensure_raises_on_invalid_repo(self) -> None:
 142         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 143         self.tempdir.cleanup()
 144         with self.assertRaises(Exception):
 145             git_cache.ensure_rev_available(self.upstream, 'main', rev)
 146
 147     def test_fetch_raises_on_invalid_ref(self) -> None:
 148         with self.assertRaises(Exception):
 149             git_cache.fetch(self.upstream, 'nobranch')
 150
 151     def test_ensure_raises_on_invalid_ref(self) -> None:
 152         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 153         with self.assertRaises(Exception):
 154             git_cache.ensure_rev_available(self.upstream, 'nobranch', rev)
 155
 156     def test_ensure_raises_on_invalid_rev(self) -> None:
 157         with self.assertRaises(Exception):
 158             git_cache.ensure_rev_available(
 159                 self.upstream,
 160                 'nobranch',
 161                 '1234567890abcdef01234567890abcdef1234567')
 162
 163     def test_ensure_raises_on_rev_from_other_branch(self) -> None:
 164         _git(self.upstream, 'checkout', '-b', 'otherbranch')
 165         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 166         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 167         with self.assertRaises(Exception):
 168             git_cache.ensure_rev_available(self.upstream, 'main', rev)
 169
 170     def test_ensure_other_branch(self) -> None:
 171         _git(self.upstream, 'checkout', '-b', 'otherbranch')
 172         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 173         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 174         d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev)
 175         self.assertEqual(_git(d, 'show', f'{rev}:foofile'), b'foo')
 176
 177     def test_catch_up(self) -> None:
 178         _git(self.upstream, 'checkout', '-b', 'otherbranch')
 179         _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
 180         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 181         d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev)
 182         self.assertEqual(_git(d, 'show', f'{rev}:foofile'), b'foo')
 183         _git(self.upstream, 'checkout', 'main')
 184         _git(self.upstream, 'merge', '--ff-only', 'otherbranch')
 185         d = git_cache.ensure_rev_available(self.upstream, 'main', rev)
 186         self.assertEqual(_git(d, 'show', f'{rev}:foofile'), b'foo')
 187
 188     def test_fetch_after_cache_deleted(self) -> None:
 189         d1, rev1 = git_cache.fetch(self.upstream, 'main')
 190         shutil.rmtree(d1)
 191         d2, rev2 = git_cache.fetch(self.upstream, 'main')
 192         self.assertEqual(d1, d2)
 193         self.assertEqual(rev1, rev2)
 194         self.assertEqual(_git(d2, 'show', f'{rev2}:file'), b'Contents')
 195
 196     def test_ensure_after_cache_deleted(self) -> None:
 197         d1, rev = git_cache.fetch(self.upstream, 'main')
 198         shutil.rmtree(d1)
 199         d2 = git_cache.ensure_rev_available(self.upstream, 'main', rev)
 200         self.assertEqual(d1, d2)
 201         self.assertEqual(_git(d2, 'show', f'{rev}:file'), b'Contents')
 202
 203     def test_fetch_raises_on_amend(self) -> None:
 204         git_cache.fetch(self.upstream, 'main')
 205         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 206         with self.assertRaises(Exception):
 207             git_cache.fetch(self.upstream, 'main')
 208
 209     def test_ensure_raises_on_amend(self) -> None:
 210         git_cache.fetch(self.upstream, 'main')
 211         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 212         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 213         with self.assertRaises(Exception):
 214             git_cache.ensure_rev_available(self.upstream, 'main', rev)
 215
 216     def test_fetch_raises_on_amend_after_cache_deleted(self) -> None:
 217         d, _ = git_cache.fetch(self.upstream, 'main')
 218         shutil.rmtree(d)
 219         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 220         with self.assertRaises(Exception):
 221             git_cache.fetch(self.upstream, 'main')
 222
 223     def test_ensure_raises_on_amend_after_cache_deleted(self) -> None:
 224         d, _ = git_cache.fetch(self.upstream, 'main')
 225         shutil.rmtree(d)
 226         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 227         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 228         with self.assertRaises(Exception):
 229             git_cache.ensure_rev_available(self.upstream, 'main', rev)
 230
 231     def test_force_fetch_after_amend(self) -> None:
 232         git_cache.fetch(self.upstream, 'main')
 233         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 234         git_cache.fetch(self.upstream, 'main', force=True)
 235
 236     def test_force_ensure_after_amend(self) -> None:
 237         git_cache.fetch(self.upstream, 'main')
 238         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 239         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 240         git_cache.ensure_rev_available(
 241             self.upstream, 'main', rev, force=True)
 242
 243     def test_force_fetch_after_amend_and_cache_delete(self) -> None:
 244         d, _ = git_cache.fetch(self.upstream, 'main')
 245         shutil.rmtree(d)
 246         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 247         git_cache.fetch(self.upstream, 'main', force=True)
 248
 249     def test_force_ensure_after_amend_and_cache_delete(self) -> None:
 250         d, _ = git_cache.fetch(self.upstream, 'main')
 251         shutil.rmtree(d)
 252         _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
 253         rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
 254         git_cache.ensure_rev_available(
 255             self.upstream, 'main', rev, force=True)
 256
 257
 258 if __name__ == '__main__':
 259     unittest.main()