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)
 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)
 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)
 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)
 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, flags=XATTR_REPLACE)
 144         self.assertRaises(EnvironmentError, xattr.set, item,
 145                           self.USER_NN, self.USER_VAL, flags=XATTR_REPLACE,
 146                           namespace=NAMESPACE)
 147         try:
 148             if use_ns:
 149                 xattr.set(item, self.USER_NN, self.USER_VAL,
 150                           namespace=NAMESPACE,
 151                           nofollow=symlink)
 152             else:
 153                 xattr.set(item, self.USER_ATTR, self.USER_VAL,
 154                           nofollow=symlink)
 155         except IOError:
 156             err = sys.exc_info()[1]
 157             if symlink and (err.errno == errno.EPERM or
 158                             err.errno == errno.ENOENT):
 159                 # symlinks may fail, in which case we abort the rest
 160                 # of the test for this case (Linux returns EPERM; OS X
 161                 # returns ENOENT)
 162                 return
 163             raise
 164         self.assertRaises(EnvironmentError, xattr.set, item,
 165                           self.USER_ATTR, self.USER_VAL, flags=XATTR_CREATE)
 166         self.assertRaises(EnvironmentError, xattr.set, item,
 167                           self.USER_NN, self.USER_VAL,
 168                           flags=XATTR_CREATE, namespace=NAMESPACE)
 169         self.checkList(xattr.list(item, nofollow=symlink), [self.USER_ATTR])
 170         self.checkList(xattr.list(item, nofollow=symlink,
 171                                    namespace=EMPTY_NS),
 172                          [self.USER_ATTR])
 173         self.assertEqual(xattr.list(item, namespace=NAMESPACE, nofollow=symlink),
 174                          [self.USER_NN])
 175         self.assertEqual(xattr.get(item, self.USER_ATTR, nofollow=symlink),
 176                          self.USER_VAL)
 177         self.assertEqual(xattr.get(item, self.USER_NN, nofollow=symlink,
 178                                    namespace=NAMESPACE), self.USER_VAL)
 179         self.checkTuples(xattr.get_all(item, nofollow=symlink),
 180                          [(self.USER_ATTR, self.USER_VAL)])
 181         self.assertEqual(xattr.get_all(item, nofollow=symlink,
 182                                        namespace=NAMESPACE),
 183                          [(self.USER_NN, self.USER_VAL)])
 184         if use_ns:
 185             xattr.remove(item, self.USER_NN, namespace=NAMESPACE)
 186         else:
 187             xattr.remove(item, self.USER_ATTR)
 188         self.checkList(xattr.list(item, symlink), [])
 189         self.checkTuples(xattr.get_all(item, nofollow=symlink),
 190                          [])
 191         self.assertRaises(EnvironmentError, xattr.remove,
 192                           item, self.USER_ATTR, nofollow=symlink)
 193         self.assertRaises(EnvironmentError, xattr.remove, item,
 194                           self.USER_NN, namespace=NAMESPACE, nofollow=symlink)
 195
 196     def testNoXattrDeprecated(self):
 197         """test no attributes (deprecated functions)"""
 198         fh, fname = self._getfile()
 199         self.checkList(xattr.listxattr(fname), [])
 200         self.checkTuples(xattr.get_all(fname), [])
 201         self.assertRaises(EnvironmentError, xattr.getxattr, fname,
 202                               self.USER_ATTR)
 203         dname = self._getdir()
 204         self.checkList(xattr.listxattr(dname), [])
 205         self.checkTuples(xattr.get_all(dname), [])
 206         self.assertRaises(EnvironmentError, xattr.getxattr, dname,
 207                               self.USER_ATTR)
 208         _, sname = self._getsymlink()
 209         self.checkList(xattr.listxattr(sname, True), [])
 210         self.checkTuples(xattr.get_all(sname, nofollow=True), [])
 211         self.assertRaises(EnvironmentError, xattr.getxattr, fname,
 212                               self.USER_ATTR, True)
 213
 214
 215     def testNoXattr(self):
 216         """test no attributes"""
 217         fh, fname = self._getfile()
 218         self.checkList(xattr.list(fname), [])
 219         self.assertEqual(xattr.list(fname, namespace=NAMESPACE), [])
 220         self.checkTuples(xattr.get_all(fname), [])
 221         self.assertEqual(xattr.get_all(fname, namespace=NAMESPACE), [])
 222         self.assertRaises(EnvironmentError, xattr.get, fname,
 223                               self.USER_NN, namespace=NAMESPACE)
 224         dname = self._getdir()
 225         self.checkList(xattr.list(dname), [])
 226         self.assertEqual(xattr.list(dname, namespace=NAMESPACE), [])
 227         self.checkTuples(xattr.get_all(dname), [])
 228         self.assertEqual(xattr.get_all(dname, namespace=NAMESPACE), [])
 229         self.assertRaises(EnvironmentError, xattr.get, dname,
 230                               self.USER_NN, namespace=NAMESPACE)
 231         _, sname = self._getsymlink()
 232         self.checkList(xattr.list(sname, nofollow=True), [])
 233         self.assertEqual(xattr.list(sname, nofollow=True,
 234                                     namespace=NAMESPACE), [])
 235         self.checkTuples(xattr.get_all(sname, nofollow=True), [])
 236         self.assertEqual(xattr.get_all(sname, nofollow=True,
 237                                            namespace=NAMESPACE), [])
 238         self.assertRaises(EnvironmentError, xattr.get, sname,
 239                               self.USER_NN, namespace=NAMESPACE, nofollow=True)
 240
 241     def testFileByNameDeprecated(self):
 242         """test set and retrieve one attribute by file name (deprecated)"""
 243         fh, fname = self._getfile()
 244         self._checkDeprecated(fname)
 245         os.close(fh)
 246
 247     def testFileByName(self):
 248         """test set and retrieve one attribute by file name"""
 249         fh, fname = self._getfile()
 250         self._checkListSetGet(fname)
 251         self._checkListSetGet(fname, use_ns=True)
 252         os.close(fh)
 253
 254     def testFileByDescriptorDeprecated(self):
 255         """test file descriptor operations (deprecated functions)"""
 256         fh, fname = self._getfile()
 257         self._checkDeprecated(fh)
 258         os.close(fh)
 259
 260     def testFileByDescriptor(self):
 261         """test file descriptor operations"""
 262         fh, fname = self._getfile()
 263         self._checkListSetGet(fh)
 264         self._checkListSetGet(fh, use_ns=True)
 265         os.close(fh)
 266
 267     def testFileByObjectDeprecated(self):
 268         """test file descriptor operations (deprecated functions)"""
 269         fh, fname = self._getfile()
 270         fo = os.fdopen(fh)
 271         self._checkDeprecated(fo)
 272         fo.close()
 273
 274     def testFileByObject(self):
 275         """test file descriptor operations"""
 276         fh, fname = self._getfile()
 277         fo = os.fdopen(fh)
 278         self._checkListSetGet(fo)
 279         self._checkListSetGet(fo, use_ns=True)
 280         fo.close()
 281
 282     def testMixedAccessDeprecated(self):
 283         """test mixed access to file (deprecated functions)"""
 284         fh, fname = self._getfile()
 285         fo = os.fdopen(fh)
 286         self.checkList(xattr.listxattr(fname), [])
 287         xattr.setxattr(fname, self.USER_ATTR, self.USER_VAL)
 288         self.checkList(xattr.listxattr(fh), [self.USER_ATTR])
 289         self.assertEqual(xattr.getxattr(fo, self.USER_ATTR), self.USER_VAL)
 290         self.checkTuples(xattr.get_all(fo), [(self.USER_ATTR, self.USER_VAL)])
 291         self.checkTuples(xattr.get_all(fname),
 292                          [(self.USER_ATTR, self.USER_VAL)])
 293         fo.close()
 294
 295     def testMixedAccess(self):
 296         """test mixed access to file"""
 297         fh, fname = self._getfile()
 298         fo = os.fdopen(fh)
 299         self.checkList(xattr.list(fname), [])
 300         xattr.set(fname, self.USER_ATTR, self.USER_VAL)
 301         self.checkList(xattr.list(fh), [self.USER_ATTR])
 302         self.assertEqual(xattr.list(fh, namespace=NAMESPACE), [self.USER_NN])
 303         self.assertEqual(xattr.get(fo, self.USER_ATTR), self.USER_VAL)
 304         self.assertEqual(xattr.get(fo, self.USER_NN, namespace=NAMESPACE),
 305                          self.USER_VAL)
 306         self.checkTuples(xattr.get_all(fo),
 307                          [(self.USER_ATTR, self.USER_VAL)])
 308         self.assertEqual(xattr.get_all(fo, namespace=NAMESPACE),
 309                          [(self.USER_NN, self.USER_VAL)])
 310         self.checkTuples(xattr.get_all(fname),
 311                          [(self.USER_ATTR, self.USER_VAL)])
 312         self.assertEqual(xattr.get_all(fname, namespace=NAMESPACE),
 313                          [(self.USER_NN, self.USER_VAL)])
 314         fo.close()
 315
 316     def testDirOpsDeprecated(self):
 317         """test attribute setting on directories (deprecated functions)"""
 318         dname = self._getdir()
 319         self._checkDeprecated(dname)
 320
 321     def testDirOps(self):
 322         """test attribute setting on directories"""
 323         dname = self._getdir()
 324         self._checkListSetGet(dname)
 325         self._checkListSetGet(dname, use_ns=True)
 326
 327     def testSymlinkOpsDeprecated(self):
 328         """test symlink operations (deprecated functions)"""
 329         _, sname = self._getsymlink()
 330         self.assertRaises(EnvironmentError, xattr.listxattr, sname)
 331         self._checkDeprecated(sname, symlink=True)
 332         target, sname = self._getsymlink(dangling=False)
 333         xattr.setxattr(target, self.USER_ATTR, self.USER_VAL)
 334         self.checkList(xattr.listxattr(target), [self.USER_ATTR])
 335         self.checkList(xattr.listxattr(sname, True), [])
 336         self.assertRaises(EnvironmentError, xattr.removexattr, sname,
 337                           self.USER_ATTR, True)
 338         xattr.removexattr(sname, self.USER_ATTR, False)
 339
 340     def testSymlinkOps(self):
 341         """test symlink operations"""
 342         _, sname = self._getsymlink()
 343         self.assertRaises(EnvironmentError, xattr.list, sname)
 344         self._checkListSetGet(sname, symlink=True)
 345         self._checkListSetGet(sname, symlink=True, use_ns=True)
 346         target, sname = self._getsymlink(dangling=False)
 347         xattr.set(target, self.USER_ATTR, self.USER_VAL)
 348         self.checkList(xattr.list(target), [self.USER_ATTR])
 349         self.checkList(xattr.list(sname, nofollow=True), [])
 350         self.assertRaises(EnvironmentError, xattr.remove, sname,
 351                           self.USER_ATTR, nofollow=True)
 352         xattr.remove(sname, self.USER_ATTR, nofollow=False)
 353
 354     def testBinaryPayloadDeprecated(self):
 355         """test binary values (deprecated functions)"""
 356         fh, fname = self._getfile()
 357         os.close(fh)
 358         BINVAL = "abc" + '\0' + "def"
 359         if PY3K:
 360             BINVAL = BINVAL.encode()
 361         xattr.setxattr(fname, self.USER_ATTR, BINVAL)
 362         self.checkList(xattr.listxattr(fname), [self.USER_ATTR])
 363         self.assertEqual(xattr.getxattr(fname, self.USER_ATTR), BINVAL)
 364         self.checkTuples(xattr.get_all(fname), [(self.USER_ATTR, BINVAL)])
 365         xattr.removexattr(fname, self.USER_ATTR)
 366
 367     def testBinaryPayload(self):
 368         """test binary values"""
 369         fh, fname = self._getfile()
 370         os.close(fh)
 371         BINVAL = "abc" + '\0' + "def"
 372         if PY3K:
 373             BINVAL = BINVAL.encode()
 374         xattr.set(fname, self.USER_ATTR, BINVAL)
 375         self.checkList(xattr.list(fname), [self.USER_ATTR])
 376         self.assertEqual(xattr.list(fname, namespace=NAMESPACE), [self.USER_NN])
 377         self.assertEqual(xattr.get(fname, self.USER_ATTR), BINVAL)
 378         self.assertEqual(xattr.get(fname, self.USER_NN,
 379                                    namespace=NAMESPACE), BINVAL)
 380         self.checkTuples(xattr.get_all(fname), [(self.USER_ATTR, BINVAL)])
 381         self.assertEqual(xattr.get_all(fname, namespace=NAMESPACE),
 382                          [(self.USER_NN, BINVAL)])
 383         xattr.remove(fname, self.USER_ATTR)
 384
 385     def testManyOpsDeprecated(self):
 386         """test many ops (deprecated functions)"""
 387         fh, fname = self._getfile()
 388         xattr.setxattr(fh, self.USER_ATTR, self.USER_VAL)
 389         VL = [self.USER_ATTR]
 390         for i in range(self.MANYOPS_COUNT):
 391             self.checkList(xattr.listxattr(fh), VL)
 392         for i in range(self.MANYOPS_COUNT):
 393             self.assertEqual(xattr.getxattr(fh, self.USER_ATTR), self.USER_VAL)
 394         for i in range(self.MANYOPS_COUNT):
 395             self.checkTuples(xattr.get_all(fh),
 396                              [(self.USER_ATTR, self.USER_VAL)])
 397
 398     def testManyOps(self):
 399         """test many ops"""
 400         fh, fname = self._getfile()
 401         xattr.set(fh, self.USER_ATTR, self.USER_VAL)
 402         VL = [self.USER_ATTR]
 403         VN = [self.USER_NN]
 404         for i in range(self.MANYOPS_COUNT):
 405             self.checkList(xattr.list(fh), VL)
 406             self.checkList(xattr.list(fh, namespace=EMPTY_NS), VL)
 407             self.assertEqual(xattr.list(fh, namespace=NAMESPACE), VN)
 408         for i in range(self.MANYOPS_COUNT):
 409             self.assertEqual(xattr.get(fh, self.USER_ATTR), self.USER_VAL)
 410             self.assertEqual(xattr.get(fh, self.USER_NN, namespace=NAMESPACE),
 411                              self.USER_VAL)
 412         for i in range(self.MANYOPS_COUNT):
 413             self.checkTuples(xattr.get_all(fh),
 414                              [(self.USER_ATTR, self.USER_VAL)])
 415             self.assertEqual(xattr.get_all(fh, namespace=NAMESPACE),
 416                              [(self.USER_NN, self.USER_VAL)])
 417
 418     def testNoneNamespace(self):
 419         fh, fname = self._getfile()
 420         self.assertRaises(TypeError, xattr.get, fh, self.USER_ATTR,
 421                           namespace=None)
 422
 423     def testEmptyValue(self):
 424         fh, fname = self._getfile()
 425         xattr.set(fh, self.USER_ATTR, self.EMPTY_VAL)
 426         self.assertEqual(xattr.get(fh, self.USER_ATTR), self.EMPTY_VAL)
 427
 428     def testWrongCall(self):
 429        for call in [xattr.get,
 430                     xattr.list, xattr.listxattr,
 431                     xattr.remove, xattr.removexattr,
 432                     xattr.set, xattr.setxattr,
 433                     xattr.get, xattr.getxattr]:
 434            self.assertRaises(TypeError, call)
 435
 436     def testWrongType(self):
 437         self.assertRaises(TypeError, xattr.get, object(), self.USER_ATTR)
 438         for call in [xattr.listxattr, xattr.list]:
 439             self.assertRaises(TypeError, call, object())
 440         for call in [xattr.remove, xattr.removexattr,
 441                      xattr.get, xattr.getxattr]:
 442             self.assertRaises(TypeError, call, object(), self.USER_ATTR)
 443         for call in [xattr.set, xattr.setxattr]:
 444             self.assertRaises(TypeError, call, object(), self.USER_ATTR, self.USER_VAL)
 445
 446
 447     def testLargeAttribute(self):
 448         fh, fname = self._getfile()
 449
 450         xattr.set(fh, self.USER_ATTR, self.LARGE_VAL)
 451         self.assertEqual(xattr.get(fh, self.USER_ATTR), self.LARGE_VAL)
 452
 453
 454 if __name__ == "__main__":
 455     unittest.main()