]> git.scottworley.com Git - git-cache/blob - test_git_cache.py
bbe71188272c9d13254c5b4f4674bb6a4e43001f
[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(prefix='git_cache_test-')
41 self.old_XDG_CACHE_HOME = os.environ.get('XDG_CACHE_HOME')
42 _setenv('XDG_CACHE_HOME', self.xdgcache.name)
43
44 os.environ['GIT_AUTHOR_NAME'] = 'test_git_cache'
45 os.environ['GIT_COMMITTER_NAME'] = 'test_git_cache'
46 os.environ['GIT_AUTHOR_EMAIL'] = 'test_git_cache@example.com'
47 os.environ['GIT_COMMITTER_EMAIL'] = 'test_git_cache@example.com'
48
49 os.environ['BACKOFF_MAX_TIME'] = '0'
50
51 self.tempdir = tempfile.TemporaryDirectory(prefix='git_cache_test-')
52 self.upstream = os.path.join(self.tempdir.name, 'upstream')
53 subprocess.run(['git', 'init', self.upstream], check=True)
54 _commit_file(self.upstream, 'file', 'Contents', 'First commit')
55
56 def tearDown(self) -> None:
57 _setenv('XDG_CACHE_HOME', self.old_XDG_CACHE_HOME)
58
59 self.tempdir.cleanup()
60 self.xdgcache.cleanup()
61
62 def test_fetch(self) -> None:
63 d, rev = git_cache.fetch(self.upstream, 'master')
64 self.assertEqual(_git(d, 'show', '%s:file' % rev), b'Contents')
65
66 def test_fetch_twice(self) -> None:
67 d1, rev1 = git_cache.fetch(self.upstream, 'master')
68 self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents')
69 d2, rev2 = git_cache.fetch(self.upstream, 'master')
70 self.assertEqual(d1, d2)
71 self.assertEqual(rev1, rev2)
72 self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
73
74 def test_fetch_then_ensure(self) -> None:
75 d1, rev = git_cache.fetch(self.upstream, 'master')
76 self.assertEqual(_git(d1, 'show', '%s:file' % rev), b'Contents')
77 d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
78 self.assertEqual(d1, d2)
79 self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')
80
81 def test_ensure_then_fetch(self) -> None:
82 rev1 = _git(
83 self.upstream, 'log', '--format=%H', '-n1').strip().decode()
84 d1 = git_cache.ensure_rev_available(self.upstream, 'master', rev1)
85 self.assertEqual(_git(d1, 'show', '%s:file' % rev1), b'Contents')
86 d2, rev2 = git_cache.fetch(self.upstream, 'master')
87 self.assertEqual(d1, d2)
88 self.assertEqual(rev1, rev2)
89 self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
90
91 def test_fetch_new_file(self) -> None:
92 d1, rev1 = git_cache.fetch(self.upstream, 'master')
93 _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
94 d2, rev2 = git_cache.fetch(self.upstream, 'master')
95 self.assertEqual(d1, d2)
96 self.assertNotEqual(rev1, rev2)
97 self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')
98
99 def test_ensure_doesnt_fetch_new_file(self) -> None:
100 d1, rev1 = git_cache.fetch(self.upstream, 'master')
101 _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
102 rev2 = _git(
103 self.upstream, 'log', '--format=%H', '-n1').strip().decode()
104 self.assertNotEqual(rev1, rev2)
105 d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev1)
106 self.assertEqual(d1, d2)
107 p = subprocess.run(
108 ['git', '-C', d2, 'show', '%s:foofile' % rev2], check=False)
109 self.assertNotEqual(p.returncode, 0)
110
111 def test_ensure_doesnt_fetch_from_deleted_upstream(self) -> None:
112 d1, rev = git_cache.fetch(self.upstream, 'master')
113 self.tempdir.cleanup()
114 d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
115 self.assertEqual(d1, d2)
116
117 def test_ensure_fetches_new_file(self) -> None:
118 d1, rev1 = git_cache.fetch(self.upstream, 'master')
119 _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
120 rev2 = _git(
121 self.upstream, 'log', '--format=%H', '-n1').strip().decode()
122 self.assertNotEqual(rev1, rev2)
123 d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev2)
124 self.assertEqual(d1, d2)
125 self.assertEqual(_git(d2, 'show', '%s:foofile' % rev2), b'foo')
126
127 def test_fetch_raises_on_invalid_repo(self) -> None:
128 self.tempdir.cleanup()
129 with self.assertRaises(Exception):
130 git_cache.fetch(self.upstream, 'master')
131
132 def test_ensure_raises_on_invalid_repo(self) -> None:
133 rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
134 self.tempdir.cleanup()
135 with self.assertRaises(Exception):
136 git_cache.ensure_rev_available(self.upstream, 'master', rev)
137
138 def test_fetch_raises_on_invalid_ref(self) -> None:
139 with self.assertRaises(Exception):
140 git_cache.fetch(self.upstream, 'nobranch')
141
142 def test_ensure_raises_on_invalid_ref(self) -> None:
143 rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
144 with self.assertRaises(Exception):
145 git_cache.ensure_rev_available(self.upstream, 'nobranch', rev)
146
147 def test_ensure_raises_on_invalid_rev(self) -> None:
148 with self.assertRaises(Exception):
149 git_cache.ensure_rev_available(
150 self.upstream,
151 'nobranch',
152 '1234567890abcdef01234567890abcdef1234567')
153
154 def test_ensure_raises_on_rev_from_other_branch(self) -> None:
155 _git(self.upstream, 'checkout', '-b', 'otherbranch')
156 _commit_file(self.upstream, 'foofile', 'foo', 'Foo')
157 rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
158 with self.assertRaises(Exception):
159 git_cache.ensure_rev_available(self.upstream, 'master', rev)
160
161 def test_ensure_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 d = git_cache.ensure_rev_available(self.upstream, 'otherbranch', rev)
166 self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
167
168 def test_catch_up(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 _git(self.upstream, 'checkout', 'master')
175 _git(self.upstream, 'merge', '--ff-only', 'otherbranch')
176 d = git_cache.ensure_rev_available(self.upstream, 'master', rev)
177 self.assertEqual(_git(d, 'show', '%s:foofile' % rev), b'foo')
178
179 def test_fetch_after_cache_deleted(self) -> None:
180 d1, rev1 = git_cache.fetch(self.upstream, 'master')
181 shutil.rmtree(d1)
182 d2, rev2 = git_cache.fetch(self.upstream, 'master')
183 self.assertEqual(d1, d2)
184 self.assertEqual(rev1, rev2)
185 self.assertEqual(_git(d2, 'show', '%s:file' % rev2), b'Contents')
186
187 def test_ensure_after_cache_deleted(self) -> None:
188 d1, rev = git_cache.fetch(self.upstream, 'master')
189 shutil.rmtree(d1)
190 d2 = git_cache.ensure_rev_available(self.upstream, 'master', rev)
191 self.assertEqual(d1, d2)
192 self.assertEqual(_git(d2, 'show', '%s:file' % rev), b'Contents')
193
194 def test_fetch_raises_on_amend(self) -> None:
195 git_cache.fetch(self.upstream, 'master')
196 _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
197 with self.assertRaises(Exception):
198 git_cache.fetch(self.upstream, 'master')
199
200 def test_ensure_raises_on_amend(self) -> None:
201 git_cache.fetch(self.upstream, 'master')
202 _git(self.upstream, 'commit', '--amend', '-m', 'Amended')
203 rev = _git(self.upstream, 'log', '--format=%H', '-n1').strip().decode()
204 with self.assertRaises(Exception):
205 git_cache.ensure_rev_available(self.upstream, 'master', rev)
206
207
208 if __name__ == '__main__':
209 unittest.main()