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

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