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