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