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