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
9 from time import localtime, strftime
11 from wvpytest import *
13 from bup import git, path, vfs
14 from bup.compat import environ
15 from bup.io import path_msg
16 from bup.metadata import Metadata
17 from bup.repo import LocalRepo, RemoteRepo
18 from buptest import ex, exo
19 from buptest.vfs import tree_dict
23 ## The clear_cache() calls below are to make sure that the test starts
24 ## from a known state since at the moment the cache entry for a given
25 ## item (like a commit) can change. For example, its meta value might
26 ## be promoted from a mode to a Metadata instance once the tree it
27 ## refers to is traversed.
29 def prep_and_test_repo(tmpdir, create_repo, test_repo):
30 bup_dir = tmpdir + b'/bup'
31 environ[b'GIT_DIR'] = bup_dir
32 environ[b'BUP_DIR'] = bup_dir
33 ex((bup_path, b'init'))
35 with create_repo(bup_dir) as repo:
36 test_repo(repo, tmpdir)
38 # Currently, we just test through the repos since LocalRepo resolve is
39 # just a straight redirection to vfs.resolve.
41 def _test_resolve(repo, tmpdir):
42 data_path = tmpdir + b'/src'
43 resolve = repo.resolve
45 save_time_str = strftime('%Y-%m-%d-%H%M%S', localtime(save_time)).encode('ascii')
47 os.mkdir(data_path + b'/dir')
48 with open(data_path + b'/file', 'wb+') as tmpfile:
49 tmpfile.write(b'canary\n')
50 symlink(b'file', data_path + b'/file-symlink')
51 symlink(b'dir', data_path + b'/dir-symlink')
52 symlink(b'not-there', data_path + b'/bad-symlink')
53 ex((bup_path, b'index', b'-v', data_path))
54 ex((bup_path, b'save', b'-d', b'%d' % save_time, b'-tvvn', b'test',
55 b'--strip', data_path))
56 ex((bup_path, b'tag', b'test-tag', b'test'))
58 tip_hash = exo((b'git', b'show-ref', b'refs/heads/test'))[0]
59 tip_oidx = tip_hash.strip().split()[0]
60 tip_oid = unhexlify(tip_oidx)
61 tip_tree_oidx = exo((b'git', b'log', b'--pretty=%T', b'-n1',
63 tip_tree_oid = unhexlify(tip_tree_oidx)
64 tip_tree = tree_dict(repo, tip_tree_oid)
65 test_revlist_w_meta = vfs.RevList(meta=tip_tree[b'.'].meta,
67 expected_latest_item = vfs.Commit(meta=S_IFDIR | 0o755,
70 expected_latest_item_w_meta = vfs.Commit(meta=tip_tree[b'.'].meta,
73 expected_latest_link = vfs.FakeLink(meta=vfs.default_symlink_mode,
75 expected_test_tag_item = expected_latest_item
80 wvpasseq(((b'', vfs._root),), res)
81 ignore, root_item = res[0]
82 root_content = frozenset(vfs.contents(repo, root_item))
83 wvpasseq(frozenset([(b'.', root_item),
85 (b'test', test_revlist_w_meta)]),
87 for path in (b'//', b'/.', b'/./', b'/..', b'/../',
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/../../..',
95 b'/test/.//./latest/dir/../../..',
96 b'/test//.//.//latest/dir/../../..'
97 b'/test//./latest/dir/../../..'):
100 wvpasseq(((b'', vfs._root),), res)
103 res = resolve(b'/.tag')
104 wvpasseq(2, len(res))
105 wvpasseq(((b'', vfs._root), (b'.tag', vfs._tags)),
107 ignore, tag_item = res[1]
108 tag_content = frozenset(vfs.contents(repo, tag_item))
109 wvpasseq(frozenset([(b'.', tag_item),
110 (b'test-tag', expected_test_tag_item)]),
114 res = resolve(b'/test')
115 wvpasseq(2, len(res))
116 wvpasseq(((b'', vfs._root), (b'test', test_revlist_w_meta)), res)
117 ignore, test_item = res[1]
118 test_content = frozenset(vfs.contents(repo, test_item))
119 # latest has metadata here due to caching
120 wvpasseq(frozenset([(b'.', test_revlist_w_meta),
121 (save_time_str, expected_latest_item_w_meta),
122 (b'latest', expected_latest_link)]),
126 res = resolve(b'/test/latest')
127 wvpasseq(3, len(res))
128 expected_latest_item_w_meta = vfs.Commit(meta=tip_tree[b'.'].meta,
131 expected = ((b'', vfs._root),
132 (b'test', test_revlist_w_meta),
133 (save_time_str, expected_latest_item_w_meta))
134 wvpasseq(expected, res)
135 ignore, latest_item = res[2]
136 latest_content = frozenset(vfs.contents(repo, latest_item))
137 expected = frozenset((x.name, vfs.Item(oid=x.oid, meta=x.meta))
138 for x in (tip_tree[name]
145 wvpasseq(expected, latest_content)
148 res = resolve(b'/test/latest/file')
149 wvpasseq(4, len(res))
150 expected_file_item_w_meta = vfs.Item(meta=tip_tree[b'file'].meta,
151 oid=tip_tree[b'file'].oid)
152 expected = ((b'', vfs._root),
153 (b'test', test_revlist_w_meta),
154 (save_time_str, expected_latest_item_w_meta),
155 (b'file', expected_file_item_w_meta))
156 wvpasseq(expected, res)
159 res = resolve(b'/test/latest/bad-symlink')
160 wvpasseq(4, len(res))
161 expected = ((b'', vfs._root),
162 (b'test', test_revlist_w_meta),
163 (save_time_str, expected_latest_item_w_meta),
164 (b'not-there', None))
165 wvpasseq(expected, res)
168 res = resolve(b'/test/latest/bad-symlink', follow=False)
169 wvpasseq(4, len(res))
170 bad_symlink_value = tip_tree[b'bad-symlink']
171 expected_bad_symlink_item_w_meta = vfs.Item(meta=bad_symlink_value.meta,
172 oid=bad_symlink_value.oid)
173 expected = ((b'', vfs._root),
174 (b'test', test_revlist_w_meta),
175 (save_time_str, expected_latest_item_w_meta),
176 (b'bad-symlink', expected_bad_symlink_item_w_meta))
177 wvpasseq(expected, res)
180 res = resolve(b'/test/latest/file-symlink')
181 wvpasseq(4, len(res))
182 expected = ((b'', vfs._root),
183 (b'test', test_revlist_w_meta),
184 (save_time_str, expected_latest_item_w_meta),
185 (b'file', expected_file_item_w_meta))
186 wvpasseq(expected, res)
189 res = resolve(b'/test/latest/file-symlink', follow=False)
190 wvpasseq(4, len(res))
191 file_symlink_value = tip_tree[b'file-symlink']
192 expected_file_symlink_item_w_meta = vfs.Item(meta=file_symlink_value.meta,
193 oid=file_symlink_value.oid)
194 expected = ((b'', vfs._root),
195 (b'test', test_revlist_w_meta),
196 (save_time_str, expected_latest_item_w_meta),
197 (b'file-symlink', expected_file_symlink_item_w_meta))
198 wvpasseq(expected, res)
201 res = resolve(b'/test/latest/missing')
202 wvpasseq(4, len(res))
204 wvpasseq(b'missing', name)
207 for path in (b'/test/latest/file/',
208 b'/test/latest/file/.',
209 b'/test/latest/file/..',
210 b'/test/latest/file/../',
211 b'/test/latest/file/../.',
212 b'/test/latest/file/../..',
213 b'/test/latest/file/foo'):
217 except vfs.IOError as res_ex:
218 wvpasseq(ENOTDIR, res_ex.errno)
219 wvpasseq([b'', b'test', save_time_str, b'file'],
220 [name for name, item in res_ex.terminus])
222 for path in (b'/test/latest/file-symlink/',
223 b'/test/latest/file-symlink/.',
224 b'/test/latest/file-symlink/..',
225 b'/test/latest/file-symlink/../',
226 b'/test/latest/file-symlink/../.',
227 b'/test/latest/file-symlink/../..'):
230 resolve(path, follow=False)
231 except vfs.IOError as res_ex:
232 wvpasseq(ENOTDIR, res_ex.errno)
233 wvpasseq([b'', b'test', save_time_str, b'file'],
234 [name for name, item in res_ex.terminus])
237 file_res = resolve(b'/test/latest/file')
239 resolve(b'foo', parent=file_res)
240 except vfs.IOError as res_ex:
241 wvpasseq(ENOTDIR, res_ex.errno)
242 wvpasseq(None, res_ex.terminus)
245 res = resolve(b'/test/latest/dir-symlink', follow=False)
246 wvpasseq(4, len(res))
247 dir_symlink_value = tip_tree[b'dir-symlink']
248 expected_dir_symlink_item_w_meta = vfs.Item(meta=dir_symlink_value.meta,
249 oid=dir_symlink_value.oid)
250 expected = ((b'', vfs._root),
251 (b'test', test_revlist_w_meta),
252 (save_time_str, expected_latest_item_w_meta),
253 (b'dir-symlink', expected_dir_symlink_item_w_meta))
254 wvpasseq(expected, res)
256 dir_value = tip_tree[b'dir']
257 expected_dir_item = vfs.Item(oid=dir_value.oid,
258 meta=tree_dict(repo, dir_value.oid)[b'.'].meta)
259 expected = ((b'', vfs._root),
260 (b'test', test_revlist_w_meta),
261 (save_time_str, expected_latest_item_w_meta),
262 (b'dir', expected_dir_item))
263 def lresolve(*args, **keys):
264 return resolve(*args, **dict(keys, follow=False))
265 for resname, resolver in (('resolve', resolve),
266 ('resolve nofollow', lresolve)):
267 for path in (b'/test/latest/dir-symlink/',
268 b'/test/latest/dir-symlink/.'):
271 wvpasseq(4, len(res))
272 wvpasseq(expected, res)
275 wvpasseq(4, len(res))
276 wvpasseq(expected, res)
278 def test_local_resolve(tmpdir):
279 prep_and_test_repo(tmpdir,
280 lambda x: LocalRepo(repo_dir=x), _test_resolve)
282 def test_remote_resolve(tmpdir):
283 prep_and_test_repo(tmpdir,
284 lambda x: RemoteRepo(x), _test_resolve)
286 def _test_resolve_loop(repo, tmpdir):
287 data_path = tmpdir + b'/src'
289 symlink(b'loop', data_path + b'/loop')
290 ex((bup_path, b'init'))
291 ex((bup_path, b'index', b'-v', data_path))
293 ex((bup_path, b'save', b'-d', b'%d' % save_utc, b'-tvvn', b'test', b'--strip',
295 save_name = strftime('%Y-%m-%d-%H%M%S', localtime(save_utc)).encode('ascii')
297 wvpasseq('this call should never return',
298 repo.resolve(b'/test/%s/loop' % save_name))
299 except vfs.IOError as res_ex:
300 wvpasseq(ELOOP, res_ex.errno)
301 wvpasseq([b'', b'test', save_name, b'loop'],
302 [name for name, item in res_ex.terminus])
304 def test_local_resolve_loop(tmpdir):
305 prep_and_test_repo(tmpdir,
306 lambda x: LocalRepo(x), _test_resolve_loop)
308 def test_remote_resolve_loop(tmpdir):
309 prep_and_test_repo(tmpdir,
310 lambda x: RemoteRepo(x), _test_resolve_loop)
312 # FIXME: add tests for the want_meta=False cases.