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