twisted/test/test_paths.py

   1 # Copyright (c) Twisted Matrix Laboratories.
   2 # See LICENSE for details.
   3
   4 """
   5 Test cases covering L{twisted.python.filepath} and L{twisted.python.zippath}.
   6 """
   7
   8 import os, time, pickle, errno, zipfile, stat
   9
  10 from twisted.python.compat import set
  11 from twisted.python.win32 import WindowsError, ERROR_DIRECTORY
  12 from twisted.python import filepath
  13 from twisted.python.zippath import ZipArchive
  14 from twisted.python.runtime import platform
  15
  16 from twisted.trial import unittest
  17
  18 from zope.interface.verify import verifyObject
  19
  20
  21 class AbstractFilePathTestCase(unittest.TestCase):
  22
  23     f1content = "file 1"
  24     f2content = "file 2"
  25
  26
  27     def _mkpath(self, *p):
  28         x = os.path.abspath(os.path.join(self.cmn, *p))
  29         self.all.append(x)
  30         return x
  31
  32
  33     def subdir(self, *dirname):
  34         os.mkdir(self._mkpath(*dirname))
  35
  36
  37     def subfile(self, *dirname):
  38         return open(self._mkpath(*dirname), "wb")
  39
  40
  41     def setUp(self):
  42         self.now = time.time()
  43         cmn = self.cmn = os.path.abspath(self.mktemp())
  44         self.all = [cmn]
  45         os.mkdir(cmn)
  46         self.subdir("sub1")
  47         f = self.subfile("file1")
  48         f.write(self.f1content)
  49         f.close()
  50         f = self.subfile("sub1", "file2")
  51         f.write(self.f2content)
  52         f.close()
  53         self.subdir('sub3')
  54         f = self.subfile("sub3", "file3.ext1")
  55         f.close()
  56         f = self.subfile("sub3", "file3.ext2")
  57         f.close()
  58         f = self.subfile("sub3", "file3.ext3")
  59         f.close()
  60         self.path = filepath.FilePath(cmn)
  61         self.root = filepath.FilePath("/")
  62
  63
  64     def test_segmentsFromPositive(self):
  65         """
  66         Verify that the segments between two paths are correctly identified.
  67         """
  68         self.assertEqual(
  69             self.path.child("a").child("b").child("c").segmentsFrom(self.path),
  70             ["a", "b", "c"])
  71
  72     def test_segmentsFromNegative(self):
  73         """
  74         Verify that segmentsFrom notices when the ancestor isn't an ancestor.
  75         """
  76         self.assertRaises(
  77             ValueError,
  78             self.path.child("a").child("b").child("c").segmentsFrom,
  79                 self.path.child("d").child("c").child("e"))
  80
  81
  82     def test_walk(self):
  83         """
  84         Verify that walking the path gives the same result as the known file
  85         hierarchy.
  86         """
  87         x = [foo.path for foo in self.path.walk()]
  88         self.assertEqual(set(x), set(self.all))
  89
  90
  91     def test_parents(self):
  92         """
  93         L{FilePath.parents()} should return an iterator of every ancestor of
  94         the L{FilePath} in question.
  95         """
  96         L = []
  97         pathobj = self.path.child("a").child("b").child("c")
  98         fullpath = pathobj.path
  99         lastpath = fullpath
 100         thispath = os.path.dirname(fullpath)
 101         while lastpath != self.root.path:
 102             L.append(thispath)
 103             lastpath = thispath
 104             thispath = os.path.dirname(thispath)
 105         self.assertEqual([x.path for x in pathobj.parents()], L)
 106
 107
 108     def test_validSubdir(self):
 109         """
 110         Verify that a valid subdirectory will show up as a directory, but not as a
 111         file, not as a symlink, and be listable.
 112         """
 113         sub1 = self.path.child('sub1')
 114         self.failUnless(sub1.exists(),
 115                         "This directory does exist.")
 116         self.failUnless(sub1.isdir(),
 117                         "It's a directory.")
 118         self.failUnless(not sub1.isfile(),
 119                         "It's a directory.")
 120         self.failUnless(not sub1.islink(),
 121                         "It's a directory.")
 122         self.assertEqual(sub1.listdir(),
 123                              ['file2'])
 124
 125
 126     def test_invalidSubdir(self):
 127         """
 128         Verify that a subdirectory that doesn't exist is reported as such.
 129         """
 130         sub2 = self.path.child('sub2')
 131         self.failIf(sub2.exists(),
 132                     "This directory does not exist.")
 133
 134     def test_validFiles(self):
 135         """
 136         Make sure that we can read existent non-empty files.
 137         """
 138         f1 = self.path.child('file1')
 139         self.assertEqual(f1.open().read(), self.f1content)
 140         f2 = self.path.child('sub1').child('file2')
 141         self.assertEqual(f2.open().read(), self.f2content)
 142
 143
 144     def test_multipleChildSegments(self):
 145         """
 146         C{fp.descendant([a, b, c])} returns the same L{FilePath} as is returned
 147         by C{fp.child(a).child(b).child(c)}.
 148         """
 149         multiple = self.path.descendant(['a', 'b', 'c'])
 150         single = self.path.child('a').child('b').child('c')
 151         self.assertEqual(multiple, single)
 152
 153
 154     def test_dictionaryKeys(self):
 155         """
 156         Verify that path instances are usable as dictionary keys.
 157         """
 158         f1 = self.path.child('file1')
 159         f1prime = self.path.child('file1')
 160         f2 = self.path.child('file2')
 161         dictoid = {}
 162         dictoid[f1] = 3
 163         dictoid[f1prime] = 4
 164         self.assertEqual(dictoid[f1], 4)
 165         self.assertEqual(dictoid.keys(), [f1])
 166         self.assertIdentical(dictoid.keys()[0], f1)
 167         self.assertNotIdentical(dictoid.keys()[0], f1prime) # sanity check
 168         dictoid[f2] = 5
 169         self.assertEqual(dictoid[f2], 5)
 170         self.assertEqual(len(dictoid), 2)
 171
 172
 173     def test_dictionaryKeyWithString(self):
 174         """
 175         Verify that path instances are usable as dictionary keys which do not clash
 176         with their string counterparts.
 177         """
 178         f1 = self.path.child('file1')
 179         dictoid = {f1: 'hello'}
 180         dictoid[f1.path] = 'goodbye'
 181         self.assertEqual(len(dictoid), 2)
 182
 183
 184     def test_childrenNonexistentError(self):
 185         """
 186         Verify that children raises the appropriate exception for non-existent
 187         directories.
 188         """
 189         self.assertRaises(filepath.UnlistableError,
 190                           self.path.child('not real').children)
 191
 192     def test_childrenNotDirectoryError(self):
 193         """
 194         Verify that listdir raises the appropriate exception for attempting to list
 195         a file rather than a directory.
 196         """
 197         self.assertRaises(filepath.UnlistableError,
 198                           self.path.child('file1').children)
 199
 200
 201     def test_newTimesAreFloats(self):
 202         """
 203         Verify that all times returned from the various new time functions are ints
 204         (and hopefully therefore 'high precision').
 205         """
 206         for p in self.path, self.path.child('file1'):
 207             self.assertEqual(type(p.getAccessTime()), float)
 208             self.assertEqual(type(p.getModificationTime()), float)
 209             self.assertEqual(type(p.getStatusChangeTime()), float)
 210
 211
 212     def test_oldTimesAreInts(self):
 213         """
 214         Verify that all times returned from the various time functions are
 215         integers, for compatibility.
 216         """
 217         for p in self.path, self.path.child('file1'):
 218             self.assertEqual(type(p.getatime()), int)
 219             self.assertEqual(type(p.getmtime()), int)
 220             self.assertEqual(type(p.getctime()), int)
 221
 222
 223
 224 class FakeWindowsPath(filepath.FilePath):
 225     """
 226     A test version of FilePath which overrides listdir to raise L{WindowsError}.
 227     """
 228
 229     def listdir(self):
 230         """
 231         @raise WindowsError: always.
 232         """
 233         raise WindowsError(
 234             ERROR_DIRECTORY,
 235             "A directory's validness was called into question")
 236
 237
 238 class ListingCompatibilityTests(unittest.TestCase):
 239     """
 240     These tests verify compatibility with legacy behavior of directory listing.
 241     """
 242
 243     def test_windowsErrorExcept(self):
 244         """
 245         Verify that when a WindowsError is raised from listdir, catching
 246         WindowsError works.
 247         """
 248         fwp = FakeWindowsPath(self.mktemp())
 249         self.assertRaises(filepath.UnlistableError, fwp.children)
 250         self.assertRaises(WindowsError, fwp.children)
 251
 252
 253     def test_alwaysCatchOSError(self):
 254         """
 255         Verify that in the normal case where a directory does not exist, we will
 256         get an OSError.
 257         """
 258         fp = filepath.FilePath(self.mktemp())
 259         self.assertRaises(OSError, fp.children)
 260
 261
 262     def test_keepOriginalAttributes(self):
 263         """
 264         Verify that the Unlistable exception raised will preserve the attributes of
 265         the previously-raised exception.
 266         """
 267         fp = filepath.FilePath(self.mktemp())
 268         ose = self.assertRaises(OSError, fp.children)
 269         d1 = ose.__dict__.keys()
 270         d1.remove('originalException')
 271         d2 = ose.originalException.__dict__.keys()
 272         d1.sort()
 273         d2.sort()
 274         self.assertEqual(d1, d2)
 275
 276
 277
 278 def zipit(dirname, zfname):
 279     """
 280     Create a zipfile on zfname, containing the contents of dirname'
 281     """
 282     zf = zipfile.ZipFile(zfname, "w")
 283     for root, ignored, files, in os.walk(dirname):
 284         for fname in files:
 285             fspath = os.path.join(root, fname)
 286             arcpath = os.path.join(root, fname)[len(dirname)+1:]
 287             # print fspath, '=>', arcpath
 288             zf.write(fspath, arcpath)
 289     zf.close()
 290
 291
 292
 293 class ZipFilePathTestCase(AbstractFilePathTestCase):
 294     """
 295     Test various L{ZipPath} path manipulations as well as reprs for L{ZipPath}
 296     and L{ZipArchive}.
 297     """
 298     def setUp(self):
 299         AbstractFilePathTestCase.setUp(self)
 300         zipit(self.cmn, self.cmn + '.zip')
 301         self.path = ZipArchive(self.cmn + '.zip')
 302         self.root = self.path
 303         self.all = [x.replace(self.cmn, self.cmn + '.zip') for x in self.all]
 304
 305
 306     def test_verifyObject(self):
 307         """
 308         ZipPaths implement IFilePath.
 309         """
 310
 311         self.assertTrue(verifyObject(filepath.IFilePath, self.path))
 312
 313
 314     def test_zipPathRepr(self):
 315         """
 316         Make sure that invoking ZipPath's repr prints the correct class name
 317         and an absolute path to the zip file.
 318         """
 319         child = self.path.child("foo")
 320         pathRepr = "ZipPath(%r)" % (
 321             os.path.abspath(self.cmn + ".zip" + os.sep + 'foo'),)
 322
 323         # Check for an absolute path
 324         self.assertEqual(repr(child), pathRepr)
 325
 326         # Create a path to the file rooted in the current working directory
 327         relativeCommon = self.cmn.replace(os.getcwd() + os.sep, "", 1) + ".zip"
 328         relpath = ZipArchive(relativeCommon)
 329         child = relpath.child("foo")
 330
 331         # Check using a path without the cwd prepended
 332         self.assertEqual(repr(child), pathRepr)
 333
 334
 335     def test_zipPathReprParentDirSegment(self):
 336         """
 337         The repr of a ZipPath with C{".."} in the internal part of its path
 338         includes the C{".."} rather than applying the usual parent directory
 339         meaning.
 340         """
 341         child = self.path.child("foo").child("..").child("bar")
 342         pathRepr = "ZipPath(%r)" % (
 343             self.cmn + ".zip" + os.sep.join(["", "foo", "..", "bar"]))
 344         self.assertEqual(repr(child), pathRepr)
 345
 346
 347     def test_zipPathReprEscaping(self):
 348         """
 349         Bytes in the ZipPath path which have special meaning in Python
 350         string literals are escaped in the ZipPath repr.
 351         """
 352         child = self.path.child("'")
 353         path = self.cmn + ".zip" + os.sep.join(["", "'"])
 354         pathRepr = "ZipPath('%s')" % (path.encode('string-escape'),)
 355         self.assertEqual(repr(child), pathRepr)
 356
 357
 358     def test_zipArchiveRepr(self):
 359         """
 360         Make sure that invoking ZipArchive's repr prints the correct class
 361         name and an absolute path to the zip file.
 362         """
 363         pathRepr = 'ZipArchive(%r)' % (os.path.abspath(self.cmn + '.zip'),)
 364
 365         # Check for an absolute path
 366         self.assertEqual(repr(self.path), pathRepr)
 367
 368         # Create a path to the file rooted in the current working directory
 369         relativeCommon = self.cmn.replace(os.getcwd() + os.sep, "", 1) + ".zip"
 370         relpath = ZipArchive(relativeCommon)
 371
 372         # Check using a path without the cwd prepended
 373         self.assertEqual(repr(relpath), pathRepr)
 374
 375
 376
 377 class ExplodingFile:
 378     """
 379     A C{file}-alike which raises exceptions from its I/O methods and keeps track
 380     of whether it has been closed.
 381
 382     @ivar closed: A C{bool} which is C{False} until C{close} is called, then it
 383         is C{True}.
 384     """
 385     closed = False
 386
 387     def read(self, n=0):
 388         """
 389         @raise IOError: Always raised.
 390         """
 391         raise IOError()
 392
 393
 394     def write(self, what):
 395         """
 396         @raise IOError: Always raised.
 397         """
 398         raise IOError()
 399
 400
 401     def close(self):
 402         """
 403         Mark the file as having been closed.
 404         """
 405         self.closed = True
 406
 407
 408
 409 class TrackingFilePath(filepath.FilePath):
 410     """
 411     A subclass of L{filepath.FilePath} which maintains a list of all other paths
 412     created by clonePath.
 413
 414     @ivar trackingList: A list of all paths created by this path via
 415         C{clonePath} (which also includes paths created by methods like
 416         C{parent}, C{sibling}, C{child}, etc (and all paths subsequently created
 417         by those paths, etc).
 418
 419     @type trackingList: C{list} of L{TrackingFilePath}
 420
 421     @ivar openedFiles: A list of all file objects opened by this
 422         L{TrackingFilePath} or any other L{TrackingFilePath} in C{trackingList}.
 423
 424     @type openedFiles: C{list} of C{file}
 425     """
 426
 427     def __init__(self, path, alwaysCreate=False, trackingList=None):
 428         filepath.FilePath.__init__(self, path, alwaysCreate)
 429         if trackingList is None:
 430             trackingList = []
 431         self.trackingList = trackingList
 432         self.openedFiles = []
 433
 434
 435     def open(self, *a, **k):
 436         """
 437         Override 'open' to track all files opened by this path.
 438         """
 439         f = filepath.FilePath.open(self, *a, **k)
 440         self.openedFiles.append(f)
 441         return f
 442
 443
 444     def openedPaths(self):
 445         """
 446         Return a list of all L{TrackingFilePath}s associated with this
 447         L{TrackingFilePath} that have had their C{open()} method called.
 448         """
 449         return [path for path in self.trackingList if path.openedFiles]
 450
 451
 452     def clonePath(self, name):
 453         """
 454         Override L{filepath.FilePath.clonePath} to give the new path a reference
 455         to the same tracking list.
 456         """
 457         clone = TrackingFilePath(name, trackingList=self.trackingList)
 458         self.trackingList.append(clone)
 459         return clone
 460
 461
 462
 463 class ExplodingFilePath(filepath.FilePath):
 464     """
 465     A specialized L{FilePath} which always returns an instance of
 466     L{ExplodingFile} from its C{open} method.
 467
 468     @ivar fp: The L{ExplodingFile} instance most recently returned from the
 469         C{open} method.
 470     """
 471
 472     def __init__(self, pathName, originalExploder=None):
 473         """
 474         Initialize an L{ExplodingFilePath} with a name and a reference to the
 475
 476         @param pathName: The path name as passed to L{filepath.FilePath}.
 477         @type pathName: C{str}
 478
 479         @param originalExploder: The L{ExplodingFilePath} to associate opened
 480         files with.
 481         @type originalExploder: L{ExplodingFilePath}
 482         """
 483         filepath.FilePath.__init__(self, pathName)
 484         if originalExploder is None:
 485             originalExploder = self
 486         self._originalExploder = originalExploder
 487
 488
 489     def open(self, mode=None):
 490         """
 491         Create, save, and return a new C{ExplodingFile}.
 492
 493         @param mode: Present for signature compatibility.  Ignored.
 494
 495         @return: A new C{ExplodingFile}.
 496         """
 497         f = self._originalExploder.fp = ExplodingFile()
 498         return f
 499
 500
 501     def clonePath(self, name):
 502         return ExplodingFilePath(name, self._originalExploder)
 503
 504
 505
 506 class PermissionsTestCase(unittest.TestCase):
 507     """
 508     Test Permissions and RWX classes
 509     """
 510
 511     def assertNotUnequal(self, first, second, msg=None):
 512         """
 513         Tests that C{first} != C{second} is false.  This method tests the
 514         __ne__ method, as opposed to L{assertEqual} (C{first} == C{second}),
 515         which tests the __eq__ method.
 516
 517         Note: this should really be part of trial
 518         """
 519         if first != second:
 520             if msg is None:
 521                 msg = '';
 522             if len(msg) > 0:
 523                 msg += '\n'
 524             raise self.failureException(
 525                 '%snot not unequal (__ne__ not implemented correctly):'
 526                 '\na = %s\nb = %s\n'
 527                 % (msg, pformat(first), pformat(second)))
 528         return first
 529
 530
 531     def test_rwxFromBools(self):
 532         """
 533         L{RWX}'s constructor takes a set of booleans
 534         """
 535         for r in (True, False):
 536             for w in (True, False):
 537                 for x in (True, False):
 538                     rwx = filepath.RWX(r, w, x)
 539                     self.assertEqual(rwx.read, r)
 540                     self.assertEqual(rwx.write, w)
 541                     self.assertEqual(rwx.execute, x)
 542         rwx = filepath.RWX(True, True, True)
 543         self.assertTrue(rwx.read and rwx.write and rwx.execute)
 544
 545
 546     def test_rwxEqNe(self):
 547         """
 548         L{RWX}'s created with the same booleans are equivalent.  If booleans
 549         are different, they are not equal.
 550         """
 551         for r in (True, False):
 552             for w in (True, False):
 553                 for x in (True, False):
 554                     self.assertEqual(filepath.RWX(r, w, x),
 555                                       filepath.RWX(r, w, x))
 556                     self.assertNotUnequal(filepath.RWX(r, w, x),
 557                                           filepath.RWX(r, w, x))
 558         self.assertNotEqual(filepath.RWX(True, True, True),
 559                             filepath.RWX(True, True, False))
 560         self.assertNotEqual(3, filepath.RWX(True, True, True))
 561
 562
 563     def test_rwxShorthand(self):
 564         """
 565         L{RWX}'s shorthand string should be 'rwx' if read, write, and execute
 566         permission bits are true.  If any of those permissions bits are false,
 567         the character is replaced by a '-'.
 568         """
 569
 570         def getChar(val, letter):
 571             if val:
 572                 return letter
 573             return '-'
 574
 575         for r in (True, False):
 576             for w in (True, False):
 577                 for x in (True, False):
 578                     rwx = filepath.RWX(r, w, x)
 579                     self.assertEqual(rwx.shorthand(),
 580                                       getChar(r, 'r') +
 581                                       getChar(w, 'w') +
 582                                       getChar(x, 'x'))
 583         self.assertEqual(filepath.RWX(True, False, True).shorthand(), "r-x")
 584
 585
 586     def test_permissionsFromStat(self):
 587         """
 588         L{Permissions}'s constructor takes a valid permissions bitmask and
 589         parsaes it to produce the correct set of boolean permissions.
 590         """
 591         def _rwxFromStat(statModeInt, who):
 592             def getPermissionBit(what, who):
 593                 return (statModeInt &
 594                         getattr(stat, "S_I%s%s" % (what, who))) > 0
 595             return filepath.RWX(*[getPermissionBit(what, who) for what in
 596                          ('R', 'W', 'X')])
 597
 598         for u in range(0, 8):
 599             for g in range(0, 8):
 600                 for o in range(0, 8):
 601                     chmodString = "%d%d%d" % (u, g, o)
 602                     chmodVal = int(chmodString, 8)
 603                     perm = filepath.Permissions(chmodVal)
 604                     self.assertEqual(perm.user,
 605                                       _rwxFromStat(chmodVal, "USR"),
 606                                       "%s: got user: %s" %
 607                                       (chmodString, perm.user))
 608                     self.assertEqual(perm.group,
 609                                       _rwxFromStat(chmodVal, "GRP"),
 610                                       "%s: got group: %s" %
 611                                       (chmodString, perm.group))
 612                     self.assertEqual(perm.other,
 613                                       _rwxFromStat(chmodVal, "OTH"),
 614                                       "%s: got other: %s" %
 615                                       (chmodString, perm.other))
 616         perm = filepath.Permissions(0777)
 617         for who in ("user", "group", "other"):
 618             for what in ("read", "write", "execute"):
 619                 self.assertTrue(getattr(getattr(perm, who), what))
 620
 621
 622     def test_permissionsEq(self):
 623         """
 624         Two L{Permissions}'s that are created with the same bitmask
 625         are equivalent
 626         """
 627         self.assertEqual(filepath.Permissions(0777),
 628                           filepath.Permissions(0777))
 629         self.assertNotUnequal(filepath.Permissions(0777),
 630                               filepath.Permissions(0777))
 631         self.assertNotEqual(filepath.Permissions(0777),
 632                             filepath.Permissions(0700))
 633         self.assertNotEqual(3, filepath.Permissions(0777))
 634
 635
 636     def test_permissionsShorthand(self):
 637         """
 638         L{Permissions}'s shorthand string is the RWX shorthand string for its
 639         user permission bits, group permission bits, and other permission bits
 640         concatenated together, without a space.
 641         """
 642         for u in range(0, 8):
 643             for g in range(0, 8):
 644                 for o in range(0, 8):
 645                     perm = filepath.Permissions(eval("0%d%d%d" % (u, g, o)))
 646                     self.assertEqual(perm.shorthand(),
 647                                       ''.join(x.shorthand() for x in (
 648                                           perm.user, perm.group, perm.other)))
 649         self.assertEqual(filepath.Permissions(0770).shorthand(), "rwxrwx---")
 650
 651
 652
 653 class FilePathTestCase(AbstractFilePathTestCase):
 654     """
 655     Test various L{FilePath} path manipulations.
 656     """
 657
 658
 659     def test_verifyObject(self):
 660         """
 661         FilePaths implement IFilePath.
 662         """
 663
 664         self.assertTrue(verifyObject(filepath.IFilePath, self.path))
 665
 666
 667     def test_chmod(self):
 668         """
 669         L{FilePath.chmod} modifies the permissions of
 670         the passed file as expected (using C{os.stat} to check). We use some
 671         basic modes that should work everywhere (even on Windows).
 672         """
 673         for mode in (0555, 0777):
 674             self.path.child("sub1").chmod(mode)
 675             self.assertEqual(
 676                 stat.S_IMODE(os.stat(self.path.child("sub1").path).st_mode),
 677                 mode)
 678
 679
 680     def symlink(self, target, name):
 681         """
 682         Create a symbolic link named C{name} pointing at C{target}.
 683
 684         @type target: C{str}
 685         @type name: C{str}
 686         @raise SkipTest: raised if symbolic links are not supported on the
 687             host platform.
 688         """
 689         if getattr(os, 'symlink', None) is None:
 690             raise unittest.SkipTest(
 691                 "Platform does not support symbolic links.")
 692         os.symlink(target, name)
 693
 694
 695     def createLinks(self):
 696         """
 697         Create several symbolic links to files and directories.
 698         """
 699         subdir = self.path.child("sub1")
 700         self.symlink(subdir.path, self._mkpath("sub1.link"))
 701         self.symlink(subdir.child("file2").path, self._mkpath("file2.link"))
 702         self.symlink(subdir.child("file2").path,
 703                      self._mkpath("sub1", "sub1.file2.link"))
 704
 705
 706     def test_realpathSymlink(self):
 707         """
 708         L{FilePath.realpath} returns the path of the ultimate target of a
 709         symlink.
 710         """
 711         self.createLinks()
 712         self.symlink(self.path.child("file2.link").path,
 713                      self.path.child("link.link").path)
 714         self.assertEqual(self.path.child("link.link").realpath(),
 715                           self.path.child("sub1").child("file2"))
 716
 717
 718     def test_realpathCyclicalSymlink(self):
 719         """
 720         L{FilePath.realpath} raises L{filepath.LinkError} if the path is a
 721         symbolic link which is part of a cycle.
 722         """
 723         self.symlink(self.path.child("link1").path, self.path.child("link2").path)
 724         self.symlink(self.path.child("link2").path, self.path.child("link1").path)
 725         self.assertRaises(filepath.LinkError,
 726                           self.path.child("link2").realpath)
 727
 728
 729     def test_realpathNoSymlink(self):
 730         """
 731         L{FilePath.realpath} returns the path itself if the path is not a
 732         symbolic link.
 733         """
 734         self.assertEqual(self.path.child("sub1").realpath(),
 735                           self.path.child("sub1"))
 736
 737
 738     def test_walkCyclicalSymlink(self):
 739         """
 740         Verify that walking a path with a cyclical symlink raises an error
 741         """
 742         self.createLinks()
 743         self.symlink(self.path.child("sub1").path,
 744                      self.path.child("sub1").child("sub1.loopylink").path)
 745         def iterateOverPath():
 746             return [foo.path for foo in self.path.walk()]
 747         self.assertRaises(filepath.LinkError, iterateOverPath)
 748
 749
 750     def test_walkObeysDescendWithCyclicalSymlinks(self):
 751         """
 752         Verify that, after making a path with cyclical symlinks, when the
 753         supplied C{descend} predicate returns C{False}, the target is not
 754         traversed, as if it was a simple symlink.
 755         """
 756         self.createLinks()
 757         # we create cyclical symlinks
 758         self.symlink(self.path.child("sub1").path,
 759                      self.path.child("sub1").child("sub1.loopylink").path)
 760         def noSymLinks(path):
 761             return not path.islink()
 762         def iterateOverPath():
 763             return [foo.path for foo in self.path.walk(descend=noSymLinks)]
 764         self.assertTrue(iterateOverPath())
 765
 766
 767     def test_walkObeysDescend(self):
 768         """
 769         Verify that when the supplied C{descend} predicate returns C{False},
 770         the target is not traversed.
 771         """
 772         self.createLinks()
 773         def noSymLinks(path):
 774             return not path.islink()
 775         x = [foo.path for foo in self.path.walk(descend=noSymLinks)]
 776         self.assertEqual(set(x), set(self.all))
 777
 778
 779     def test_getAndSet(self):
 780         content = 'newcontent'
 781         self.path.child('new').setContent(content)
 782         newcontent = self.path.child('new').getContent()
 783         self.assertEqual(content, newcontent)
 784         content = 'content'
 785         self.path.child('new').setContent(content, '.tmp')
 786         newcontent = self.path.child('new').getContent()
 787         self.assertEqual(content, newcontent)
 788
 789
 790     def test_getContentFileClosing(self):
 791         """
 792         If reading from the underlying file raises an exception,
 793         L{FilePath.getContent} raises that exception after closing the file.
 794         """
 795         fp = ExplodingFilePath("")
 796         self.assertRaises(IOError, fp.getContent)
 797         self.assertTrue(fp.fp.closed)
 798
 799
 800     def test_setContentFileClosing(self):
 801         """
 802         If writing to the underlying file raises an exception,
 803         L{FilePath.setContent} raises that exception after closing the file.
 804         """
 805         fp = ExplodingFilePath("")
 806         self.assertRaises(IOError, fp.setContent, "blah")
 807         self.assertTrue(fp.fp.closed)
 808
 809
 810     def test_setContentNameCollision(self):
 811         """
 812         L{FilePath.setContent} will use a different temporary filename on each
 813         invocation, so that multiple processes, threads, or reentrant
 814         invocations will not collide with each other.
 815         """
 816         fp = TrackingFilePath(self.mktemp())
 817         fp.setContent("alpha")
 818         fp.setContent("beta")
 819
 820         # Sanity check: setContent should only open one derivative path each
 821         # time to store the temporary file.
 822         openedSiblings = fp.openedPaths()
 823         self.assertEqual(len(openedSiblings), 2)
 824         self.assertNotEquals(openedSiblings[0], openedSiblings[1])
 825
 826
 827     def test_setContentExtension(self):
 828         """
 829         L{FilePath.setContent} creates temporary files with a user-supplied
 830         extension, so that if it is somehow interrupted while writing them, the
 831         file that it leaves behind will be identifiable.
 832         """
 833         fp = TrackingFilePath(self.mktemp())
 834         fp.setContent("hello")
 835         opened = fp.openedPaths()
 836         self.assertEqual(len(opened), 1)
 837         self.assertTrue(opened[0].basename().endswith(".new"),
 838                         "%s does not end with default '.new' extension" % (
 839                             opened[0].basename()))
 840         fp.setContent("goodbye", "-something-else")
 841         opened = fp.openedPaths()
 842         self.assertEqual(len(opened), 2)
 843         self.assertTrue(opened[1].basename().endswith("-something-else"),
 844                         "%s does not end with -something-else extension" % (
 845                             opened[1].basename()))
 846
 847
 848     def test_symbolicLink(self):
 849         """
 850         Verify the behavior of the C{isLink} method against links and
 851         non-links. Also check that the symbolic link shares the directory
 852         property with its target.
 853         """
 854         s4 = self.path.child("sub4")
 855         s3 = self.path.child("sub3")
 856         self.symlink(s3.path, s4.path)
 857         self.assertTrue(s4.islink())
 858         self.assertFalse(s3.islink())
 859         self.assertTrue(s4.isdir())
 860         self.assertTrue(s3.isdir())
 861
 862
 863     def test_linkTo(self):
 864         """
 865         Verify that symlink creates a valid symlink that is both a link and a
 866         file if its target is a file, or a directory if its target is a
 867         directory.
 868         """
 869         targetLinks = [
 870             (self.path.child("sub2"), self.path.child("sub2.link")),
 871             (self.path.child("sub2").child("file3.ext1"),
 872              self.path.child("file3.ext1.link"))
 873             ]
 874         for target, link in targetLinks:
 875             target.linkTo(link)
 876             self.assertTrue(link.islink(), "This is a link")
 877             self.assertEqual(target.isdir(), link.isdir())
 878             self.assertEqual(target.isfile(), link.isfile())
 879
 880
 881     def test_linkToErrors(self):
 882         """
 883         Verify C{linkTo} fails in the following case:
 884             - the target is in a directory that doesn't exist
 885             - the target already exists
 886         """
 887         self.assertRaises(OSError, self.path.child("file1").linkTo,
 888                           self.path.child('nosub').child('file1'))
 889         self.assertRaises(OSError, self.path.child("file1").linkTo,
 890                           self.path.child('sub1').child('file2'))
 891
 892
 893     if not getattr(os, "symlink", None):
 894         skipMsg = "Your platform does not support symbolic links."
 895         test_symbolicLink.skip = skipMsg
 896         test_linkTo.skip = skipMsg
 897         test_linkToErrors.skip = skipMsg
 898
 899
 900     def testMultiExt(self):
 901         f3 = self.path.child('sub3').child('file3')
 902         exts = '.foo','.bar', 'ext1','ext2','ext3'
 903         self.failIf(f3.siblingExtensionSearch(*exts))
 904         f3e = f3.siblingExtension(".foo")
 905         f3e.touch()
 906         self.failIf(not f3.siblingExtensionSearch(*exts).exists())
 907         self.failIf(not f3.siblingExtensionSearch('*').exists())
 908         f3e.remove()
 909         self.failIf(f3.siblingExtensionSearch(*exts))
 910
 911     def testPreauthChild(self):
 912         fp = filepath.FilePath('.')
 913         fp.preauthChild('foo/bar')
 914         self.assertRaises(filepath.InsecurePath, fp.child, '/foo')
 915
 916     def testStatCache(self):
 917         p = self.path.child('stattest')
 918         p.touch()
 919         self.assertEqual(p.getsize(), 0)
 920         self.assertEqual(abs(p.getmtime() - time.time()) // 20, 0)
 921         self.assertEqual(abs(p.getctime() - time.time()) // 20, 0)
 922         self.assertEqual(abs(p.getatime() - time.time()) // 20, 0)
 923         self.assertEqual(p.exists(), True)
 924         self.assertEqual(p.exists(), True)
 925         # OOB removal: FilePath.remove() will automatically restat
 926         os.remove(p.path)
 927         # test caching
 928         self.assertEqual(p.exists(), True)
 929         p.restat(reraise=False)
 930         self.assertEqual(p.exists(), False)
 931         self.assertEqual(p.islink(), False)
 932         self.assertEqual(p.isdir(), False)
 933         self.assertEqual(p.isfile(), False)
 934
 935     def testPersist(self):
 936         newpath = pickle.loads(pickle.dumps(self.path))
 937         self.assertEqual(self.path.__class__, newpath.__class__)
 938         self.assertEqual(self.path.path, newpath.path)
 939
 940     def testInsecureUNIX(self):
 941         self.assertRaises(filepath.InsecurePath, self.path.child, "..")
 942         self.assertRaises(filepath.InsecurePath, self.path.child, "/etc")
 943         self.assertRaises(filepath.InsecurePath, self.path.child, "../..")
 944
 945     def testInsecureWin32(self):
 946         self.assertRaises(filepath.InsecurePath, self.path.child, r"..\..")
 947         self.assertRaises(filepath.InsecurePath, self.path.child, r"C:randomfile")
 948
 949     if platform.getType() != 'win32':
 950         testInsecureWin32.skip = "Test will run only on Windows."
 951
 952     def testInsecureWin32Whacky(self):
 953         """
 954         Windows has 'special' filenames like NUL and CON and COM1 and LPR
 955         and PRN and ... god knows what else.  They can be located anywhere in
 956         the filesystem.  For obvious reasons, we do not wish to normally permit
 957         access to these.
 958         """
 959         self.assertRaises(filepath.InsecurePath, self.path.child, "CON")
 960         self.assertRaises(filepath.InsecurePath, self.path.child, "C:CON")
 961         self.assertRaises(filepath.InsecurePath, self.path.child, r"C:\CON")
 962
 963     if platform.getType() != 'win32':
 964         testInsecureWin32Whacky.skip = "Test will run only on Windows."
 965
 966     def testComparison(self):
 967         self.assertEqual(filepath.FilePath('a'),
 968                           filepath.FilePath('a'))
 969         self.failUnless(filepath.FilePath('z') >
 970                         filepath.FilePath('a'))
 971         self.failUnless(filepath.FilePath('z') >=
 972                         filepath.FilePath('a'))
 973         self.failUnless(filepath.FilePath('a') >=
 974                         filepath.FilePath('a'))
 975         self.failUnless(filepath.FilePath('a') <=
 976                         filepath.FilePath('a'))
 977         self.failUnless(filepath.FilePath('a') <
 978                         filepath.FilePath('z'))
 979         self.failUnless(filepath.FilePath('a') <=
 980                         filepath.FilePath('z'))
 981         self.failUnless(filepath.FilePath('a') !=
 982                         filepath.FilePath('z'))
 983         self.failUnless(filepath.FilePath('z') !=
 984                         filepath.FilePath('a'))
 985
 986         self.failIf(filepath.FilePath('z') !=
 987                     filepath.FilePath('z'))
 988
 989
 990     def test_descendantOnly(self):
 991         """
 992         If C{".."} is in the sequence passed to L{FilePath.descendant},
 993         L{InsecurePath} is raised.
 994         """
 995         self.assertRaises(
 996             filepath.InsecurePath, self.path.descendant, ['a', '..'])
 997
 998
 999     def testSibling(self):
1000         p = self.path.child('sibling_start')
1001         ts = p.sibling('sibling_test')
1002         self.assertEqual(ts.dirname(), p.dirname())
1003         self.assertEqual(ts.basename(), 'sibling_test')
1004         ts.createDirectory()
1005         self.assertIn(ts, self.path.children())
1006
1007     def testTemporarySibling(self):
1008         ts = self.path.temporarySibling()
1009         self.assertEqual(ts.dirname(), self.path.dirname())
1010         self.assertNotIn(ts.basename(), self.path.listdir())
1011         ts.createDirectory()
1012         self.assertIn(ts, self.path.parent().children())
1013
1014
1015     def test_temporarySiblingExtension(self):
1016         """
1017         If L{FilePath.temporarySibling} is given an extension argument, it will
1018         produce path objects with that extension appended to their names.
1019         """
1020         testExtension = ".test-extension"
1021         ts = self.path.temporarySibling(testExtension)
1022         self.assertTrue(ts.basename().endswith(testExtension),
1023                         "%s does not end with %s" % (
1024                             ts.basename(), testExtension))
1025
1026
1027     def test_removeDirectory(self):
1028         """
1029         L{FilePath.remove} on a L{FilePath} that refers to a directory will
1030         recursively delete its contents.
1031         """
1032         self.path.remove()
1033         self.failIf(self.path.exists())
1034
1035
1036     def test_removeWithSymlink(self):
1037         """
1038         For a path which is a symbolic link, L{FilePath.remove} just deletes
1039         the link, not the target.
1040         """
1041         link = self.path.child("sub1.link")
1042         # setUp creates the sub1 child
1043         self.symlink(self.path.child("sub1").path, link.path)
1044         link.remove()
1045         self.assertFalse(link.exists())
1046         self.assertTrue(self.path.child("sub1").exists())
1047
1048
1049     def test_copyToDirectory(self):
1050         """
1051         L{FilePath.copyTo} makes a copy of all the contents of the directory
1052         named by that L{FilePath} if it is able to do so.
1053         """
1054         oldPaths = list(self.path.walk()) # Record initial state
1055         fp = filepath.FilePath(self.mktemp())
1056         self.path.copyTo(fp)
1057         self.path.remove()
1058         fp.copyTo(self.path)
1059         newPaths = list(self.path.walk()) # Record double-copy state
1060         newPaths.sort()
1061         oldPaths.sort()
1062         self.assertEqual(newPaths, oldPaths)
1063
1064
1065     def test_copyToMissingDestFileClosing(self):
1066         """
1067         If an exception is raised while L{FilePath.copyTo} is trying to open
1068         source file to read from, the destination file is closed and the
1069         exception is raised to the caller of L{FilePath.copyTo}.
1070         """
1071         nosuch = self.path.child("nothere")
1072         # Make it look like something to copy, even though it doesn't exist.
1073         # This could happen if the file is deleted between the isfile check and
1074         # the file actually being opened.
1075         nosuch.isfile = lambda: True
1076
1077         # We won't get as far as writing to this file, but it's still useful for
1078         # tracking whether we closed it.
1079         destination = ExplodingFilePath(self.mktemp())
1080
1081         self.assertRaises(IOError, nosuch.copyTo, destination)
1082         self.assertTrue(destination.fp.closed)
1083
1084
1085     def test_copyToFileClosing(self):
1086         """
1087         If an exception is raised while L{FilePath.copyTo} is copying bytes
1088         between two regular files, the source and destination files are closed
1089         and the exception propagates to the caller of L{FilePath.copyTo}.
1090         """
1091         destination = ExplodingFilePath(self.mktemp())
1092         source = ExplodingFilePath(__file__)
1093         self.assertRaises(IOError, source.copyTo, destination)
1094         self.assertTrue(source.fp.closed)
1095         self.assertTrue(destination.fp.closed)
1096
1097
1098     def test_copyToDirectoryItself(self):
1099         """
1100         L{FilePath.copyTo} fails with an OSError or IOError (depending on
1101         platform, as it propagates errors from open() and write()) when
1102         attempting to copy a directory to a child of itself.
1103         """
1104         self.assertRaises((OSError, IOError),
1105                           self.path.copyTo, self.path.child('file1'))
1106
1107
1108     def test_copyToWithSymlink(self):
1109         """
1110         Verify that copying with followLinks=True copies symlink targets
1111         instead of symlinks
1112         """
1113         self.symlink(self.path.child("sub1").path,
1114                      self.path.child("link1").path)
1115         fp = filepath.FilePath(self.mktemp())
1116         self.path.copyTo(fp)
1117         self.assertFalse(fp.child("link1").islink())
1118         self.assertEqual([x.basename() for x in fp.child("sub1").children()],
1119                           [x.basename() for x in fp.child("link1").children()])
1120
1121
1122     def test_copyToWithoutSymlink(self):
1123         """
1124         Verify that copying with followLinks=False copies symlinks as symlinks
1125         """
1126         self.symlink("sub1", self.path.child("link1").path)
1127         fp = filepath.FilePath(self.mktemp())
1128         self.path.copyTo(fp, followLinks=False)
1129         self.assertTrue(fp.child("link1").islink())
1130         self.assertEqual(os.readlink(self.path.child("link1").path),
1131                           os.readlink(fp.child("link1").path))
1132
1133
1134     def test_copyToMissingSource(self):
1135         """
1136         If the source path is missing, L{FilePath.copyTo} raises L{OSError}.
1137         """
1138         path = filepath.FilePath(self.mktemp())
1139         exc = self.assertRaises(OSError, path.copyTo, 'some other path')
1140         self.assertEqual(exc.errno, errno.ENOENT)
1141
1142
1143     def test_moveTo(self):
1144         """
1145         Verify that moving an entire directory results into another directory
1146         with the same content.
1147         """
1148         oldPaths = list(self.path.walk()) # Record initial state
1149         fp = filepath.FilePath(self.mktemp())
1150         self.path.moveTo(fp)
1151         fp.moveTo(self.path)
1152         newPaths = list(self.path.walk()) # Record double-move state
1153         newPaths.sort()
1154         oldPaths.sort()
1155         self.assertEqual(newPaths, oldPaths)
1156
1157
1158     def test_moveToExistsCache(self):
1159         """
1160         A L{FilePath} that has been moved aside with L{FilePath.moveTo} no
1161         longer registers as existing.  Its previously non-existent target
1162         exists, though, as it was created by the call to C{moveTo}.
1163         """
1164         fp = filepath.FilePath(self.mktemp())
1165         fp2 = filepath.FilePath(self.mktemp())
1166         fp.touch()
1167
1168         # Both a sanity check (make sure the file status looks right) and an
1169         # enticement for stat-caching logic to kick in and remember that these
1170         # exist / don't exist.
1171         self.assertEqual(fp.exists(), True)
1172         self.assertEqual(fp2.exists(), False)
1173
1174         fp.moveTo(fp2)
1175         self.assertEqual(fp.exists(), False)
1176         self.assertEqual(fp2.exists(), True)
1177
1178
1179     def test_moveToExistsCacheCrossMount(self):
1180         """
1181         The assertion of test_moveToExistsCache should hold in the case of a
1182         cross-mount move.
1183         """
1184         self.setUpFaultyRename()
1185         self.test_moveToExistsCache()
1186
1187
1188     def test_moveToSizeCache(self, hook=lambda : None):
1189         """
1190         L{FilePath.moveTo} clears its destination's status cache, such that
1191         calls to L{FilePath.getsize} after the call to C{moveTo} will report the
1192         new size, not the old one.
1193
1194         This is a separate test from C{test_moveToExistsCache} because it is
1195         intended to cover the fact that the destination's cache is dropped;
1196         test_moveToExistsCache doesn't cover this case because (currently) a
1197         file that doesn't exist yet does not cache the fact of its non-
1198         existence.
1199         """
1200         fp = filepath.FilePath(self.mktemp())
1201         fp2 = filepath.FilePath(self.mktemp())
1202         fp.setContent("1234")
1203         fp2.setContent("1234567890")
1204         hook()
1205
1206         # Sanity check / kick off caching.
1207         self.assertEqual(fp.getsize(), 4)
1208         self.assertEqual(fp2.getsize(), 10)
1209         # Actually attempting to replace a file on Windows would fail with
1210         # ERROR_ALREADY_EXISTS, but we don't need to test that, just the cached
1211         # metadata, so, delete the file ...
1212         os.remove(fp2.path)
1213         # ... but don't clear the status cache, as fp2.remove() would.
1214         self.assertEqual(fp2.getsize(), 10)
1215
1216         fp.moveTo(fp2)
1217         self.assertEqual(fp2.getsize(), 4)
1218
1219
1220     def test_moveToSizeCacheCrossMount(self):
1221         """
1222         The assertion of test_moveToSizeCache should hold in the case of a
1223         cross-mount move.
1224         """
1225         self.test_moveToSizeCache(hook=self.setUpFaultyRename)
1226
1227
1228     def test_moveToError(self):
1229         """
1230         Verify error behavior of moveTo: it should raises one of OSError or
1231         IOError if you want to move a path into one of its child. It's simply
1232         the error raised by the underlying rename system call.
1233         """
1234         self.assertRaises((OSError, IOError), self.path.moveTo, self.path.child('file1'))
1235
1236
1237     def setUpFaultyRename(self):
1238         """
1239         Set up a C{os.rename} that will fail with L{errno.EXDEV} on first call.
1240         This is used to simulate a cross-device rename failure.
1241
1242         @return: a list of pair (src, dest) of calls to C{os.rename}
1243         @rtype: C{list} of C{tuple}
1244         """
1245         invokedWith = []
1246         def faultyRename(src, dest):
1247             invokedWith.append((src, dest))
1248             if len(invokedWith) == 1:
1249                 raise OSError(errno.EXDEV, 'Test-induced failure simulating '
1250                                            'cross-device rename failure')
1251             return originalRename(src, dest)
1252
1253         originalRename = os.rename
1254         self.patch(os, "rename", faultyRename)
1255         return invokedWith
1256
1257
1258     def test_crossMountMoveTo(self):
1259         """
1260         C{moveTo} should be able to handle C{EXDEV} error raised by
1261         C{os.rename} when trying to move a file on a different mounted
1262         filesystem.
1263         """
1264         invokedWith = self.setUpFaultyRename()
1265         # Bit of a whitebox test - force os.rename, which moveTo tries
1266         # before falling back to a slower method, to fail, forcing moveTo to
1267         # use the slower behavior.
1268         self.test_moveTo()
1269         # A bit of a sanity check for this whitebox test - if our rename
1270         # was never invoked, the test has probably fallen into disrepair!
1271         self.assertTrue(invokedWith)
1272
1273
1274     def test_crossMountMoveToWithSymlink(self):
1275         """
1276         By default, when moving a symlink, it should follow the link and
1277         actually copy the content of the linked node.
1278         """
1279         invokedWith = self.setUpFaultyRename()
1280         f2 = self.path.child('file2')
1281         f3 = self.path.child('file3')
1282         self.symlink(self.path.child('file1').path, f2.path)
1283         f2.moveTo(f3)
1284         self.assertFalse(f3.islink())
1285         self.assertEqual(f3.getContent(), 'file 1')
1286         self.assertTrue(invokedWith)
1287
1288
1289     def test_crossMountMoveToWithoutSymlink(self):
1290         """
1291         Verify that moveTo called with followLinks=False actually create
1292         another symlink.
1293         """
1294         invokedWith = self.setUpFaultyRename()
1295         f2 = self.path.child('file2')
1296         f3 = self.path.child('file3')
1297         self.symlink(self.path.child('file1').path, f2.path)
1298         f2.moveTo(f3, followLinks=False)
1299         self.assertTrue(f3.islink())
1300         self.assertEqual(f3.getContent(), 'file 1')
1301         self.assertTrue(invokedWith)
1302
1303
1304     def test_createBinaryMode(self):
1305         """
1306         L{FilePath.create} should always open (and write to) files in binary
1307         mode; line-feed octets should be unmodified.
1308
1309         (While this test should pass on all platforms, it is only really
1310         interesting on platforms which have the concept of binary mode, i.e.
1311         Windows platforms.)
1312         """
1313         path = filepath.FilePath(self.mktemp())
1314         f = path.create()
1315         self.failUnless("b" in f.mode)
1316         f.write("\n")
1317         f.close()
1318         read = open(path.path, "rb").read()
1319         self.assertEqual(read, "\n")
1320
1321
1322     def testOpen(self):
1323         # Opening a file for reading when it does not already exist is an error
1324         nonexistent = self.path.child('nonexistent')
1325         e = self.assertRaises(IOError, nonexistent.open)
1326         self.assertEqual(e.errno, errno.ENOENT)
1327
1328         # Opening a file for writing when it does not exist is okay
1329         writer = self.path.child('writer')
1330         f = writer.open('w')
1331         f.write('abc\ndef')
1332         f.close()
1333
1334         # Make sure those bytes ended up there - and test opening a file for
1335         # reading when it does exist at the same time
1336         f = writer.open()
1337         self.assertEqual(f.read(), 'abc\ndef')
1338         f.close()
1339
1340         # Re-opening that file in write mode should erase whatever was there.
1341         f = writer.open('w')
1342         f.close()
1343         f = writer.open()
1344         self.assertEqual(f.read(), '')
1345         f.close()
1346
1347         # Put some bytes in a file so we can test that appending does not
1348         # destroy them.
1349         appender = self.path.child('appender')
1350         f = appender.open('w')
1351         f.write('abc')
1352         f.close()
1353
1354         f = appender.open('a')
1355         f.write('def')
1356         f.close()
1357
1358         f = appender.open('r')
1359         self.assertEqual(f.read(), 'abcdef')
1360         f.close()
1361
1362         # read/write should let us do both without erasing those bytes
1363         f = appender.open('r+')
1364         self.assertEqual(f.read(), 'abcdef')
1365         # ANSI C *requires* an fseek or an fgetpos between an fread and an
1366         # fwrite or an fwrite and a fread.  We can't reliable get Python to
1367         # invoke fgetpos, so we seek to a 0 byte offset from the current
1368         # position instead.  Also, Python sucks for making this seek
1369         # relative to 1 instead of a symbolic constant representing the
1370         # current file position.
1371         f.seek(0, 1)
1372         # Put in some new bytes for us to test for later.
1373         f.write('ghi')
1374         f.close()
1375
1376         # Make sure those new bytes really showed up
1377         f = appender.open('r')
1378         self.assertEqual(f.read(), 'abcdefghi')
1379         f.close()
1380
1381         # write/read should let us do both, but erase anything that's there
1382         # already.
1383         f = appender.open('w+')
1384         self.assertEqual(f.read(), '')
1385         f.seek(0, 1) # Don't forget this!
1386         f.write('123')
1387         f.close()
1388
1389         # super append mode should let us read and write and also position the
1390         # cursor at the end of the file, without erasing everything.
1391         f = appender.open('a+')
1392
1393         # The order of these lines may seem surprising, but it is necessary.
1394         # The cursor is not at the end of the file until after the first write.
1395         f.write('456')
1396         f.seek(0, 1) # Asinine.
1397         self.assertEqual(f.read(), '')
1398
1399         f.seek(0, 0)
1400         self.assertEqual(f.read(), '123456')
1401         f.close()
1402
1403         # Opening a file exclusively must fail if that file exists already.
1404         nonexistent.requireCreate(True)
1405         nonexistent.open('w').close()
1406         existent = nonexistent
1407         del nonexistent
1408         self.assertRaises((OSError, IOError), existent.open)
1409
1410
1411     def test_openWithExplicitBinaryMode(self):
1412         """
1413         Due to a bug in Python 2.7 on Windows including multiple 'b'
1414         characters in the mode passed to the built-in open() will cause an
1415         error.  FilePath.open() ensures that only a single 'b' character is
1416         included in the mode passed to the built-in open().
1417
1418         See http://bugs.python.org/issue7686 for details about the bug.
1419         """
1420         writer = self.path.child('explicit-binary')
1421         file = writer.open('wb')
1422         file.write('abc\ndef')
1423         file.close()
1424         self.assertTrue(writer.exists)
1425
1426
1427     def test_openWithRedundantExplicitBinaryModes(self):
1428         """
1429         Due to a bug in Python 2.7 on Windows including multiple 'b'
1430         characters in the mode passed to the built-in open() will cause an
1431         error.  No matter how many 'b' modes are specified, FilePath.open()
1432         ensures that only a single 'b' character is included in the mode
1433         passed to the built-in open().
1434
1435         See http://bugs.python.org/issue7686 for details about the bug.
1436         """
1437         writer = self.path.child('multiple-binary')
1438         file = writer.open('wbb')
1439         file.write('abc\ndef')
1440         file.close()
1441         self.assertTrue(writer.exists)
1442
1443
1444     def test_existsCache(self):
1445         """
1446         Check that C{filepath.FilePath.exists} correctly restat the object if
1447         an operation has occurred in the mean time.
1448         """
1449         fp = filepath.FilePath(self.mktemp())
1450         self.assertEqual(fp.exists(), False)
1451
1452         fp.makedirs()
1453         self.assertEqual(fp.exists(), True)
1454
1455
1456     def test_changed(self):
1457         """
1458         L{FilePath.changed} indicates that the L{FilePath} has changed, but does
1459         not re-read the status information from the filesystem until it is
1460         queried again via another method, such as C{getsize}.
1461         """
1462         fp = filepath.FilePath(self.mktemp())
1463         fp.setContent("12345")
1464         self.assertEqual(fp.getsize(), 5)
1465
1466         # Someone else comes along and changes the file.
1467         fObj = open(fp.path, 'wb')
1468         fObj.write("12345678")
1469         fObj.close()
1470
1471         # Sanity check for caching: size should still be 5.
1472         self.assertEqual(fp.getsize(), 5)
1473         fp.changed()
1474
1475         # This path should look like we don't know what status it's in, not that
1476         # we know that it didn't exist when last we checked.
1477         self.assertEqual(fp.statinfo, None)
1478         self.assertEqual(fp.getsize(), 8)
1479
1480
1481     def test_getPermissions_POSIX(self):
1482         """
1483         Getting permissions for a file returns a L{Permissions} object for
1484         POSIX platforms (which supports separate user, group, and other
1485         permissions bits.
1486         """
1487         for mode in (0777, 0700):
1488             self.path.child("sub1").chmod(mode)
1489             self.assertEqual(self.path.child("sub1").getPermissions(),
1490                               filepath.Permissions(mode))
1491         self.path.child("sub1").chmod(0764) #sanity check
1492         self.assertEqual(self.path.child("sub1").getPermissions().shorthand(),
1493                           "rwxrw-r--")
1494
1495
1496     def test_getPermissions_Windows(self):
1497         """
1498         Getting permissions for a file returns a L{Permissions} object in
1499         Windows.  Windows requires a different test, because user permissions
1500         = group permissions = other permissions.  Also, chmod may not be able
1501         to set the execute bit, so we are skipping tests that set the execute
1502         bit.
1503         """
1504         for mode in (0777, 0555):
1505             self.path.child("sub1").chmod(mode)
1506             self.assertEqual(self.path.child("sub1").getPermissions(),
1507                               filepath.Permissions(mode))
1508         self.path.child("sub1").chmod(0511) #sanity check to make sure that
1509         # user=group=other permissions
1510         self.assertEqual(self.path.child("sub1").getPermissions().shorthand(),
1511                           "r-xr-xr-x")
1512
1513
1514     def test_whetherBlockOrSocket(self):
1515         """
1516         Ensure that a file is not a block or socket
1517         """
1518         self.assertFalse(self.path.isBlockDevice())
1519         self.assertFalse(self.path.isSocket())
1520
1521
1522     def test_statinfoBitsNotImplementedInWindows(self):
1523         """
1524         Verify that certain file stats are not available on Windows
1525         """
1526         self.assertRaises(NotImplementedError, self.path.getInodeNumber)
1527         self.assertRaises(NotImplementedError, self.path.getDevice)
1528         self.assertRaises(NotImplementedError, self.path.getNumberOfHardLinks)
1529         self.assertRaises(NotImplementedError, self.path.getUserID)
1530         self.assertRaises(NotImplementedError, self.path.getGroupID)
1531
1532
1533     def test_statinfoBitsAreNumbers(self):
1534         """
1535         Verify that file inode/device/nlinks/uid/gid stats are numbers in
1536         a POSIX environment
1537         """
1538         c = self.path.child('file1')
1539         for p in self.path, c:
1540             self.assertIsInstance(p.getInodeNumber(), long)
1541             self.assertIsInstance(p.getDevice(), long)
1542             self.assertIsInstance(p.getNumberOfHardLinks(), int)
1543             self.assertIsInstance(p.getUserID(), int)
1544             self.assertIsInstance(p.getGroupID(), int)
1545         self.assertEqual(self.path.getUserID(), c.getUserID())
1546         self.assertEqual(self.path.getGroupID(), c.getGroupID())
1547
1548
1549     def test_statinfoNumbersAreValid(self):
1550         """
1551         Verify that the right numbers come back from the right accessor methods
1552         for file inode/device/nlinks/uid/gid (in a POSIX environment)
1553         """
1554         # specify fake statinfo information
1555         class FakeStat:
1556             st_ino = 200
1557             st_dev = 300
1558             st_nlink = 400
1559             st_uid = 500
1560             st_gid = 600
1561
1562         # monkey patch in a fake restat method for self.path
1563         fake = FakeStat()
1564         def fakeRestat(*args, **kwargs):
1565             self.path.statinfo = fake
1566         self.path.restat = fakeRestat
1567
1568         # ensure that restat will need to be called to get values
1569         self.path.statinfo = None
1570
1571         self.assertEqual(self.path.getInodeNumber(), fake.st_ino)
1572         self.assertEqual(self.path.getDevice(), fake.st_dev)
1573         self.assertEqual(self.path.getNumberOfHardLinks(), fake.st_nlink)
1574         self.assertEqual(self.path.getUserID(), fake.st_uid)
1575         self.assertEqual(self.path.getGroupID(), fake.st_gid)
1576
1577
1578     if platform.isWindows():
1579         test_statinfoBitsAreNumbers.skip = True
1580         test_statinfoNumbersAreValid.skip = True
1581         test_getPermissions_POSIX.skip = True
1582     else:
1583         test_statinfoBitsNotImplementedInWindows.skip = "Test will run only on Windows."
1584         test_getPermissions_Windows.skip = "Test will run only on Windows."
1585
1586
1587
1588 from twisted.python import urlpath
1589
1590 class URLPathTestCase(unittest.TestCase):
1591     def setUp(self):
1592         self.path = urlpath.URLPath.fromString("http://example.com/foo/bar?yes=no&no=yes#footer")
1593
1594     def testStringConversion(self):
1595         self.assertEqual(str(self.path), "http://example.com/foo/bar?yes=no&no=yes#footer")
1596
1597     def testChildString(self):
1598         self.assertEqual(str(self.path.child('hello')), "http://example.com/foo/bar/hello")
1599         self.assertEqual(str(self.path.child('hello').child('')), "http://example.com/foo/bar/hello/")
1600
1601     def testSiblingString(self):
1602         self.assertEqual(str(self.path.sibling('baz')), 'http://example.com/foo/baz')
1603
1604         # The sibling of http://example.com/foo/bar/
1605         #     is http://example.comf/foo/bar/baz
1606         # because really we are constructing a sibling of
1607         # http://example.com/foo/bar/index.html
1608         self.assertEqual(str(self.path.child('').sibling('baz')), 'http://example.com/foo/bar/baz')
1609
1610     def testParentString(self):
1611         # parent should be equivalent to '..'
1612         # 'foo' is the current directory, '/' is the parent directory
1613         self.assertEqual(str(self.path.parent()), 'http://example.com/')
1614         self.assertEqual(str(self.path.child('').parent()), 'http://example.com/foo/')
1615         self.assertEqual(str(self.path.child('baz').parent()), 'http://example.com/foo/')
1616         self.assertEqual(str(self.path.parent().parent().parent().parent().parent()), 'http://example.com/')
1617
1618     def testHereString(self):
1619         # here should be equivalent to '.'
1620         self.assertEqual(str(self.path.here()), 'http://example.com/foo/')
1621         self.assertEqual(str(self.path.child('').here()), 'http://example.com/foo/bar/')
1622