test/test_xattr.py

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