test/int/test_resolve.py

   1
   2 from __future__ import absolute_import, print_function
   3 from binascii import unhexlify
   4 from errno import ELOOP, ENOTDIR
   5 from os import symlink
   6 from stat import S_IFDIR
   7 from sys import stderr
   8 import os
   9 from time import localtime, strftime
  10
  11 from wvpytest import *
  12
  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
  20
  21 bup_path = path.exe()
  22
  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.
  28
  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'))
  34     git.repodir = bup_dir
  35     with create_repo(bup_dir) as repo:
  36         test_repo(repo, tmpdir)
  37
  38 # Currently, we just test through the repos since LocalRepo resolve is
  39 # just a straight redirection to vfs.resolve.
  40
  41 def _test_resolve(repo, tmpdir):
  42     data_path = tmpdir + b'/src'
  43     resolve = repo.resolve
  44     save_time = 100000
  45     save_time_str = strftime('%Y-%m-%d-%H%M%S', localtime(save_time)).encode('ascii')
  46     os.mkdir(data_path)
  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'))
  57
  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',
  62                          tip_oidx))[0].strip()
  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,
  66                                       oid=tip_oid)
  67     expected_latest_item = vfs.Commit(meta=S_IFDIR | 0o755,
  68                                       oid=tip_tree_oid,
  69                                       coid=tip_oid)
  70     expected_latest_item_w_meta = vfs.Commit(meta=tip_tree[b'.'].meta,
  71                                              oid=tip_tree_oid,
  72                                              coid=tip_oid)
  73     expected_latest_link = vfs.FakeLink(meta=vfs.default_symlink_mode,
  74                                         target=save_time_str)
  75     expected_test_tag_item = expected_latest_item
  76
  77     vfs.clear_cache()
  78     res = resolve(b'/')
  79     wvpasseq(1, len(res))
  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),
  84                         (b'.tag', vfs._tags),
  85                         (b'test', test_revlist_w_meta)]),
  86              root_content)
  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/../../..'):
  98         vfs.clear_cache()
  99         res = resolve(path)
 100         wvpasseq(((b'', vfs._root),), res)
 101
 102     vfs.clear_cache()
 103     res = resolve(b'/.tag')
 104     wvpasseq(2, len(res))
 105     wvpasseq(((b'', vfs._root), (b'.tag', vfs._tags)),
 106              res)
 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)]),
 111              tag_content)
 112
 113     vfs.clear_cache()
 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)]),
 123              test_content)
 124
 125     vfs.clear_cache()
 126     res = resolve(b'/test/latest')
 127     wvpasseq(3, len(res))
 128     expected_latest_item_w_meta = vfs.Commit(meta=tip_tree[b'.'].meta,
 129                                              oid=tip_tree_oid,
 130                                              coid=tip_oid)
 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]
 139                                    for name in (b'.',
 140                                                 b'bad-symlink',
 141                                                 b'dir',
 142                                                 b'dir-symlink',
 143                                                 b'file',
 144                                                 b'file-symlink')))
 145     wvpasseq(expected, latest_content)
 146
 147     vfs.clear_cache()
 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)
 157
 158     vfs.clear_cache()
 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)
 166
 167     vfs.clear_cache()
 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)
 178
 179     vfs.clear_cache()
 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)
 187
 188     vfs.clear_cache()
 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)
 199
 200     vfs.clear_cache()
 201     res = resolve(b'/test/latest/missing')
 202     wvpasseq(4, len(res))
 203     name, item = res[-1]
 204     wvpasseq(b'missing', name)
 205     wvpass(item is None)
 206
 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'):
 214         vfs.clear_cache()
 215         try:
 216             resolve(path)
 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])
 221
 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/../..'):
 228         vfs.clear_cache()
 229         try:
 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])
 235
 236     vfs.clear_cache()
 237     file_res = resolve(b'/test/latest/file')
 238     try:
 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)
 243
 244     vfs.clear_cache()
 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)
 255
 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/.'):
 269             vfs.clear_cache()
 270             res = resolver(path)
 271             wvpasseq(4, len(res))
 272             wvpasseq(expected, res)
 273     vfs.clear_cache()
 274     res = resolve(path)
 275     wvpasseq(4, len(res))
 276     wvpasseq(expected, res)
 277
 278 def test_local_resolve(tmpdir):
 279     prep_and_test_repo(tmpdir,
 280                        lambda x: LocalRepo(repo_dir=x), _test_resolve)
 281
 282 def test_remote_resolve(tmpdir):
 283     prep_and_test_repo(tmpdir,
 284                        lambda x: RemoteRepo(x), _test_resolve)
 285
 286 def _test_resolve_loop(repo, tmpdir):
 287     data_path = tmpdir + b'/src'
 288     os.mkdir(data_path)
 289     symlink(b'loop', data_path + b'/loop')
 290     ex((bup_path, b'init'))
 291     ex((bup_path, b'index', b'-v', data_path))
 292     save_utc = 100000
 293     ex((bup_path, b'save', b'-d', b'%d' % save_utc, b'-tvvn', b'test', b'--strip',
 294         data_path))
 295     save_name = strftime('%Y-%m-%d-%H%M%S', localtime(save_utc)).encode('ascii')
 296     try:
 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])
 303
 304 def test_local_resolve_loop(tmpdir):
 305     prep_and_test_repo(tmpdir,
 306                        lambda x: LocalRepo(x), _test_resolve_loop)
 307
 308 def test_remote_resolve_loop(tmpdir):
 309     prep_and_test_repo(tmpdir,
 310                        lambda x: RemoteRepo(x), _test_resolve_loop)
 311
 312 # FIXME: add tests for the want_meta=False cases.