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