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 import os
   8 from time import localtime, strftime
   9
  10 from wvpytest import *
  11
  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
  17
  18 bup_path = path.exe()
  19
  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.
  25
  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'))
  31     git.repodir = bup_dir
  32     with create_repo(bup_dir) as repo:
  33         test_repo(repo, tmpdir)
  34
  35 # Currently, we just test through the repos since LocalRepo resolve is
  36 # just a straight redirection to vfs.resolve.
  37
  38 def _test_resolve(repo, tmpdir):
  39     data_path = tmpdir + b'/src'
  40     resolve = repo.resolve
  41     save_time = 100000
  42     save_time_str = strftime('%Y-%m-%d-%H%M%S', localtime(save_time)).encode('ascii')
  43     os.mkdir(data_path)
  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'))
  54
  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',
  59                          tip_oidx))[0].strip()
  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,
  63                                       oid=tip_oid)
  64     expected_latest_item = vfs.Commit(meta=S_IFDIR | 0o755,
  65                                       oid=tip_tree_oid,
  66                                       coid=tip_oid)
  67     expected_latest_item_w_meta = vfs.Commit(meta=tip_tree[b'.'].meta,
  68                                              oid=tip_tree_oid,
  69                                              coid=tip_oid)
  70     expected_latest_link = vfs.FakeLink(meta=vfs.default_symlink_mode,
  71                                         target=save_time_str)
  72     expected_test_tag_item = expected_latest_item
  73
  74     vfs.clear_cache()
  75     res = resolve(b'/')
  76     wvpasseq(1, len(res))
  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),
  81                         (b'.tag', vfs._tags),
  82                         (b'test', test_revlist_w_meta)]),
  83              root_content)
  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/../../..'):
  95         vfs.clear_cache()
  96         res = resolve(path)
  97         wvpasseq(((b'', vfs._root),), res)
  98
  99     vfs.clear_cache()
 100     res = resolve(b'/.tag')
 101     wvpasseq(2, len(res))
 102     wvpasseq(((b'', vfs._root), (b'.tag', vfs._tags)),
 103              res)
 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)]),
 108              tag_content)
 109
 110     vfs.clear_cache()
 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)]),
 120              test_content)
 121
 122     vfs.clear_cache()
 123     res = resolve(b'/test/latest')
 124     wvpasseq(3, len(res))
 125     expected_latest_item_w_meta = vfs.Commit(meta=tip_tree[b'.'].meta,
 126                                              oid=tip_tree_oid,
 127                                              coid=tip_oid)
 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]
 136                                    for name in (b'.',
 137                                                 b'bad-symlink',
 138                                                 b'dir',
 139                                                 b'dir-symlink',
 140                                                 b'file',
 141                                                 b'file-symlink')))
 142     wvpasseq(expected, latest_content)
 143
 144     vfs.clear_cache()
 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)
 154
 155     vfs.clear_cache()
 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)
 163
 164     vfs.clear_cache()
 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)
 175
 176     vfs.clear_cache()
 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)
 184
 185     vfs.clear_cache()
 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)
 196
 197     vfs.clear_cache()
 198     res = resolve(b'/test/latest/missing')
 199     wvpasseq(4, len(res))
 200     name, item = res[-1]
 201     wvpasseq(b'missing', name)
 202     wvpass(item is None)
 203
 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'):
 211         vfs.clear_cache()
 212         try:
 213             resolve(path)
 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])
 218
 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/../..'):
 225         vfs.clear_cache()
 226         try:
 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])
 232
 233     vfs.clear_cache()
 234     file_res = resolve(b'/test/latest/file')
 235     try:
 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)
 240
 241     vfs.clear_cache()
 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)
 252
 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/.'):
 266             vfs.clear_cache()
 267             res = resolver(path)
 268             wvpasseq(4, len(res))
 269             wvpasseq(expected, res)
 270     vfs.clear_cache()
 271     res = resolve(path)
 272     wvpasseq(4, len(res))
 273     wvpasseq(expected, res)
 274
 275 def test_local_resolve(tmpdir):
 276     prep_and_test_repo(tmpdir,
 277                        lambda x: LocalRepo(repo_dir=x), _test_resolve)
 278
 279 def test_remote_resolve(tmpdir):
 280     prep_and_test_repo(tmpdir,
 281                        lambda x: RemoteRepo(x), _test_resolve)
 282
 283 def _test_resolve_loop(repo, tmpdir):
 284     data_path = tmpdir + b'/src'
 285     os.mkdir(data_path)
 286     symlink(b'loop', data_path + b'/loop')
 287     ex((bup_path, b'init'))
 288     ex((bup_path, b'index', b'-v', data_path))
 289     save_utc = 100000
 290     ex((bup_path, b'save', b'-d', b'%d' % save_utc, b'-tvvn', b'test', b'--strip',
 291         data_path))
 292     save_name = strftime('%Y-%m-%d-%H%M%S', localtime(save_utc)).encode('ascii')
 293     try:
 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])
 300
 301 def test_local_resolve_loop(tmpdir):
 302     prep_and_test_repo(tmpdir,
 303                        lambda x: LocalRepo(x), _test_resolve_loop)
 304
 305 def test_remote_resolve_loop(tmpdir):
 306     prep_and_test_repo(tmpdir,
 307                        lambda x: RemoteRepo(x), _test_resolve_loop)
 308
 309 # FIXME: add tests for the want_meta=False cases.