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