test/test_xattr.py

   1 #
   2 #
   3
   4 import sys
   5 import unittest
   6 import tempfile
   7 import os
   8 import errno
   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 if sys.hexversion >= 0x03000000:
  16     PY3K = True
  17     EMPTY_NS = bytes()
  18 else:
  19     PY3K = False
  20     EMPTY_NS = ''
  21
  22 TEST_DIR = os.environ.get("TEST_DIR", ".")
  23 TEST_IGNORE_XATTRS = os.environ.get("TEST_IGNORE_XATTRS", "")
  24 if TEST_IGNORE_XATTRS == "":
  25     TEST_IGNORE_XATTRS = []
  26 else:
  27     TEST_IGNORE_XATTRS = TEST_IGNORE_XATTRS.split(",")
  28     # The following has to be a list comprehension, not a generator, to
  29     # avoid weird consequences of lazy evaluation.
  30     TEST_IGNORE_XATTRS.extend([a.encode() for a in TEST_IGNORE_XATTRS])
  31
  32 USER_NN = "test"
  33 USER_ATTR = NAMESPACE.decode() + "." + USER_NN
  34 USER_VAL = "abc"
  35 EMPTY_VAL = ""
  36 LARGE_VAL = "x" * 2048
  37 MANYOPS_COUNT = 131072
  38
  39 if PY3K:
  40     USER_NN = USER_NN.encode()
  41     USER_VAL = USER_VAL.encode()
  42     USER_ATTR = USER_ATTR.encode()
  43     EMPTY_VAL = EMPTY_VAL.encode()
  44     LARGE_VAL = LARGE_VAL.encode()
  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
  65 class xattrTest(unittest.TestCase):
  66     def setUp(self):
  67         """set up function"""
  68         self.rmfiles = []
  69         self.rmdirs = []
  70
  71     def tearDown(self):
  72         """tear down function"""
  73         for fname in self.rmfiles:
  74             try:
  75                 os.unlink(fname)
  76             except EnvironmentError:
  77                 continue
  78         for dname in self.rmdirs:
  79             try:
  80                 os.rmdir(dname)
  81             except EnvironmentError:
  82                 continue
  83
  84     def _getfile(self):
  85         """create a temp file"""
  86         fh, fname = tempfile.mkstemp(".test", "xattr-", TEST_DIR)
  87         self.rmfiles.append(fname)
  88         return fh, fname
  89
  90     def _getdir(self):
  91         """create a temp dir"""
  92         dname = tempfile.mkdtemp(".test", "xattr-", TEST_DIR)
  93         self.rmdirs.append(dname)
  94         return dname
  95
  96     def _getsymlink(self, dangling=True):
  97         """create a symlink"""
  98         fh, fname = self._getfile()
  99         os.close(fh)
 100         if dangling:
 101             os.unlink(fname)
 102         sname = fname + ".symlink"
 103         os.symlink(fname, sname)
 104         self.rmfiles.append(sname)
 105         return fname, sname
 106
 107     def _checkDeprecated(self, item, symlink=False):
 108         """check deprecated list, set, get operations against an item"""
 109         lists_equal(xattr.listxattr(item, symlink), [])
 110         self.assertRaises(EnvironmentError, xattr.setxattr, item,
 111                           USER_ATTR, USER_VAL,
 112                           XATTR_REPLACE, symlink)
 113         try:
 114             xattr.setxattr(item, USER_ATTR, USER_VAL, 0, symlink)
 115         except IOError:
 116             err = sys.exc_info()[1]
 117             if symlink and (err.errno == errno.EPERM or
 118                             err.errno == errno.ENOENT):
 119                 # symlinks may fail, in which case we abort the rest
 120                 # of the test for this case (Linux returns EPERM; OS X
 121                 # returns ENOENT)
 122                 return
 123             raise
 124         self.assertRaises(EnvironmentError, xattr.setxattr, item,
 125                           USER_ATTR, USER_VAL, XATTR_CREATE, symlink)
 126         lists_equal(xattr.listxattr(item, symlink), [USER_ATTR])
 127         self.assertEqual(xattr.getxattr(item, USER_ATTR, symlink),
 128                          USER_VAL)
 129         tuples_equal(xattr.get_all(item, nofollow=symlink),
 130                           [(USER_ATTR, USER_VAL)])
 131         xattr.removexattr(item, USER_ATTR, symlink)
 132         lists_equal(xattr.listxattr(item, symlink), [])
 133         tuples_equal(xattr.get_all(item, nofollow=symlink),
 134                          [])
 135         self.assertRaises(EnvironmentError, xattr.removexattr,
 136                           item, USER_ATTR, symlink)
 137
 138     def _checkListSetGet(self, item, symlink=False, use_ns=False):
 139         """check list, set, get operations against an item"""
 140         lists_equal(xattr.list(item, symlink), [])
 141         self.assertRaises(EnvironmentError, xattr.set, item,
 142                           USER_ATTR, USER_VAL,
 143                           flags=XATTR_REPLACE,
 144                           nofollow=symlink)
 145         self.assertRaises(EnvironmentError, xattr.set, item,
 146                           USER_NN, USER_VAL,
 147                           flags=XATTR_REPLACE,
 148                           namespace=NAMESPACE,
 149                           nofollow=symlink)
 150         try:
 151             if use_ns:
 152                 xattr.set(item, USER_NN, USER_VAL,
 153                           namespace=NAMESPACE,
 154                           nofollow=symlink)
 155             else:
 156                 xattr.set(item, USER_ATTR, USER_VAL,
 157                           nofollow=symlink)
 158         except IOError:
 159             err = sys.exc_info()[1]
 160             if symlink and (err.errno == errno.EPERM or
 161                             err.errno == errno.ENOENT):
 162                 # symlinks may fail, in which case we abort the rest
 163                 # of the test for this case (Linux returns EPERM; OS X
 164                 # returns ENOENT)
 165                 return
 166             raise
 167         self.assertRaises(EnvironmentError, xattr.set, item,
 168                           USER_ATTR, USER_VAL,
 169                           flags=XATTR_CREATE,
 170                           nofollow=symlink)
 171         self.assertRaises(EnvironmentError, xattr.set, item,
 172                           USER_NN, USER_VAL,
 173                           flags=XATTR_CREATE,
 174                           namespace=NAMESPACE,
 175                           nofollow=symlink)
 176         lists_equal(xattr.list(item, nofollow=symlink), [USER_ATTR])
 177         lists_equal(xattr.list(item, nofollow=symlink,
 178                                   namespace=EMPTY_NS),
 179                        [USER_ATTR])
 180         self.assertEqual(xattr.list(item, namespace=NAMESPACE, nofollow=symlink),
 181                          [USER_NN])
 182         self.assertEqual(xattr.get(item, USER_ATTR, nofollow=symlink),
 183                          USER_VAL)
 184         self.assertEqual(xattr.get(item, USER_NN, nofollow=symlink,
 185                                    namespace=NAMESPACE), USER_VAL)
 186         tuples_equal(xattr.get_all(item, nofollow=symlink),
 187                          [(USER_ATTR, USER_VAL)])
 188         self.assertEqual(xattr.get_all(item, nofollow=symlink,
 189                                        namespace=NAMESPACE),
 190                          [(USER_NN, USER_VAL)])
 191         if use_ns:
 192             xattr.remove(item, USER_NN, namespace=NAMESPACE, nofollow=symlink)
 193         else:
 194             xattr.remove(item, USER_ATTR, nofollow=symlink)
 195         lists_equal(xattr.list(item, nofollow=symlink), [])
 196         tuples_equal(xattr.get_all(item, nofollow=symlink),
 197                          [])
 198         self.assertRaises(EnvironmentError, xattr.remove,
 199                           item, USER_ATTR, nofollow=symlink)
 200         self.assertRaises(EnvironmentError, xattr.remove, item,
 201                           USER_NN, namespace=NAMESPACE, nofollow=symlink)
 202
 203     def testNoXattrDeprecated(self):
 204         """test no attributes (deprecated functions)"""
 205         fh, fname = self._getfile()
 206         lists_equal(xattr.listxattr(fname), [])
 207         tuples_equal(xattr.get_all(fname), [])
 208         self.assertRaises(EnvironmentError, xattr.getxattr, fname,
 209                               USER_ATTR)
 210         dname = self._getdir()
 211         lists_equal(xattr.listxattr(dname), [])
 212         tuples_equal(xattr.get_all(dname), [])
 213         self.assertRaises(EnvironmentError, xattr.getxattr, dname,
 214                               USER_ATTR)
 215         _, sname = self._getsymlink()
 216         lists_equal(xattr.listxattr(sname, True), [])
 217         tuples_equal(xattr.get_all(sname, nofollow=True), [])
 218         self.assertRaises(EnvironmentError, xattr.getxattr, fname,
 219                               USER_ATTR, True)
 220
 221
 222     def testNoXattr(self):
 223         """test no attributes"""
 224         fh, fname = self._getfile()
 225         lists_equal(xattr.list(fname), [])
 226         self.assertEqual(xattr.list(fname, namespace=NAMESPACE), [])
 227         tuples_equal(xattr.get_all(fname), [])
 228         self.assertEqual(xattr.get_all(fname, namespace=NAMESPACE), [])
 229         self.assertRaises(EnvironmentError, xattr.get, fname,
 230                               USER_NN, namespace=NAMESPACE)
 231         dname = self._getdir()
 232         lists_equal(xattr.list(dname), [])
 233         self.assertEqual(xattr.list(dname, namespace=NAMESPACE), [])
 234         tuples_equal(xattr.get_all(dname), [])
 235         self.assertEqual(xattr.get_all(dname, namespace=NAMESPACE), [])
 236         self.assertRaises(EnvironmentError, xattr.get, dname,
 237                               USER_NN, namespace=NAMESPACE)
 238         _, sname = self._getsymlink()
 239         lists_equal(xattr.list(sname, nofollow=True), [])
 240         self.assertEqual(xattr.list(sname, nofollow=True,
 241                                     namespace=NAMESPACE), [])
 242         tuples_equal(xattr.get_all(sname, nofollow=True), [])
 243         self.assertEqual(xattr.get_all(sname, nofollow=True,
 244                                            namespace=NAMESPACE), [])
 245         self.assertRaises(EnvironmentError, xattr.get, sname,
 246                               USER_NN, namespace=NAMESPACE, nofollow=True)
 247
 248     def testFileByNameDeprecated(self):
 249         """test set and retrieve one attribute by file name (deprecated)"""
 250         fh, fname = self._getfile()
 251         self._checkDeprecated(fname)
 252         os.close(fh)
 253
 254     def testFileByName(self):
 255         """test set and retrieve one attribute by file name"""
 256         fh, fname = self._getfile()
 257         self._checkListSetGet(fname)
 258         self._checkListSetGet(fname, use_ns=True)
 259         os.close(fh)
 260
 261     def testFileByDescriptorDeprecated(self):
 262         """test file descriptor operations (deprecated functions)"""
 263         fh, fname = self._getfile()
 264         self._checkDeprecated(fh)
 265         os.close(fh)
 266
 267     def testFileByDescriptor(self):
 268         """test file descriptor operations"""
 269         fh, fname = self._getfile()
 270         self._checkListSetGet(fh)
 271         self._checkListSetGet(fh, use_ns=True)
 272         os.close(fh)
 273
 274     def testFileByObjectDeprecated(self):
 275         """test file descriptor operations (deprecated functions)"""
 276         fh, fname = self._getfile()
 277         fo = os.fdopen(fh)
 278         self._checkDeprecated(fo)
 279         fo.close()
 280
 281     def testFileByObject(self):
 282         """test file descriptor operations"""
 283         fh, fname = self._getfile()
 284         fo = os.fdopen(fh)
 285         self._checkListSetGet(fo)
 286         self._checkListSetGet(fo, use_ns=True)
 287         fo.close()
 288
 289     def testMixedAccessDeprecated(self):
 290         """test mixed access to file (deprecated functions)"""
 291         fh, fname = self._getfile()
 292         fo = os.fdopen(fh)
 293         lists_equal(xattr.listxattr(fname), [])
 294         xattr.setxattr(fname, USER_ATTR, USER_VAL)
 295         lists_equal(xattr.listxattr(fh), [USER_ATTR])
 296         self.assertEqual(xattr.getxattr(fo, USER_ATTR), USER_VAL)
 297         tuples_equal(xattr.get_all(fo), [(USER_ATTR, USER_VAL)])
 298         tuples_equal(xattr.get_all(fname),
 299                          [(USER_ATTR, USER_VAL)])
 300         fo.close()
 301
 302     def testMixedAccess(self):
 303         """test mixed access to file"""
 304         fh, fname = self._getfile()
 305         fo = os.fdopen(fh)
 306         lists_equal(xattr.list(fname), [])
 307         xattr.set(fname, USER_ATTR, USER_VAL)
 308         lists_equal(xattr.list(fh), [USER_ATTR])
 309         self.assertEqual(xattr.list(fh, namespace=NAMESPACE), [USER_NN])
 310         self.assertEqual(xattr.get(fo, USER_ATTR), USER_VAL)
 311         self.assertEqual(xattr.get(fo, USER_NN, namespace=NAMESPACE),
 312                          USER_VAL)
 313         tuples_equal(xattr.get_all(fo),
 314                          [(USER_ATTR, USER_VAL)])
 315         self.assertEqual(xattr.get_all(fo, namespace=NAMESPACE),
 316                          [(USER_NN, USER_VAL)])
 317         tuples_equal(xattr.get_all(fname),
 318                          [(USER_ATTR, USER_VAL)])
 319         self.assertEqual(xattr.get_all(fname, namespace=NAMESPACE),
 320                          [(USER_NN, USER_VAL)])
 321         fo.close()
 322
 323     def testDirOpsDeprecated(self):
 324         """test attribute setting on directories (deprecated functions)"""
 325         dname = self._getdir()
 326         self._checkDeprecated(dname)
 327
 328     def testDirOps(self):
 329         """test attribute setting on directories"""
 330         dname = self._getdir()
 331         self._checkListSetGet(dname)
 332         self._checkListSetGet(dname, use_ns=True)
 333
 334     def testSymlinkOpsDeprecated(self):
 335         """test symlink operations (deprecated functions)"""
 336         _, sname = self._getsymlink()
 337         self.assertRaises(EnvironmentError, xattr.listxattr, sname)
 338         self._checkDeprecated(sname, symlink=True)
 339         target, sname = self._getsymlink(dangling=False)
 340         xattr.setxattr(target, USER_ATTR, USER_VAL)
 341         lists_equal(xattr.listxattr(target), [USER_ATTR])
 342         lists_equal(xattr.listxattr(sname, True), [])
 343         self.assertRaises(EnvironmentError, xattr.removexattr, sname,
 344                           USER_ATTR, True)
 345         xattr.removexattr(sname, USER_ATTR, False)
 346
 347     def testSymlinkOps(self):
 348         """test symlink operations"""
 349         _, sname = self._getsymlink()
 350         self.assertRaises(EnvironmentError, xattr.list, sname)
 351         self._checkListSetGet(sname, symlink=True)
 352         self._checkListSetGet(sname, symlink=True, use_ns=True)
 353         target, sname = self._getsymlink(dangling=False)
 354         xattr.set(target, USER_ATTR, USER_VAL)
 355         lists_equal(xattr.list(target), [USER_ATTR])
 356         lists_equal(xattr.list(sname, nofollow=True), [])
 357         self.assertRaises(EnvironmentError, xattr.remove, sname,
 358                           USER_ATTR, nofollow=True)
 359         xattr.remove(sname, USER_ATTR, nofollow=False)
 360
 361     def testBinaryPayloadDeprecated(self):
 362         """test binary values (deprecated functions)"""
 363         fh, fname = self._getfile()
 364         os.close(fh)
 365         BINVAL = "abc" + '\0' + "def"
 366         if PY3K:
 367             BINVAL = BINVAL.encode()
 368         xattr.setxattr(fname, USER_ATTR, BINVAL)
 369         lists_equal(xattr.listxattr(fname), [USER_ATTR])
 370         self.assertEqual(xattr.getxattr(fname, USER_ATTR), BINVAL)
 371         tuples_equal(xattr.get_all(fname), [(USER_ATTR, BINVAL)])
 372         xattr.removexattr(fname, USER_ATTR)
 373
 374     def testBinaryPayload(self):
 375         """test binary values"""
 376         fh, fname = self._getfile()
 377         os.close(fh)
 378         BINVAL = "abc" + '\0' + "def"
 379         if PY3K:
 380             BINVAL = BINVAL.encode()
 381         xattr.set(fname, USER_ATTR, BINVAL)
 382         lists_equal(xattr.list(fname), [USER_ATTR])
 383         self.assertEqual(xattr.list(fname, namespace=NAMESPACE), [USER_NN])
 384         self.assertEqual(xattr.get(fname, USER_ATTR), BINVAL)
 385         self.assertEqual(xattr.get(fname, USER_NN,
 386                                    namespace=NAMESPACE), BINVAL)
 387         tuples_equal(xattr.get_all(fname), [(USER_ATTR, BINVAL)])
 388         self.assertEqual(xattr.get_all(fname, namespace=NAMESPACE),
 389                          [(USER_NN, BINVAL)])
 390         xattr.remove(fname, USER_ATTR)
 391
 392     def testManyOpsDeprecated(self):
 393         """test many ops (deprecated functions)"""
 394         fh, fname = self._getfile()
 395         xattr.setxattr(fh, USER_ATTR, USER_VAL)
 396         VL = [USER_ATTR]
 397         for i in range(MANYOPS_COUNT):
 398             lists_equal(xattr.listxattr(fh), VL)
 399         for i in range(MANYOPS_COUNT):
 400             self.assertEqual(xattr.getxattr(fh, USER_ATTR), USER_VAL)
 401         for i in range(MANYOPS_COUNT):
 402             tuples_equal(xattr.get_all(fh),
 403                              [(USER_ATTR, USER_VAL)])
 404
 405     def testManyOps(self):
 406         """test many ops"""
 407         fh, fname = self._getfile()
 408         xattr.set(fh, USER_ATTR, USER_VAL)
 409         VL = [USER_ATTR]
 410         VN = [USER_NN]
 411         for i in range(MANYOPS_COUNT):
 412             lists_equal(xattr.list(fh), VL)
 413             lists_equal(xattr.list(fh, namespace=EMPTY_NS), VL)
 414             self.assertEqual(xattr.list(fh, namespace=NAMESPACE), VN)
 415         for i in range(MANYOPS_COUNT):
 416             self.assertEqual(xattr.get(fh, USER_ATTR), USER_VAL)
 417             self.assertEqual(xattr.get(fh, USER_NN, namespace=NAMESPACE),
 418                              USER_VAL)
 419         for i in range(MANYOPS_COUNT):
 420             tuples_equal(xattr.get_all(fh),
 421                              [(USER_ATTR, USER_VAL)])
 422             self.assertEqual(xattr.get_all(fh, namespace=NAMESPACE),
 423                              [(USER_NN, USER_VAL)])
 424
 425     def testNoneNamespace(self):
 426         fh, fname = self._getfile()
 427         self.assertRaises(TypeError, xattr.get, fh, USER_ATTR,
 428                           namespace=None)
 429
 430     def testEmptyValue(self):
 431         fh, fname = self._getfile()
 432         xattr.set(fh, USER_ATTR, EMPTY_VAL)
 433         self.assertEqual(xattr.get(fh, USER_ATTR), EMPTY_VAL)
 434
 435     def testWrongCall(self):
 436        for call in [xattr.get,
 437                     xattr.list, xattr.listxattr,
 438                     xattr.remove, xattr.removexattr,
 439                     xattr.set, xattr.setxattr,
 440                     xattr.get, xattr.getxattr]:
 441            self.assertRaises(TypeError, call)
 442
 443     def testWrongType(self):
 444         self.assertRaises(TypeError, xattr.get, object(), USER_ATTR)
 445         for call in [xattr.listxattr, xattr.list]:
 446             self.assertRaises(TypeError, call, object())
 447         for call in [xattr.remove, xattr.removexattr,
 448                      xattr.get, xattr.getxattr]:
 449             self.assertRaises(TypeError, call, object(), USER_ATTR)
 450         for call in [xattr.set, xattr.setxattr]:
 451             self.assertRaises(TypeError, call, object(), USER_ATTR, USER_VAL)
 452
 453
 454     def testLargeAttribute(self):
 455         fh, fname = self._getfile()
 456
 457         xattr.set(fh, USER_ATTR, LARGE_VAL)
 458         self.assertEqual(xattr.get(fh, USER_ATTR), LARGE_VAL)
 459
 460
 461 if __name__ == "__main__":
 462     unittest.main()