from io import BytesIO
from os import environ, symlink
from random import Random, randint
-from stat import S_IFDIR, S_IFREG, S_ISDIR, S_ISREG
+from stat import S_IFDIR, S_IFLNK, S_IFREG, S_ISDIR, S_ISREG
from sys import stderr
from time import localtime, strftime
print(shstr(cmd), file=stderr)
return exc(cmd, **kwargs)
+@wvtest
+def test_default_modes():
+ wvpasseq(S_IFREG | 0o644, vfs.default_file_mode)
+ wvpasseq(S_IFDIR | 0o755, vfs.default_dir_mode)
+ wvpasseq(S_IFLNK | 0o755, vfs.default_symlink_mode)
+
@wvtest
def test_cache_behavior():
orig_max = vfs._cache_max_items
try:
maybe_meta = lambda : Metadata.read(bupm) if bupm else None
m = maybe_meta()
- if m:
+ if m and m.size is None:
m.size = 0
yield TreeDictValue(name='.', oid=oid, meta=m)
tree_ents = vfs.ordered_tree_entries(tree_data, bupm=True)
file_path, file_size,
link_path, link_target):
_, file_item = vfs.resolve(repo, file_path)[-1]
- _, link_item = vfs.lresolve(repo, link_path)[-1]
+ _, link_item = vfs.resolve(repo, link_path, follow=False)[-1]
wvpass(isinstance(file_item.meta, Metadata))
wvpass(isinstance(link_item.meta, Metadata))
# Note: normally, modifying item.meta values is forbidden
with no_lingering_errors():
with test_tempdir('bup-tvfs-') as tmpdir:
resolve = vfs.resolve
- lresolve = vfs.lresolve
bup_dir = tmpdir + '/bup'
environ['GIT_DIR'] = bup_dir
environ['BUP_DIR'] = bup_dir
expected_latest_item_w_meta = vfs.Commit(meta=tip_tree['.'].meta,
oid=tip_tree_oid,
coid=tip_oid)
+ expected_latest_link = vfs.FakeLink(meta=vfs.default_symlink_mode,
+ target=save_time_str)
expected_test_tag_item = expected_latest_item
wvstart('resolve: /')
# latest has metadata here due to caching
wvpasseq(frozenset([('.', test_revlist_w_meta),
(save_time_str, expected_latest_item_w_meta),
- ('latest', expected_latest_item_w_meta)]),
+ ('latest', expected_latest_link)]),
test_content)
wvstart('resolve: /test/latest')
coid=tip_oid)
expected = (('', vfs._root),
('test', test_revlist_w_meta),
- ('latest', expected_latest_item_w_meta))
+ (save_time_str, expected_latest_item_w_meta))
wvpasseq(expected, res)
ignore, latest_item = res[2]
latest_content = frozenset(vfs.contents(repo, latest_item))
oid=tip_tree['file'].oid)
expected = (('', vfs._root),
('test', test_revlist_w_meta),
- ('latest', expected_latest_item_w_meta),
+ (save_time_str, expected_latest_item_w_meta),
('file', expected_file_item_w_meta))
wvpasseq(expected, res)
wvpasseq(4, len(res))
expected = (('', vfs._root),
('test', test_revlist_w_meta),
- ('latest', expected_latest_item_w_meta),
+ (save_time_str, expected_latest_item_w_meta),
('not-there', None))
wvpasseq(expected, res)
- wvstart('lresolve: /test/latest/bad-symlink')
+ wvstart('resolve nofollow: /test/latest/bad-symlink')
vfs.clear_cache()
- res = lresolve(repo, '/test/latest/bad-symlink')
+ res = resolve(repo, '/test/latest/bad-symlink', follow=False)
wvpasseq(4, len(res))
bad_symlink_value = tip_tree['bad-symlink']
expected_bad_symlink_item_w_meta = vfs.Item(meta=bad_symlink_value.meta,
oid=bad_symlink_value.oid)
expected = (('', vfs._root),
('test', test_revlist_w_meta),
- ('latest', expected_latest_item_w_meta),
+ (save_time_str, expected_latest_item_w_meta),
('bad-symlink', expected_bad_symlink_item_w_meta))
wvpasseq(expected, res)
wvpasseq(4, len(res))
expected = (('', vfs._root),
('test', test_revlist_w_meta),
- ('latest', expected_latest_item_w_meta),
+ (save_time_str, expected_latest_item_w_meta),
('file', expected_file_item_w_meta))
wvpasseq(expected, res)
- wvstart('lresolve: /test/latest/file-symlink')
+ wvstart('resolve nofollow: /test/latest/file-symlink')
vfs.clear_cache()
- res = lresolve(repo, '/test/latest/file-symlink')
+ res = resolve(repo, '/test/latest/file-symlink', follow=False)
wvpasseq(4, len(res))
file_symlink_value = tip_tree['file-symlink']
expected_file_symlink_item_w_meta = vfs.Item(meta=file_symlink_value.meta,
oid=file_symlink_value.oid)
expected = (('', vfs._root),
('test', test_revlist_w_meta),
- ('latest', expected_latest_item_w_meta),
+ (save_time_str, expected_latest_item_w_meta),
('file-symlink', expected_file_symlink_item_w_meta))
wvpasseq(expected, res)
resolve(repo, path)
except vfs.IOError as res_ex:
wvpasseq(ENOTDIR, res_ex.errno)
- wvpasseq(['', 'test', 'latest', 'file'],
+ wvpasseq(['', 'test', save_time_str, 'file'],
[name for name, item in res_ex.terminus])
for path in ('/test/latest/file-symlink/',
'/test/latest/file-symlink/../',
'/test/latest/file-symlink/../.',
'/test/latest/file-symlink/../..'):
- wvstart('lresolve: ' + path)
+ wvstart('resolve nofollow: ' + path)
vfs.clear_cache()
try:
- lresolve(repo, path)
+ resolve(repo, path, follow=False)
except vfs.IOError as res_ex:
wvpasseq(ENOTDIR, res_ex.errno)
- wvpasseq(['', 'test', 'latest', 'file'],
+ wvpasseq(['', 'test', save_time_str, 'file'],
[name for name, item in res_ex.terminus])
wvstart('resolve: non-directory parent')
wvpasseq(ENOTDIR, res_ex.errno)
wvpasseq(None, res_ex.terminus)
- wvstart('lresolve: /test/latest/dir-symlink')
+ wvstart('resolve nofollow: /test/latest/dir-symlink')
vfs.clear_cache()
- res = lresolve(repo, '/test/latest/dir-symlink')
+ res = resolve(repo, '/test/latest/dir-symlink', follow=False)
wvpasseq(4, len(res))
dir_symlink_value = tip_tree['dir-symlink']
expected_dir_symlink_item_w_meta = vfs.Item(meta=dir_symlink_value.meta,
oid=dir_symlink_value.oid)
expected = (('', vfs._root),
('test', test_revlist_w_meta),
- ('latest', expected_latest_item_w_meta),
+ (save_time_str, expected_latest_item_w_meta),
('dir-symlink', expected_dir_symlink_item_w_meta))
wvpasseq(expected, res)
meta=tree_dict(repo, dir_value.oid)['.'].meta)
expected = (('', vfs._root),
('test', test_revlist_w_meta),
- ('latest', expected_latest_item_w_meta),
+ (save_time_str, expected_latest_item_w_meta),
('dir', expected_dir_item))
+ def lresolve(*args, **keys):
+ return resolve(*args, **dict(keys, follow=False))
for resname, resolver in (('resolve', resolve),
- ('lresolve', lresolve)):
+ ('resolve nofollow', lresolve)):
for path in ('/test/latest/dir-symlink/',
'/test/latest/dir-symlink/.'):
wvstart(resname + ': ' + path)
with no_lingering_errors():
with test_tempdir('bup-tvfs-resloop-') as tmpdir:
resolve = vfs.resolve
- lresolve = vfs.lresolve
bup_dir = tmpdir + '/bup'
environ['GIT_DIR'] = bup_dir
environ['BUP_DIR'] = bup_dir
symlink('loop', data_path + '/loop')
ex((bup_path, 'init'))
ex((bup_path, 'index', '-v', data_path))
- ex((bup_path, 'save', '-d', '100000', '-tvvn', 'test', '--strip',
+ save_utc = 100000
+ ex((bup_path, 'save', '-d', str(save_utc), '-tvvn', 'test', '--strip',
data_path))
+ save_name = strftime('%Y-%m-%d-%H%M%S', localtime(save_utc))
try:
- resolve(repo, '/test/latest/loop')
+ wvpasseq('this call should never return',
+ resolve(repo, '/test/%s/loop' % save_name))
except vfs.IOError as res_ex:
wvpasseq(ELOOP, res_ex.errno)
- wvpasseq(['', 'test', 'latest', 'loop'],
+ wvpasseq(['', 'test', save_name, 'loop'],
[name for name, item in res_ex.terminus])
@wvtest
tmpfile.write(b'canary\n')
ex((bup_path, 'init'))
ex((bup_path, 'index', '-v', data_path))
- ex((bup_path, 'save', '-tvvn', 'test', '--strip',
- data_path))
+ save_utc = 100000
+ save_name = strftime('%Y-%m-%d-%H%M%S', localtime(save_utc))
+ ex((bup_path, 'save', '-tvvn', 'test', '-d', str(save_utc),
+ '--strip', data_path))
repo = LocalRepo()
tip_sref = exo(('git', 'show-ref', 'refs/heads/test'))[0]
tip_oidx = tip_sref.strip().split()[0]
tip_tree = tree_dict(repo, tip_tree_oid)
name, item = vfs.resolve(repo, '/test/latest')[2]
- wvpasseq('latest', name)
+ wvpasseq(save_name, name)
expected = frozenset((x.name, vfs.Item(oid=x.oid, meta=x.meta))
for x in (tip_tree[name]
for name in ('.', 'foo', 'foo.')))