]> git.scottworley.com Git - git-cache/blob - test_git_cache.py
64151fe75b45766af8efa2a1bb9cce742907fbaa
[git-cache] / 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_fetch_after_cache_deleted(self) -> None:
163 d1, rev1 = git_cache.fetch(self.upstream, 'master')
164 shutil.rmtree(d1)
165 d2, rev2 = git_cache.fetch(self.upstream, 'master')
166 self.assertEqual(d1, d2)
167 self.assertEqual(rev1, rev2)
168 self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
169
170 def test_ensure_after_cache_deleted(self) -> None:
171 d1, rev = git_cache.fetch(self.upstream, 'master')
172 shutil.rmtree(d1)
173 d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
174 self.assertEqual(d1, d2)
175 self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')
176
177 def test_fetch_raises_on_amend(self) -> None:
178 git_cache.fetch(self.upstream, 'master')
179 _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
180 with self.assertRaises(Exception):
181 git_cache.fetch(self.upstream, 'master')
182
183 def test_ensure_raises_on_amend(self) -> None:
184 git_cache.fetch(self.upstream, 'master')
185 _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
186 rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
187 with self.assertRaises(Exception):
188 git_cache.ensure_rev_available(self.upstream, 'master', rev)
189
190
191 if __name__ == '__main__':
192 unittest.main()