test/test_xattr.py

   1 #
   2 #
   3
   4 import sys
   5 import tempfile
   6 import os
   7 import errno
   8 import pytest
   9
  10 import xattr
  11 from xattr import NS_USER, XATTR_CREATE, XATTR_REPLACE
  12
  13 NAMESPACE = os.environ.get("NAMESPACE", NS_USER)
  14
  15 EMPTY_NS = bytes()
  16
  17 TEST_DIR = os.environ.get("TEST_DIR", ".")
  18 TEST_IGNORE_XATTRS = os.environ.get("TEST_IGNORE_XATTRS", "")
  19 if TEST_IGNORE_XATTRS == "":
  20     TEST_IGNORE_XATTRS = []
  21 else:
  22     TEST_IGNORE_XATTRS = TEST_IGNORE_XATTRS.split(",")
  23     # The following has to be a list comprehension, not a generator, to
  24     # avoid weird consequences of lazy evaluation.
  25     TEST_IGNORE_XATTRS.extend([a.encode() for a in TEST_IGNORE_XATTRS])
  26
  27 USER_NN = "test"
  28 USER_ATTR = NAMESPACE.decode() + "." + USER_NN
  29 USER_VAL = "abc"
  30 EMPTY_VAL = ""
  31 LARGE_VAL = "x" * 2048
  32 MANYOPS_COUNT = 16384
  33
  34 USER_NN = USER_NN.encode()
  35 USER_VAL = USER_VAL.encode()
  36 USER_ATTR = USER_ATTR.encode()
  37 EMPTY_VAL = EMPTY_VAL.encode()
  38 LARGE_VAL = LARGE_VAL.encode()
  39
  40 # Helper functions
  41
  42 def ignore_tuples(attrs):
  43     """Remove ignored attributes from the output of xattr.get_all."""
  44     return [attr for attr in attrs
  45             if attr[0] not in TEST_IGNORE_XATTRS]
  46
  47 def ignore(attrs):
  48     """Remove ignored attributes from the output of xattr.list"""
  49     return [attr for attr in attrs
  50             if attr not in TEST_IGNORE_XATTRS]
  51
  52 def lists_equal(attrs, value):
  53     """Helper to check list equivalence, skipping TEST_IGNORE_XATTRS."""
  54     assert ignore(attrs) == value
  55
  56 def tuples_equal(attrs, value):
  57     """Helper to check list equivalence, skipping TEST_IGNORE_XATTRS."""
  58     assert ignore_tuples(attrs) == value
  59
  60 # Fixtures and helpers
  61
  62 @pytest.fixture
  63 def testdir():
  64     """per-test temp dir based in TEST_DIR"""
  65     with tempfile.TemporaryDirectory(dir=TEST_DIR) as dname:
  66         yield dname
  67
  68 def get_file(path):
  69     fh, fname = tempfile.mkstemp(".test", "xattr-", path)
  70     return fh, fname
  71
  72 def get_file_name(path):
  73     fh, fname = get_file(path)
  74     os.close(fh)
  75     return fname
  76
  77 def get_file_fd(path):
  78     return get_file(path)[0]
  79
  80 def get_file_object(path):
  81     fd = get_file(path)[0]
  82     return os.fdopen(fd)
  83
  84 def get_dir(path):
  85     return tempfile.mkdtemp(".test", "xattr-", path)
  86
  87 def get_symlink(path, dangling=True):
  88     """create a symlink"""
  89     fh, fname = get_file(path)
  90     os.close(fh)
  91     if dangling:
  92         os.unlink(fname)
  93     sname = fname + ".symlink"
  94     os.symlink(fname, sname)
  95     return fname, sname
  96
  97 def get_valid_symlink(path):
  98     return get_symlink(path, dangling=False)[1]
  99
 100 def get_dangling_symlink(path):
 101     return get_symlink(path, dangling=True)[1]
 102
 103 def as_bytes(call):
 104     def f(path):
 105         return call(path).encode()
 106     return f
 107
 108 # Note: user attributes are only allowed on files and directories, so
 109 # we have to skip the symlinks here. See xattr(7).
 110 ITEMS_P = [
 111     (get_file_name, False),
 112     (as_bytes(get_file_name), False),
 113     (get_file_fd, False),
 114     (get_file_object, False),
 115     (get_dir, False),
 116     (as_bytes(get_dir), False),
 117     (get_valid_symlink, False),
 118     (as_bytes(get_valid_symlink), False),
 119 ]
 120
 121 ITEMS_D = [
 122     "file name",
 123     "file name (bytes)",
 124     "file FD",
 125     "file object",
 126     "directory",
 127     "directory (bytes)",
 128     "file via symlink",
 129     "file via symlink (bytes)",
 130 ]
 131
 132 ALL_ITEMS_P = ITEMS_P + [
 133     (get_valid_symlink, True),
 134     (as_bytes(get_valid_symlink), True),
 135     (get_dangling_symlink, True),
 136     (as_bytes(get_dangling_symlink), True),
 137 ]
 138
 139 ALL_ITEMS_D = ITEMS_D + [
 140     "valid symlink",
 141     "valid symlink (bytes)",
 142     "dangling symlink",
 143     "dangling symlink (bytes)"
 144 ]
 145
 146 @pytest.fixture(params=ITEMS_P, ids=ITEMS_D)
 147 def subject(testdir, request):
 148     return request.param[0](testdir), request.param[1]
 149
 150 @pytest.fixture(params=ALL_ITEMS_P, ids=ALL_ITEMS_D)
 151 def any_subject(testdir, request):
 152     return request.param[0](testdir), request.param[1]
 153
 154 @pytest.fixture(params=[True, False], ids=["with namespace", "no namespace"])
 155 def use_ns(request):
 156     return request.param
 157
 158 @pytest.fixture(params=[True, False], ids=["dangling", "valid"])
 159 def use_dangling(request):
 160     return request.param
 161
 162 ### Test functions
 163
 164 def test_empty_value(subject):
 165     item, nofollow = subject
 166     xattr.set(item, USER_ATTR, EMPTY_VAL, nofollow=nofollow)
 167     assert xattr.get(item, USER_ATTR, nofollow=nofollow) == EMPTY_VAL
 168
 169 def test_large_value(subject):
 170     item, nofollow = subject
 171     xattr.set(item, USER_ATTR, LARGE_VAL)
 172     assert xattr.get(item, USER_ATTR, nofollow=nofollow) == LARGE_VAL
 173
 174
 175 def test_file_mixed_access_deprecated(testdir):
 176     """test mixed access to file (deprecated functions)"""
 177     fh, fname = get_file(testdir)
 178     with os.fdopen(fh) as fo:
 179         lists_equal(xattr.listxattr(fname), [])
 180         xattr.setxattr(fname, USER_ATTR, USER_VAL)
 181         lists_equal(xattr.listxattr(fh), [USER_ATTR])
 182         assert xattr.getxattr(fo, USER_ATTR) == USER_VAL
 183         tuples_equal(xattr.get_all(fo), [(USER_ATTR, USER_VAL)])
 184         tuples_equal(xattr.get_all(fname),
 185                      [(USER_ATTR, USER_VAL)])
 186
 187 def test_file_mixed_access(testdir):
 188     """test mixed access to file"""
 189     fh, fname = get_file(testdir)
 190     with os.fdopen(fh) as fo:
 191         lists_equal(xattr.list(fname), [])
 192         xattr.set(fname, USER_ATTR, USER_VAL)
 193         lists_equal(xattr.list(fh), [USER_ATTR])
 194         assert xattr.list(fh, namespace=NAMESPACE) == [USER_NN]
 195         assert xattr.get(fo, USER_ATTR) == USER_VAL
 196         assert xattr.get(fo, USER_NN, namespace=NAMESPACE) == USER_VAL
 197         tuples_equal(xattr.get_all(fo),
 198                      [(USER_ATTR, USER_VAL)])
 199         assert xattr.get_all(fo, namespace=NAMESPACE) == \
 200             [(USER_NN, USER_VAL)]
 201         tuples_equal(xattr.get_all(fname), [(USER_ATTR, USER_VAL)])
 202         assert xattr.get_all(fname, namespace=NAMESPACE) == \
 203             [(USER_NN, USER_VAL)]
 204
 205 def test_replace_on_missing(subject, use_ns):
 206     item = subject[0]
 207     lists_equal(xattr.list(item), [])
 208     with pytest.raises(EnvironmentError):
 209         if use_ns:
 210             xattr.set(item, USER_NN, USER_VAL, flags=XATTR_REPLACE,
 211                       namespace=NAMESPACE)
 212         else:
 213             xattr.set(item, USER_ATTR, USER_VAL, flags=XATTR_REPLACE)
 214
 215 def test_create_on_existing(subject, use_ns):
 216     item = subject[0]
 217     lists_equal(xattr.list(item), [])
 218     if use_ns:
 219         xattr.set(item, USER_NN, USER_VAL,
 220                   namespace=NAMESPACE)
 221     else:
 222         xattr.set(item, USER_ATTR, USER_VAL)
 223     with pytest.raises(EnvironmentError):
 224         if use_ns:
 225             xattr.set(item, USER_NN, USER_VAL,
 226                       flags=XATTR_CREATE, namespace=NAMESPACE)
 227         else:
 228             xattr.set(item, USER_ATTR, USER_VAL, flags=XATTR_CREATE)
 229
 230 def test_remove_on_missing(any_subject, use_ns):
 231     item, nofollow = any_subject
 232     lists_equal(xattr.list(item, nofollow=nofollow), [])
 233     with pytest.raises(EnvironmentError):
 234         if use_ns:
 235             xattr.remove(item, USER_NN, namespace=NAMESPACE,
 236                          nofollow=nofollow)
 237         else:
 238             xattr.remove(item, USER_ATTR, nofollow=nofollow)
 239
 240 def test_set_get_remove(subject, use_ns):
 241     item = subject[0]
 242     lists_equal(xattr.list(item), [])
 243     if use_ns:
 244         xattr.set(item, USER_NN, USER_VAL,
 245                   namespace=NAMESPACE)
 246     else:
 247         xattr.set(item, USER_ATTR, USER_VAL)
 248     if use_ns:
 249         assert xattr.list(item, namespace=NAMESPACE) == [USER_NN]
 250     else:
 251         lists_equal(xattr.list(item), [USER_ATTR])
 252         lists_equal(xattr.list(item, namespace=EMPTY_NS),
 253                     [USER_ATTR])
 254     if use_ns:
 255         assert xattr.get(item, USER_NN, namespace=NAMESPACE) == USER_VAL
 256     else:
 257         assert xattr.get(item, USER_ATTR) == USER_VAL
 258     if use_ns:
 259         assert xattr.get_all(item, namespace=NAMESPACE) == \
 260             [(USER_NN, USER_VAL)]
 261     else:
 262         tuples_equal(xattr.get_all(item),
 263                      [(USER_ATTR, USER_VAL)])
 264     if use_ns:
 265         xattr.remove(item, USER_NN, namespace=NAMESPACE)
 266     else:
 267         xattr.remove(item, USER_ATTR)
 268     lists_equal(xattr.list(item), [])
 269     tuples_equal(xattr.get_all(item), [])
 270
 271 def test_replace_on_missing_deprecated(subject):
 272     item = subject[0]
 273     lists_equal(xattr.listxattr(item), [])
 274     with pytest.raises(EnvironmentError):
 275         xattr.setxattr(item, USER_ATTR, USER_VAL, XATTR_REPLACE)
 276
 277 def test_create_on_existing_deprecated(subject):
 278     item = subject[0]
 279     lists_equal(xattr.listxattr(item), [])
 280     xattr.setxattr(item, USER_ATTR, USER_VAL, 0)
 281     with pytest.raises(EnvironmentError):
 282         xattr.setxattr(item, USER_ATTR, USER_VAL, XATTR_CREATE)
 283
 284 def test_remove_on_missing_deprecated(any_subject):
 285     """check deprecated list, set, get operations against an item"""
 286     item, nofollow = any_subject
 287     lists_equal(xattr.listxattr(item, nofollow), [])
 288     with pytest.raises(EnvironmentError):
 289         xattr.removexattr(item, USER_ATTR)
 290
 291 def test_set_get_remove_deprecated(subject):
 292     """check deprecated list, set, get operations against an item"""
 293     item = subject[0]
 294     lists_equal(xattr.listxattr(item), [])
 295     xattr.setxattr(item, USER_ATTR, USER_VAL, 0)
 296     lists_equal(xattr.listxattr(item), [USER_ATTR])
 297     assert xattr.getxattr(item, USER_ATTR) == USER_VAL
 298     tuples_equal(xattr.get_all(item), [(USER_ATTR, USER_VAL)])
 299     xattr.removexattr(item, USER_ATTR)
 300     lists_equal(xattr.listxattr(item), [])
 301     tuples_equal(xattr.get_all(item), [])
 302
 303 def test_many_ops(subject):
 304     """test many ops"""
 305     item = subject[0]
 306     xattr.set(item, USER_ATTR, USER_VAL)
 307     VL = [USER_ATTR]
 308     VN = [USER_NN]
 309     for i in range(MANYOPS_COUNT):
 310         lists_equal(xattr.list(item), VL)
 311         lists_equal(xattr.list(item, namespace=EMPTY_NS), VL)
 312         assert xattr.list(item, namespace=NAMESPACE) == VN
 313     for i in range(MANYOPS_COUNT):
 314         assert xattr.get(item, USER_ATTR) == USER_VAL
 315         assert xattr.get(item, USER_NN, namespace=NAMESPACE) == USER_VAL
 316     for i in range(MANYOPS_COUNT):
 317         tuples_equal(xattr.get_all(item),
 318                      [(USER_ATTR, USER_VAL)])
 319         assert xattr.get_all(item, namespace=NAMESPACE) == \
 320             [(USER_NN, USER_VAL)]
 321
 322 def test_many_ops_deprecated(subject):
 323     """test many ops (deprecated functions)"""
 324     item = subject[0]
 325     xattr.setxattr(item, USER_ATTR, USER_VAL)
 326     VL = [USER_ATTR]
 327     for i in range(MANYOPS_COUNT):
 328         lists_equal(xattr.listxattr(item), VL)
 329     for i in range(MANYOPS_COUNT):
 330         assert xattr.getxattr(item, USER_ATTR) == USER_VAL
 331     for i in range(MANYOPS_COUNT):
 332         tuples_equal(xattr.get_all(item),
 333                      [(USER_ATTR, USER_VAL)])
 334
 335 def test_no_attributes_deprecated(any_subject):
 336     """test no attributes (deprecated functions)"""
 337     item, nofollow = any_subject
 338     lists_equal(xattr.listxattr(item, True), [])
 339     tuples_equal(xattr.get_all(item, True), [])
 340     with pytest.raises(EnvironmentError):
 341         xattr.getxattr(item, USER_ATTR, True)
 342
 343 def test_no_attributes(any_subject):
 344     """test no attributes"""
 345     item, nofollow = any_subject
 346     lists_equal(xattr.list(item, nofollow=nofollow), [])
 347     assert xattr.list(item, nofollow=nofollow,
 348                       namespace=NAMESPACE) == []
 349     tuples_equal(xattr.get_all(item, nofollow=nofollow), [])
 350     assert xattr.get_all(item, nofollow=nofollow,
 351                          namespace=NAMESPACE) == []
 352     with pytest.raises(EnvironmentError):
 353         xattr.get(item, USER_NN, nofollow=nofollow,
 354                   namespace=NAMESPACE)
 355
 356 def test_binary_payload_deprecated(subject):
 357     """test binary values (deprecated functions)"""
 358     item = subject[0]
 359     BINVAL = b"abc\0def"
 360     xattr.setxattr(item, USER_ATTR, BINVAL)
 361     lists_equal(xattr.listxattr(item), [USER_ATTR])
 362     assert xattr.getxattr(item, USER_ATTR) == BINVAL
 363     tuples_equal(xattr.get_all(item), [(USER_ATTR, BINVAL)])
 364     xattr.removexattr(item, USER_ATTR)
 365
 366 def test_binary_payload(subject):
 367     """test binary values"""
 368     item = subject[0]
 369     BINVAL = b"abc\0def"
 370     xattr.set(item, USER_ATTR, BINVAL)
 371     lists_equal(xattr.list(item), [USER_ATTR])
 372     assert xattr.list(item, namespace=NAMESPACE) == [USER_NN]
 373     assert xattr.get(item, USER_ATTR) == BINVAL
 374     assert xattr.get(item, USER_NN, namespace=NAMESPACE) == BINVAL
 375     tuples_equal(xattr.get_all(item), [(USER_ATTR, BINVAL)])
 376     assert xattr.get_all(item, namespace=NAMESPACE) == [(USER_NN, BINVAL)]
 377     xattr.remove(item, USER_ATTR)
 378
 379 def test_symlinks_user_fail(testdir, use_dangling):
 380     _, sname = get_symlink(testdir, dangling=use_dangling)
 381     with pytest.raises(IOError):
 382         xattr.set(sname, USER_ATTR, USER_VAL, nofollow=True)
 383     with pytest.raises(IOError):
 384         xattr.set(sname, USER_NN, USER_VAL, namespace=NAMESPACE,
 385                   nofollow=True)
 386     with pytest.raises(IOError):
 387         xattr.setxattr(sname, USER_ATTR, USER_VAL, XATTR_CREATE, True)
 388
 389 @pytest.mark.parametrize(
 390     "call, args", [(xattr.get, [USER_ATTR]),
 391                    (xattr.list, []),
 392                    (xattr.remove, [USER_ATTR]),
 393                    (xattr.get, [USER_ATTR]),
 394                    (xattr.set, [USER_ATTR, USER_VAL])])
 395 def test_none_namespace(testdir, call, args):
 396     # Don't want to use subject, since that would prevent xfail test
 397     # on path objects (due to hiding the exception here).
 398     f = get_file_name(testdir)
 399     with pytest.raises(TypeError):
 400         call(f, *args, namespace=None)
 401     fd = get_file_fd(testdir)
 402     with pytest.raises(TypeError):
 403         call(fd, *args, namespace=None)
 404
 405 @pytest.mark.parametrize(
 406     "call",
 407     [xattr.get, xattr.list, xattr.listxattr,
 408      xattr.remove, xattr.removexattr,
 409      xattr.set, xattr.setxattr,
 410      xattr.get, xattr.getxattr])
 411 def test_wrong_call(call):
 412     with pytest.raises(TypeError):
 413         call()
 414
 415 @pytest.mark.parametrize(
 416     "call, args", [(xattr.get, [USER_ATTR]),
 417                    (xattr.listxattr, []),
 418                    (xattr.list, []),
 419                    (xattr.remove, [USER_ATTR]),
 420                    (xattr.removexattr, [USER_ATTR]),
 421                    (xattr.get, [USER_ATTR]),
 422                    (xattr.getxattr, [USER_ATTR]),
 423                    (xattr.set, [USER_ATTR, USER_VAL]),
 424                    (xattr.setxattr, [USER_ATTR, USER_VAL])])
 425 def test_wrong_argument_type(call, args):
 426     with pytest.raises(TypeError):
 427         call(object(), *args)