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