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