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