lib/bup/t/tmetadata.py

   1 import grp
   2 import pwd
   3 import stat
   4 import subprocess
   5 import tempfile
   6 from bup import metadata
   7 from bup.helpers import detect_fakeroot
   8 from wvtest import *
   9
  10
  11 @wvtest
  12 def test__normalize_ts():
  13     normalize = metadata._normalize_ts
  14     bns = 10**9
  15     for ts in ((0, 0), (-1, 0), (0, bns - 1), (-1, bns - 1)):
  16         WVPASSEQ(normalize(ts), ts)
  17     WVPASSEQ(normalize((0, -1)), (-1, bns - 1))
  18     WVPASSEQ(normalize((-1, -1)), (-2, bns - 1))
  19     WVPASSEQ(normalize((0, bns)), (1, 0))
  20     WVPASSEQ(normalize((0, bns + 1)), (1, 1))
  21     WVPASSEQ(normalize((0, -bns)), (-1, 0))
  22     WVPASSEQ(normalize((0, -(bns + 1))), (-2, bns - 1))
  23     WVPASSEQ(normalize((0, 3 * bns)), (3, 0))
  24     WVPASSEQ(normalize((0, -3 * bns)), (-3, 0))
  25
  26
  27 @wvtest
  28 def test_clean_up_archive_path():
  29     cleanup = metadata._clean_up_path_for_archive
  30     WVPASSEQ(cleanup('foo'), 'foo')
  31     WVPASSEQ(cleanup('/foo'), 'foo')
  32     WVPASSEQ(cleanup('///foo'), 'foo')
  33     WVPASSEQ(cleanup('/foo/bar'), 'foo/bar')
  34     WVPASSEQ(cleanup('foo/./bar'), 'foo/bar')
  35     WVPASSEQ(cleanup('/foo/./bar'), 'foo/bar')
  36     WVPASSEQ(cleanup('/foo/./bar/././baz'), 'foo/bar/baz')
  37     WVPASSEQ(cleanup('/foo/./bar///././baz'), 'foo/bar/baz')
  38     WVPASSEQ(cleanup('//./foo/./bar///././baz/.///'), 'foo/bar/baz/')
  39     WVPASSEQ(cleanup('./foo/./.bar'), 'foo/.bar')
  40     WVPASSEQ(cleanup('./foo/.'), 'foo')
  41     WVPASSEQ(cleanup('./foo/..'), '.')
  42     WVPASSEQ(cleanup('//./..//.../..//.'), '.')
  43     WVPASSEQ(cleanup('//./..//..././/.'), '...')
  44     WVPASSEQ(cleanup('/////.'), '.')
  45     WVPASSEQ(cleanup('/../'), '.')
  46     WVPASSEQ(cleanup(''), '.')
  47
  48
  49 @wvtest
  50 def test_risky_path():
  51     risky = metadata._risky_path
  52     WVPASS(risky('/foo'))
  53     WVPASS(risky('///foo'))
  54     WVPASS(risky('/../foo'))
  55     WVPASS(risky('../foo'))
  56     WVPASS(risky('foo/..'))
  57     WVPASS(risky('foo/../'))
  58     WVPASS(risky('foo/../bar'))
  59     WVFAIL(risky('foo'))
  60     WVFAIL(risky('foo/'))
  61     WVFAIL(risky('foo///'))
  62     WVFAIL(risky('./foo'))
  63     WVFAIL(risky('foo/.'))
  64     WVFAIL(risky('./foo/.'))
  65     WVFAIL(risky('foo/bar'))
  66     WVFAIL(risky('foo/./bar'))
  67
  68
  69 @wvtest
  70 def test_clean_up_extract_path():
  71     cleanup = metadata._clean_up_extract_path
  72     WVPASSEQ(cleanup('/foo'), 'foo')
  73     WVPASSEQ(cleanup('///foo'), 'foo')
  74     WVFAIL(cleanup('/../foo'))
  75     WVFAIL(cleanup('../foo'))
  76     WVFAIL(cleanup('foo/..'))
  77     WVFAIL(cleanup('foo/../'))
  78     WVFAIL(cleanup('foo/../bar'))
  79     WVPASSEQ(cleanup('foo'), 'foo')
  80     WVPASSEQ(cleanup('foo/'), 'foo/')
  81     WVPASSEQ(cleanup('foo///'), 'foo///')
  82     WVPASSEQ(cleanup('./foo'), './foo')
  83     WVPASSEQ(cleanup('foo/.'), 'foo/.')
  84     WVPASSEQ(cleanup('./foo/.'), './foo/.')
  85     WVPASSEQ(cleanup('foo/bar'), 'foo/bar')
  86     WVPASSEQ(cleanup('foo/./bar'), 'foo/./bar')
  87     WVPASSEQ(cleanup('/'), '.')
  88     WVPASSEQ(cleanup('./'), './')
  89     WVPASSEQ(cleanup('///foo/bar'), 'foo/bar')
  90     WVPASSEQ(cleanup('///foo/bar'), 'foo/bar')
  91
  92
  93 @wvtest
  94 def test_from_path_error():
  95     if os.geteuid == 0 or detect_fakeroot():
  96         return
  97     tmpdir = tempfile.mkdtemp(prefix='bup-tmetadata-')
  98     try:
  99         path = tmpdir + '/foo'
 100         subprocess.call(['mkdir', path])
 101         m = metadata.from_path(path, archive_path=path, save_symlinks=True)
 102         WVPASSEQ(m.path, path)
 103         subprocess.call(['chmod', '000', path])
 104         WVEXCEPT(metadata.MetadataAcquisitionError,
 105                  metadata.from_path,
 106                  path,
 107                  archive_path=path,
 108                  save_symlinks=True)
 109     finally:
 110         subprocess.call(['rm', '-rf', tmpdir])
 111
 112
 113 @wvtest
 114 def test_apply_to_path_error():
 115     if os.geteuid == 0 or detect_fakeroot():
 116         return
 117     tmpdir = tempfile.mkdtemp(prefix='bup-tmetadata-')
 118     try:
 119         path = tmpdir + '/foo'
 120         subprocess.call(['mkdir', path])
 121         m = metadata.from_path(path, archive_path=path, save_symlinks=True)
 122         WVPASSEQ(m.path, path)
 123         subprocess.call(['chmod', '000', tmpdir])
 124         WVEXCEPT(metadata.MetadataApplicationError,
 125                  m.apply_to_path, path)
 126         subprocess.call(['chmod', '700', tmpdir])
 127     finally:
 128         subprocess.call(['rm', '-rf', tmpdir])
 129
 130
 131 @wvtest
 132 def test_restore_restricted_user_group():
 133     if os.geteuid == 0 or detect_fakeroot():
 134         return
 135     tmpdir = tempfile.mkdtemp(prefix='bup-tmetadata-')
 136     try:
 137         path = tmpdir + '/foo'
 138         subprocess.call(['mkdir', path])
 139         m = metadata.from_path(path, archive_path=path, save_symlinks=True)
 140         WVPASSEQ(m.path, path)
 141         WVPASSEQ(m.apply_to_path(path), None)
 142         orig_uid = m.uid
 143         m.uid = 0;
 144         WVEXCEPT(metadata.MetadataApplicationError,
 145                  m.apply_to_path, path, restore_numeric_ids=True)
 146         m.uid = orig_uid
 147         m.gid = 0;
 148         WVEXCEPT(metadata.MetadataApplicationError,
 149                  m.apply_to_path, path, restore_numeric_ids=True)
 150     finally:
 151         subprocess.call(['rm', '-rf', tmpdir])
 152
 153
 154 @wvtest
 155 def test_restore_nonexistent_user_group():
 156     tmpdir = tempfile.mkdtemp(prefix='bup-tmetadata-')
 157     try:
 158         path = tmpdir + '/foo'
 159         subprocess.call(['mkdir', path])
 160         m = metadata.from_path(path, archive_path=path, save_symlinks=True)
 161         WVPASSEQ(m.path, path)
 162         m.owner = max([x.pw_name for x in pwd.getpwall()], key=len) + 'x'
 163         m.group = max([x.gr_name for x in grp.getgrall()], key=len) + 'x'
 164         WVPASSEQ(m.apply_to_path(path, restore_numeric_ids=True), None)
 165         WVPASSEQ(os.stat(path).st_uid, m.uid)
 166         WVPASSEQ(os.stat(path).st_gid, m.gid)
 167         WVPASSEQ(m.apply_to_path(path, restore_numeric_ids=False), None)
 168         WVPASSEQ(os.stat(path).st_uid, os.geteuid())
 169         WVPASSEQ(os.stat(path).st_uid, os.getgid())
 170     finally:
 171         subprocess.call(['rm', '-rf', tmpdir])
 172
 173
 174 @wvtest
 175 def test_restore_over_existing_target():
 176     tmpdir = tempfile.mkdtemp(prefix='bup-tmetadata-')
 177     try:
 178         path = tmpdir + '/foo'
 179         os.mkdir(path)
 180         dir_m = metadata.from_path(path, archive_path=path, save_symlinks=True)
 181         os.rmdir(path)
 182         open(path, 'w').close()
 183         file_m = metadata.from_path(path, archive_path=path, save_symlinks=True)
 184         # Restore dir over file.
 185         WVPASSEQ(dir_m.create_path(path, create_symlinks=True), None)
 186         WVPASS(stat.S_ISDIR(os.stat(path).st_mode))
 187         # Restore dir over dir.
 188         WVPASSEQ(dir_m.create_path(path, create_symlinks=True), None)
 189         WVPASS(stat.S_ISDIR(os.stat(path).st_mode))
 190         # Restore file over dir.
 191         WVPASSEQ(file_m.create_path(path, create_symlinks=True), None)
 192         WVPASS(stat.S_ISREG(os.stat(path).st_mode))
 193         # Restore file over file.
 194         WVPASSEQ(file_m.create_path(path, create_symlinks=True), None)
 195         WVPASS(stat.S_ISREG(os.stat(path).st_mode))
 196         # Restore file over non-empty dir.
 197         os.remove(path)
 198         os.mkdir(path)
 199         open(path + '/bar', 'w').close()
 200         WVEXCEPT(Exception, file_m.create_path, path, create_symlinks=True)
 201         # Restore dir over non-empty dir.
 202         os.remove(path + '/bar')
 203         os.mkdir(path + '/bar')
 204         WVEXCEPT(Exception, dir_m.create_path, path, create_symlinks=True)
 205     finally:
 206         subprocess.call(['rm', '-rf', tmpdir])