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