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