2 from __future__ import absolute_import, print_function
3 from binascii import unhexlify
4 from errno import ELOOP, ENOTDIR
6 from stat import S_IFDIR
8 from time import localtime, strftime
10 from wvpytest import *
12 from bup import git, path, vfs
13 from bup.compat import environ
14 from bup.repo import LocalRepo, RemoteRepo
15 from buptest import ex, exo
16 from buptest.vfs import tree_dict
20 ## The clear_cache() calls below are to make sure that the test starts
21 ## from a known state since at the moment the cache entry for a given
22 ## item (like a commit) can change. For example, its meta value might
23 ## be promoted from a mode to a Metadata instance once the tree it
24 ## refers to is traversed.
26 def prep_and_test_repo(tmpdir, create_repo, test_repo):
27 bup_dir = tmpdir + b'/bup'
28 environ[b'GIT_DIR'] = bup_dir
29 environ[b'BUP_DIR'] = bup_dir
30 ex((bup_path, b'init'))
32 with create_repo(bup_dir) as repo:
33 test_repo(repo, tmpdir)
35 # Currently, we just test through the repos since LocalRepo resolve is
36 # just a straight redirection to vfs.resolve.
38 def _test_resolve(repo, tmpdir):
39 data_path = tmpdir + b'/src'
40 resolve = repo.resolve
42 save_time_str = strftime('%Y-%m-%d-%H%M%S', localtime(save_time)).encode('ascii')
44 os.mkdir(data_path + b'/dir')
45 with open(data_path + b'/file', 'wb+') as tmpfile:
46 tmpfile.write(b'canary\n')
47 symlink(b'file', data_path + b'/file-symlink')
48 symlink(b'dir', data_path + b'/dir-symlink')
49 symlink(b'not-there', data_path + b'/bad-symlink')
50 ex((bup_path, b'index', b'-v', data_path))
51 ex((bup_path, b'save', b'-d', b'%d' % save_time, b'-tvvn', b'test',
52 b'--strip', data_path))
53 ex((bup_path, b'tag', b'test-tag', b'test'))
55 tip_hash = exo((b'git', b'show-ref', b'refs/heads/test'))[0]
56 tip_oidx = tip_hash.strip().split()[0]
57 tip_oid = unhexlify(tip_oidx)
58 tip_tree_oidx = exo((b'git', b'log', b'--pretty=%T', b'-n1',
60 tip_tree_oid = unhexlify(tip_tree_oidx)
61 tip_tree = tree_dict(repo, tip_tree_oid)
62 test_revlist_w_meta = vfs.RevList(meta=tip_tree[b'.'].meta,
64 expected_latest_item = vfs.Commit(meta=S_IFDIR | 0o755,
67 expected_latest_item_w_meta = vfs.Commit(meta=tip_tree[b'.'].meta,
70 expected_latest_link = vfs.FakeLink(meta=vfs.default_symlink_mode,
72 expected_test_tag_item = expected_latest_item
77 wvpasseq(((b'', vfs._root),), res)
78 ignore, root_item = res[0]
79 root_content = frozenset(vfs.contents(repo, root_item))
80 wvpasseq(frozenset([(b'.', root_item),
82 (b'test', test_revlist_w_meta)]),
84 for path in (b'//', b'/.', b'/./', b'/..', b'/../',
85 b'/test/latest/dir/../../..',
86 b'/test/latest/dir/../../../',
87 b'/test/latest/dir/../../../.',
88 b'/test/latest/dir/../../..//',
89 b'/test//latest/dir/../../..',
90 b'/test/./latest/dir/../../..',
91 b'/test/././latest/dir/../../..',
92 b'/test/.//./latest/dir/../../..',
93 b'/test//.//.//latest/dir/../../..'
94 b'/test//./latest/dir/../../..'):
97 wvpasseq(((b'', vfs._root),), res)
100 res = resolve(b'/.tag')
101 wvpasseq(2, len(res))
102 wvpasseq(((b'', vfs._root), (b'.tag', vfs._tags)),
104 ignore, tag_item = res[1]
105 tag_content = frozenset(vfs.contents(repo, tag_item))
106 wvpasseq(frozenset([(b'.', tag_item),
107 (b'test-tag', expected_test_tag_item)]),
111 res = resolve(b'/test')
112 wvpasseq(2, len(res))
113 wvpasseq(((b'', vfs._root), (b'test', test_revlist_w_meta)), res)
114 ignore, test_item = res[1]
115 test_content = frozenset(vfs.contents(repo, test_item))
116 # latest has metadata here due to caching
117 wvpasseq(frozenset([(b'.', test_revlist_w_meta),
118 (save_time_str, expected_latest_item_w_meta),
119 (b'latest', expected_latest_link)]),
123 res = resolve(b'/test/latest')
124 wvpasseq(3, len(res))
125 expected_latest_item_w_meta = vfs.Commit(meta=tip_tree[b'.'].meta,
128 expected = ((b'', vfs._root),
129 (b'test', test_revlist_w_meta),
130 (save_time_str, expected_latest_item_w_meta))
131 wvpasseq(expected, res)
132 ignore, latest_item = res[2]
133 latest_content = frozenset(vfs.contents(repo, latest_item))
134 expected = frozenset((x.name, vfs.Item(oid=x.oid, meta=x.meta))
135 for x in (tip_tree[name]
142 wvpasseq(expected, latest_content)
145 res = resolve(b'/test/latest/file')
146 wvpasseq(4, len(res))
147 expected_file_item_w_meta = vfs.Item(meta=tip_tree[b'file'].meta,
148 oid=tip_tree[b'file'].oid)
149 expected = ((b'', vfs._root),
150 (b'test', test_revlist_w_meta),
151 (save_time_str, expected_latest_item_w_meta),
152 (b'file', expected_file_item_w_meta))
153 wvpasseq(expected, res)
156 res = resolve(b'/test/latest/bad-symlink')
157 wvpasseq(4, len(res))
158 expected = ((b'', vfs._root),
159 (b'test', test_revlist_w_meta),
160 (save_time_str, expected_latest_item_w_meta),
161 (b'not-there', None))
162 wvpasseq(expected, res)
165 res = resolve(b'/test/latest/bad-symlink', follow=False)
166 wvpasseq(4, len(res))
167 bad_symlink_value = tip_tree[b'bad-symlink']
168 expected_bad_symlink_item_w_meta = vfs.Item(meta=bad_symlink_value.meta,
169 oid=bad_symlink_value.oid)
170 expected = ((b'', vfs._root),
171 (b'test', test_revlist_w_meta),
172 (save_time_str, expected_latest_item_w_meta),
173 (b'bad-symlink', expected_bad_symlink_item_w_meta))
174 wvpasseq(expected, res)
177 res = resolve(b'/test/latest/file-symlink')
178 wvpasseq(4, len(res))
179 expected = ((b'', vfs._root),
180 (b'test', test_revlist_w_meta),
181 (save_time_str, expected_latest_item_w_meta),
182 (b'file', expected_file_item_w_meta))
183 wvpasseq(expected, res)
186 res = resolve(b'/test/latest/file-symlink', follow=False)
187 wvpasseq(4, len(res))
188 file_symlink_value = tip_tree[b'file-symlink']
189 expected_file_symlink_item_w_meta = vfs.Item(meta=file_symlink_value.meta,
190 oid=file_symlink_value.oid)
191 expected = ((b'', vfs._root),
192 (b'test', test_revlist_w_meta),
193 (save_time_str, expected_latest_item_w_meta),
194 (b'file-symlink', expected_file_symlink_item_w_meta))
195 wvpasseq(expected, res)
198 res = resolve(b'/test/latest/missing')
199 wvpasseq(4, len(res))
201 wvpasseq(b'missing', name)
204 for path in (b'/test/latest/file/',
205 b'/test/latest/file/.',
206 b'/test/latest/file/..',
207 b'/test/latest/file/../',
208 b'/test/latest/file/../.',
209 b'/test/latest/file/../..',
210 b'/test/latest/file/foo'):
214 except vfs.IOError as res_ex:
215 wvpasseq(ENOTDIR, res_ex.errno)
216 wvpasseq([b'', b'test', save_time_str, b'file'],
217 [name for name, item in res_ex.terminus])
219 for path in (b'/test/latest/file-symlink/',
220 b'/test/latest/file-symlink/.',
221 b'/test/latest/file-symlink/..',
222 b'/test/latest/file-symlink/../',
223 b'/test/latest/file-symlink/../.',
224 b'/test/latest/file-symlink/../..'):
227 resolve(path, follow=False)
228 except vfs.IOError as res_ex:
229 wvpasseq(ENOTDIR, res_ex.errno)
230 wvpasseq([b'', b'test', save_time_str, b'file'],
231 [name for name, item in res_ex.terminus])
234 file_res = resolve(b'/test/latest/file')
236 resolve(b'foo', parent=file_res)
237 except vfs.IOError as res_ex:
238 wvpasseq(ENOTDIR, res_ex.errno)
239 wvpasseq(None, res_ex.terminus)
242 res = resolve(b'/test/latest/dir-symlink', follow=False)
243 wvpasseq(4, len(res))
244 dir_symlink_value = tip_tree[b'dir-symlink']
245 expected_dir_symlink_item_w_meta = vfs.Item(meta=dir_symlink_value.meta,
246 oid=dir_symlink_value.oid)
247 expected = ((b'', vfs._root),
248 (b'test', test_revlist_w_meta),
249 (save_time_str, expected_latest_item_w_meta),
250 (b'dir-symlink', expected_dir_symlink_item_w_meta))
251 wvpasseq(expected, res)
253 dir_value = tip_tree[b'dir']
254 expected_dir_item = vfs.Item(oid=dir_value.oid,
255 meta=tree_dict(repo, dir_value.oid)[b'.'].meta)
256 expected = ((b'', vfs._root),
257 (b'test', test_revlist_w_meta),
258 (save_time_str, expected_latest_item_w_meta),
259 (b'dir', expected_dir_item))
260 def lresolve(*args, **keys):
261 return resolve(*args, **dict(keys, follow=False))
262 for resname, resolver in (('resolve', resolve),
263 ('resolve nofollow', lresolve)):
264 for path in (b'/test/latest/dir-symlink/',
265 b'/test/latest/dir-symlink/.'):
268 wvpasseq(4, len(res))
269 wvpasseq(expected, res)
272 wvpasseq(4, len(res))
273 wvpasseq(expected, res)
275 def test_local_resolve(tmpdir):
276 prep_and_test_repo(tmpdir,
277 lambda x: LocalRepo(repo_dir=x), _test_resolve)
279 def test_remote_resolve(tmpdir):
280 prep_and_test_repo(tmpdir,
281 lambda x: RemoteRepo(x), _test_resolve)
283 def _test_resolve_loop(repo, tmpdir):
284 data_path = tmpdir + b'/src'
286 symlink(b'loop', data_path + b'/loop')
287 ex((bup_path, b'init'))
288 ex((bup_path, b'index', b'-v', data_path))
290 ex((bup_path, b'save', b'-d', b'%d' % save_utc, b'-tvvn', b'test', b'--strip',
292 save_name = strftime('%Y-%m-%d-%H%M%S', localtime(save_utc)).encode('ascii')
294 wvpasseq('this call should never return',
295 repo.resolve(b'/test/%s/loop' % save_name))
296 except vfs.IOError as res_ex:
297 wvpasseq(ELOOP, res_ex.errno)
298 wvpasseq([b'', b'test', save_name, b'loop'],
299 [name for name, item in res_ex.terminus])
301 def test_local_resolve_loop(tmpdir):
302 prep_and_test_repo(tmpdir,
303 lambda x: LocalRepo(x), _test_resolve_loop)
305 def test_remote_resolve_loop(tmpdir):
306 prep_and_test_repo(tmpdir,
307 lambda x: RemoteRepo(x), _test_resolve_loop)
309 # FIXME: add tests for the want_meta=False cases.