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