1 # Copyright (c) Twisted Matrix Laboratories.
2 # See LICENSE for details.
5 Test cases covering L{twisted.python.filepath} and L{twisted.python.zippath}.
8 import os, time, pickle, errno, zipfile, stat
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
16 from twisted.trial import unittest
18 from zope.interface.verify import verifyObject
21 class AbstractFilePathTestCase(unittest.TestCase):
27 def _mkpath(self, *p):
28 x = os.path.abspath(os.path.join(self.cmn, *p))
33 def subdir(self, *dirname):
34 os.mkdir(self._mkpath(*dirname))
37 def subfile(self, *dirname):
38 return open(self._mkpath(*dirname), "wb")
42 self.now = time.time()
43 cmn = self.cmn = os.path.abspath(self.mktemp())
47 f = self.subfile("file1")
48 f.write(self.f1content)
50 f = self.subfile("sub1", "file2")
51 f.write(self.f2content)
54 f = self.subfile("sub3", "file3.ext1")
56 f = self.subfile("sub3", "file3.ext2")
58 f = self.subfile("sub3", "file3.ext3")
60 self.path = filepath.FilePath(cmn)
61 self.root = filepath.FilePath("/")
64 def test_segmentsFromPositive(self):
66 Verify that the segments between two paths are correctly identified.
69 self.path.child("a").child("b").child("c").segmentsFrom(self.path),
72 def test_segmentsFromNegative(self):
74 Verify that segmentsFrom notices when the ancestor isn't an ancestor.
78 self.path.child("a").child("b").child("c").segmentsFrom,
79 self.path.child("d").child("c").child("e"))
84 Verify that walking the path gives the same result as the known file
87 x = [foo.path for foo in self.path.walk()]
88 self.assertEqual(set(x), set(self.all))
91 def test_parents(self):
93 L{FilePath.parents()} should return an iterator of every ancestor of
94 the L{FilePath} in question.
97 pathobj = self.path.child("a").child("b").child("c")
98 fullpath = pathobj.path
100 thispath = os.path.dirname(fullpath)
101 while lastpath != self.root.path:
104 thispath = os.path.dirname(thispath)
105 self.assertEqual([x.path for x in pathobj.parents()], L)
108 def test_validSubdir(self):
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.
113 sub1 = self.path.child('sub1')
114 self.failUnless(sub1.exists(),
115 "This directory does exist.")
116 self.failUnless(sub1.isdir(),
118 self.failUnless(not sub1.isfile(),
120 self.failUnless(not sub1.islink(),
122 self.assertEqual(sub1.listdir(),
126 def test_invalidSubdir(self):
128 Verify that a subdirectory that doesn't exist is reported as such.
130 sub2 = self.path.child('sub2')
131 self.failIf(sub2.exists(),
132 "This directory does not exist.")
134 def test_validFiles(self):
136 Make sure that we can read existent non-empty files.
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)
144 def test_multipleChildSegments(self):
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)}.
149 multiple = self.path.descendant(['a', 'b', 'c'])
150 single = self.path.child('a').child('b').child('c')
151 self.assertEqual(multiple, single)
154 def test_dictionaryKeys(self):
156 Verify that path instances are usable as dictionary keys.
158 f1 = self.path.child('file1')
159 f1prime = self.path.child('file1')
160 f2 = self.path.child('file2')
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
169 self.assertEqual(dictoid[f2], 5)
170 self.assertEqual(len(dictoid), 2)
173 def test_dictionaryKeyWithString(self):
175 Verify that path instances are usable as dictionary keys which do not clash
176 with their string counterparts.
178 f1 = self.path.child('file1')
179 dictoid = {f1: 'hello'}
180 dictoid[f1.path] = 'goodbye'
181 self.assertEqual(len(dictoid), 2)
184 def test_childrenNonexistentError(self):
186 Verify that children raises the appropriate exception for non-existent
189 self.assertRaises(filepath.UnlistableError,
190 self.path.child('not real').children)
192 def test_childrenNotDirectoryError(self):
194 Verify that listdir raises the appropriate exception for attempting to list
195 a file rather than a directory.
197 self.assertRaises(filepath.UnlistableError,
198 self.path.child('file1').children)
201 def test_newTimesAreFloats(self):
203 Verify that all times returned from the various new time functions are ints
204 (and hopefully therefore 'high precision').
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)
212 def test_oldTimesAreInts(self):
214 Verify that all times returned from the various time functions are
215 integers, for compatibility.
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)
224 class FakeWindowsPath(filepath.FilePath):
226 A test version of FilePath which overrides listdir to raise L{WindowsError}.
231 @raise WindowsError: always.
235 "A directory's validness was called into question")
238 class ListingCompatibilityTests(unittest.TestCase):
240 These tests verify compatibility with legacy behavior of directory listing.
243 def test_windowsErrorExcept(self):
245 Verify that when a WindowsError is raised from listdir, catching
248 fwp = FakeWindowsPath(self.mktemp())
249 self.assertRaises(filepath.UnlistableError, fwp.children)
250 self.assertRaises(WindowsError, fwp.children)
253 def test_alwaysCatchOSError(self):
255 Verify that in the normal case where a directory does not exist, we will
258 fp = filepath.FilePath(self.mktemp())
259 self.assertRaises(OSError, fp.children)
262 def test_keepOriginalAttributes(self):
264 Verify that the Unlistable exception raised will preserve the attributes of
265 the previously-raised exception.
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()
274 self.assertEqual(d1, d2)
278 def zipit(dirname, zfname):
280 Create a zipfile on zfname, containing the contents of dirname'
282 zf = zipfile.ZipFile(zfname, "w")
283 for root, ignored, files, in os.walk(dirname):
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)
293 class ZipFilePathTestCase(AbstractFilePathTestCase):
295 Test various L{ZipPath} path manipulations as well as reprs for L{ZipPath}
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]
306 def test_verifyObject(self):
308 ZipPaths implement IFilePath.
311 self.assertTrue(verifyObject(filepath.IFilePath, self.path))
314 def test_zipPathRepr(self):
316 Make sure that invoking ZipPath's repr prints the correct class name
317 and an absolute path to the zip file.
319 child = self.path.child("foo")
320 pathRepr = "ZipPath(%r)" % (
321 os.path.abspath(self.cmn + ".zip" + os.sep + 'foo'),)
323 # Check for an absolute path
324 self.assertEqual(repr(child), pathRepr)
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")
331 # Check using a path without the cwd prepended
332 self.assertEqual(repr(child), pathRepr)
335 def test_zipPathReprParentDirSegment(self):
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
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)
347 def test_zipPathReprEscaping(self):
349 Bytes in the ZipPath path which have special meaning in Python
350 string literals are escaped in the ZipPath repr.
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)
358 def test_zipArchiveRepr(self):
360 Make sure that invoking ZipArchive's repr prints the correct class
361 name and an absolute path to the zip file.
363 pathRepr = 'ZipArchive(%r)' % (os.path.abspath(self.cmn + '.zip'),)
365 # Check for an absolute path
366 self.assertEqual(repr(self.path), pathRepr)
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)
372 # Check using a path without the cwd prepended
373 self.assertEqual(repr(relpath), pathRepr)
379 A C{file}-alike which raises exceptions from its I/O methods and keeps track
380 of whether it has been closed.
382 @ivar closed: A C{bool} which is C{False} until C{close} is called, then it
389 @raise IOError: Always raised.
394 def write(self, what):
396 @raise IOError: Always raised.
403 Mark the file as having been closed.
409 class TrackingFilePath(filepath.FilePath):
411 A subclass of L{filepath.FilePath} which maintains a list of all other paths
412 created by clonePath.
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).
419 @type trackingList: C{list} of L{TrackingFilePath}
421 @ivar openedFiles: A list of all file objects opened by this
422 L{TrackingFilePath} or any other L{TrackingFilePath} in C{trackingList}.
424 @type openedFiles: C{list} of C{file}
427 def __init__(self, path, alwaysCreate=False, trackingList=None):
428 filepath.FilePath.__init__(self, path, alwaysCreate)
429 if trackingList is None:
431 self.trackingList = trackingList
432 self.openedFiles = []
435 def open(self, *a, **k):
437 Override 'open' to track all files opened by this path.
439 f = filepath.FilePath.open(self, *a, **k)
440 self.openedFiles.append(f)
444 def openedPaths(self):
446 Return a list of all L{TrackingFilePath}s associated with this
447 L{TrackingFilePath} that have had their C{open()} method called.
449 return [path for path in self.trackingList if path.openedFiles]
452 def clonePath(self, name):
454 Override L{filepath.FilePath.clonePath} to give the new path a reference
455 to the same tracking list.
457 clone = TrackingFilePath(name, trackingList=self.trackingList)
458 self.trackingList.append(clone)
463 class ExplodingFilePath(filepath.FilePath):
465 A specialized L{FilePath} which always returns an instance of
466 L{ExplodingFile} from its C{open} method.
468 @ivar fp: The L{ExplodingFile} instance most recently returned from the
472 def __init__(self, pathName, originalExploder=None):
474 Initialize an L{ExplodingFilePath} with a name and a reference to the
476 @param pathName: The path name as passed to L{filepath.FilePath}.
477 @type pathName: C{str}
479 @param originalExploder: The L{ExplodingFilePath} to associate opened
481 @type originalExploder: L{ExplodingFilePath}
483 filepath.FilePath.__init__(self, pathName)
484 if originalExploder is None:
485 originalExploder = self
486 self._originalExploder = originalExploder
489 def open(self, mode=None):
491 Create, save, and return a new C{ExplodingFile}.
493 @param mode: Present for signature compatibility. Ignored.
495 @return: A new C{ExplodingFile}.
497 f = self._originalExploder.fp = ExplodingFile()
501 def clonePath(self, name):
502 return ExplodingFilePath(name, self._originalExploder)
506 class PermissionsTestCase(unittest.TestCase):
508 Test Permissions and RWX classes
511 def assertNotUnequal(self, first, second, msg=None):
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.
517 Note: this should really be part of trial
524 raise self.failureException(
525 '%snot not unequal (__ne__ not implemented correctly):'
527 % (msg, pformat(first), pformat(second)))
531 def test_rwxFromBools(self):
533 L{RWX}'s constructor takes a set of booleans
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)
546 def test_rwxEqNe(self):
548 L{RWX}'s created with the same booleans are equivalent. If booleans
549 are different, they are not equal.
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))
563 def test_rwxShorthand(self):
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 '-'.
570 def getChar(val, letter):
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(),
583 self.assertEqual(filepath.RWX(True, False, True).shorthand(), "r-x")
586 def test_permissionsFromStat(self):
588 L{Permissions}'s constructor takes a valid permissions bitmask and
589 parsaes it to produce the correct set of boolean permissions.
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
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"),
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))
622 def test_permissionsEq(self):
624 Two L{Permissions}'s that are created with the same bitmask
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))
636 def test_permissionsShorthand(self):
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.
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---")
653 class FilePathTestCase(AbstractFilePathTestCase):
655 Test various L{FilePath} path manipulations.
659 def test_verifyObject(self):
661 FilePaths implement IFilePath.
664 self.assertTrue(verifyObject(filepath.IFilePath, self.path))
667 def test_chmod(self):
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).
673 for mode in (0555, 0777):
674 self.path.child("sub1").chmod(mode)
676 stat.S_IMODE(os.stat(self.path.child("sub1").path).st_mode),
680 def symlink(self, target, name):
682 Create a symbolic link named C{name} pointing at C{target}.
686 @raise SkipTest: raised if symbolic links are not supported on the
689 if getattr(os, 'symlink', None) is None:
690 raise unittest.SkipTest(
691 "Platform does not support symbolic links.")
692 os.symlink(target, name)
695 def createLinks(self):
697 Create several symbolic links to files and directories.
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"))
706 def test_realpathSymlink(self):
708 L{FilePath.realpath} returns the path of the ultimate target of a
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"))
718 def test_realpathCyclicalSymlink(self):
720 L{FilePath.realpath} raises L{filepath.LinkError} if the path is a
721 symbolic link which is part of a cycle.
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)
729 def test_realpathNoSymlink(self):
731 L{FilePath.realpath} returns the path itself if the path is not a
734 self.assertEqual(self.path.child("sub1").realpath(),
735 self.path.child("sub1"))
738 def test_walkCyclicalSymlink(self):
740 Verify that walking a path with a cyclical symlink raises an error
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)
750 def test_walkObeysDescendWithCyclicalSymlinks(self):
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.
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())
767 def test_walkObeysDescend(self):
769 Verify that when the supplied C{descend} predicate returns C{False},
770 the target is not traversed.
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))
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)
785 self.path.child('new').setContent(content, '.tmp')
786 newcontent = self.path.child('new').getContent()
787 self.assertEqual(content, newcontent)
790 def test_getContentFileClosing(self):
792 If reading from the underlying file raises an exception,
793 L{FilePath.getContent} raises that exception after closing the file.
795 fp = ExplodingFilePath("")
796 self.assertRaises(IOError, fp.getContent)
797 self.assertTrue(fp.fp.closed)
800 def test_setContentFileClosing(self):
802 If writing to the underlying file raises an exception,
803 L{FilePath.setContent} raises that exception after closing the file.
805 fp = ExplodingFilePath("")
806 self.assertRaises(IOError, fp.setContent, "blah")
807 self.assertTrue(fp.fp.closed)
810 def test_setContentNameCollision(self):
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.
816 fp = TrackingFilePath(self.mktemp())
817 fp.setContent("alpha")
818 fp.setContent("beta")
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])
827 def test_setContentExtension(self):
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.
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()))
848 def test_symbolicLink(self):
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.
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())
863 def test_linkTo(self):
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
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"))
874 for target, link in targetLinks:
876 self.assertTrue(link.islink(), "This is a link")
877 self.assertEqual(target.isdir(), link.isdir())
878 self.assertEqual(target.isfile(), link.isfile())
881 def test_linkToErrors(self):
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
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'))
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
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")
906 self.failIf(not f3.siblingExtensionSearch(*exts).exists())
907 self.failIf(not f3.siblingExtensionSearch('*').exists())
909 self.failIf(f3.siblingExtensionSearch(*exts))
911 def testPreauthChild(self):
912 fp = filepath.FilePath('.')
913 fp.preauthChild('foo/bar')
914 self.assertRaises(filepath.InsecurePath, fp.child, '/foo')
916 def testStatCache(self):
917 p = self.path.child('stattest')
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
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)
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)
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, "../..")
945 def testInsecureWin32(self):
946 self.assertRaises(filepath.InsecurePath, self.path.child, r"..\..")
947 self.assertRaises(filepath.InsecurePath, self.path.child, r"C:randomfile")
949 if platform.getType() != 'win32':
950 testInsecureWin32.skip = "Test will run only on Windows."
952 def testInsecureWin32Whacky(self):
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
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")
963 if platform.getType() != 'win32':
964 testInsecureWin32Whacky.skip = "Test will run only on Windows."
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'))
986 self.failIf(filepath.FilePath('z') !=
987 filepath.FilePath('z'))
990 def test_descendantOnly(self):
992 If C{".."} is in the sequence passed to L{FilePath.descendant},
993 L{InsecurePath} is raised.
996 filepath.InsecurePath, self.path.descendant, ['a', '..'])
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())
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())
1015 def test_temporarySiblingExtension(self):
1017 If L{FilePath.temporarySibling} is given an extension argument, it will
1018 produce path objects with that extension appended to their names.
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))
1027 def test_removeDirectory(self):
1029 L{FilePath.remove} on a L{FilePath} that refers to a directory will
1030 recursively delete its contents.
1033 self.failIf(self.path.exists())
1036 def test_removeWithSymlink(self):
1038 For a path which is a symbolic link, L{FilePath.remove} just deletes
1039 the link, not the target.
1041 link = self.path.child("sub1.link")
1042 # setUp creates the sub1 child
1043 self.symlink(self.path.child("sub1").path, link.path)
1045 self.assertFalse(link.exists())
1046 self.assertTrue(self.path.child("sub1").exists())
1049 def test_copyToDirectory(self):
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.
1054 oldPaths = list(self.path.walk()) # Record initial state
1055 fp = filepath.FilePath(self.mktemp())
1056 self.path.copyTo(fp)
1058 fp.copyTo(self.path)
1059 newPaths = list(self.path.walk()) # Record double-copy state
1062 self.assertEqual(newPaths, oldPaths)
1065 def test_copyToMissingDestFileClosing(self):
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}.
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
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())
1081 self.assertRaises(IOError, nosuch.copyTo, destination)
1082 self.assertTrue(destination.fp.closed)
1085 def test_copyToFileClosing(self):
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}.
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)
1098 def test_copyToDirectoryItself(self):
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.
1104 self.assertRaises((OSError, IOError),
1105 self.path.copyTo, self.path.child('file1'))
1108 def test_copyToWithSymlink(self):
1110 Verify that copying with followLinks=True copies symlink targets
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()])
1122 def test_copyToWithoutSymlink(self):
1124 Verify that copying with followLinks=False copies symlinks as symlinks
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))
1134 def test_copyToMissingSource(self):
1136 If the source path is missing, L{FilePath.copyTo} raises L{OSError}.
1138 path = filepath.FilePath(self.mktemp())
1139 exc = self.assertRaises(OSError, path.copyTo, 'some other path')
1140 self.assertEqual(exc.errno, errno.ENOENT)
1143 def test_moveTo(self):
1145 Verify that moving an entire directory results into another directory
1146 with the same content.
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
1155 self.assertEqual(newPaths, oldPaths)
1158 def test_moveToExistsCache(self):
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}.
1164 fp = filepath.FilePath(self.mktemp())
1165 fp2 = filepath.FilePath(self.mktemp())
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)
1175 self.assertEqual(fp.exists(), False)
1176 self.assertEqual(fp2.exists(), True)
1179 def test_moveToExistsCacheCrossMount(self):
1181 The assertion of test_moveToExistsCache should hold in the case of a
1184 self.setUpFaultyRename()
1185 self.test_moveToExistsCache()
1188 def test_moveToSizeCache(self, hook=lambda : None):
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.
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-
1200 fp = filepath.FilePath(self.mktemp())
1201 fp2 = filepath.FilePath(self.mktemp())
1202 fp.setContent("1234")
1203 fp2.setContent("1234567890")
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 ...
1213 # ... but don't clear the status cache, as fp2.remove() would.
1214 self.assertEqual(fp2.getsize(), 10)
1217 self.assertEqual(fp2.getsize(), 4)
1220 def test_moveToSizeCacheCrossMount(self):
1222 The assertion of test_moveToSizeCache should hold in the case of a
1225 self.test_moveToSizeCache(hook=self.setUpFaultyRename)
1228 def test_moveToError(self):
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.
1234 self.assertRaises((OSError, IOError), self.path.moveTo, self.path.child('file1'))
1237 def setUpFaultyRename(self):
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.
1242 @return: a list of pair (src, dest) of calls to C{os.rename}
1243 @rtype: C{list} of C{tuple}
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)
1253 originalRename = os.rename
1254 self.patch(os, "rename", faultyRename)
1258 def test_crossMountMoveTo(self):
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
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.
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)
1274 def test_crossMountMoveToWithSymlink(self):
1276 By default, when moving a symlink, it should follow the link and
1277 actually copy the content of the linked node.
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)
1284 self.assertFalse(f3.islink())
1285 self.assertEqual(f3.getContent(), 'file 1')
1286 self.assertTrue(invokedWith)
1289 def test_crossMountMoveToWithoutSymlink(self):
1291 Verify that moveTo called with followLinks=False actually create
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)
1304 def test_createBinaryMode(self):
1306 L{FilePath.create} should always open (and write to) files in binary
1307 mode; line-feed octets should be unmodified.
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.
1313 path = filepath.FilePath(self.mktemp())
1315 self.failUnless("b" in f.mode)
1318 read = open(path.path, "rb").read()
1319 self.assertEqual(read, "\n")
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)
1328 # Opening a file for writing when it does not exist is okay
1329 writer = self.path.child('writer')
1330 f = writer.open('w')
1334 # Make sure those bytes ended up there - and test opening a file for
1335 # reading when it does exist at the same time
1337 self.assertEqual(f.read(), 'abc\ndef')
1340 # Re-opening that file in write mode should erase whatever was there.
1341 f = writer.open('w')
1344 self.assertEqual(f.read(), '')
1347 # Put some bytes in a file so we can test that appending does not
1349 appender = self.path.child('appender')
1350 f = appender.open('w')
1354 f = appender.open('a')
1358 f = appender.open('r')
1359 self.assertEqual(f.read(), 'abcdef')
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.
1372 # Put in some new bytes for us to test for later.
1376 # Make sure those new bytes really showed up
1377 f = appender.open('r')
1378 self.assertEqual(f.read(), 'abcdefghi')
1381 # write/read should let us do both, but erase anything that's there
1383 f = appender.open('w+')
1384 self.assertEqual(f.read(), '')
1385 f.seek(0, 1) # Don't forget this!
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+')
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.
1396 f.seek(0, 1) # Asinine.
1397 self.assertEqual(f.read(), '')
1400 self.assertEqual(f.read(), '123456')
1403 # Opening a file exclusively must fail if that file exists already.
1404 nonexistent.requireCreate(True)
1405 nonexistent.open('w').close()
1406 existent = nonexistent
1408 self.assertRaises((OSError, IOError), existent.open)
1411 def test_openWithExplicitBinaryMode(self):
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().
1418 See http://bugs.python.org/issue7686 for details about the bug.
1420 writer = self.path.child('explicit-binary')
1421 file = writer.open('wb')
1422 file.write('abc\ndef')
1424 self.assertTrue(writer.exists)
1427 def test_openWithRedundantExplicitBinaryModes(self):
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().
1435 See http://bugs.python.org/issue7686 for details about the bug.
1437 writer = self.path.child('multiple-binary')
1438 file = writer.open('wbb')
1439 file.write('abc\ndef')
1441 self.assertTrue(writer.exists)
1444 def test_existsCache(self):
1446 Check that C{filepath.FilePath.exists} correctly restat the object if
1447 an operation has occurred in the mean time.
1449 fp = filepath.FilePath(self.mktemp())
1450 self.assertEqual(fp.exists(), False)
1453 self.assertEqual(fp.exists(), True)
1456 def test_changed(self):
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}.
1462 fp = filepath.FilePath(self.mktemp())
1463 fp.setContent("12345")
1464 self.assertEqual(fp.getsize(), 5)
1466 # Someone else comes along and changes the file.
1467 fObj = open(fp.path, 'wb')
1468 fObj.write("12345678")
1471 # Sanity check for caching: size should still be 5.
1472 self.assertEqual(fp.getsize(), 5)
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)
1481 def test_getPermissions_POSIX(self):
1483 Getting permissions for a file returns a L{Permissions} object for
1484 POSIX platforms (which supports separate user, group, and other
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(),
1496 def test_getPermissions_Windows(self):
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
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(),
1514 def test_whetherBlockOrSocket(self):
1516 Ensure that a file is not a block or socket
1518 self.assertFalse(self.path.isBlockDevice())
1519 self.assertFalse(self.path.isSocket())
1522 def test_statinfoBitsNotImplementedInWindows(self):
1524 Verify that certain file stats are not available on Windows
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)
1533 def test_statinfoBitsAreNumbers(self):
1535 Verify that file inode/device/nlinks/uid/gid stats are numbers in
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())
1549 def test_statinfoNumbersAreValid(self):
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)
1554 # specify fake statinfo information
1562 # monkey patch in a fake restat method for self.path
1564 def fakeRestat(*args, **kwargs):
1565 self.path.statinfo = fake
1566 self.path.restat = fakeRestat
1568 # ensure that restat will need to be called to get values
1569 self.path.statinfo = None
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)
1578 if platform.isWindows():
1579 test_statinfoBitsAreNumbers.skip = True
1580 test_statinfoNumbersAreValid.skip = True
1581 test_getPermissions_POSIX.skip = True
1583 test_statinfoBitsNotImplementedInWindows.skip = "Test will run only on Windows."
1584 test_getPermissions_Windows.skip = "Test will run only on Windows."
1588 from twisted.python import urlpath
1590 class URLPathTestCase(unittest.TestCase):
1592 self.path = urlpath.URLPath.fromString("http://example.com/foo/bar?yes=no&no=yes#footer")
1594 def testStringConversion(self):
1595 self.assertEqual(str(self.path), "http://example.com/foo/bar?yes=no&no=yes#footer")
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/")
1601 def testSiblingString(self):
1602 self.assertEqual(str(self.path.sibling('baz')), 'http://example.com/foo/baz')
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')
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/')
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/')