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