]> git.scottworley.com Git - git-cache/blob - test_git_cache.py
529ac94334e97ad9197e6e073f590689f25e7994
[git-cache] / 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()