/usr/lib/python2.7/dist-packages/twext/python/test/test_filepath.py is in python-twext 0.1.b2.dev15059-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 | ##
# Copyright (c) 2010-2015 Apple Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
##
"""
Tests for specialized behavior of L{CachingFilePath}
"""
from errno import EINVAL
from os.path import join as pathjoin
from twisted.internet.task import Clock
from twisted.trial.unittest import TestCase
from twext.python.filepath import CachingFilePath
# Cheat and pull in the Twisted test cases for FilePath. XXX: Twisteds should
# provide a supported way of doing this for exported interfaces. Also, it
# should export IFilePath. --glyph
# from twisted.test.test_paths import FilePathTests
# class BaseVerification(FilePathTests):
# """
# Make sure that L{CachingFilePath} doesn't break the contracts that
# L{FilePath} tries to provide.
# """
# def setUp(self):
# """
# Set up the test case to set the base attributes to point at
# L{AbstractFilePathTestCase}.
# """
# FilePathTests.setUp(self)
# self.root = CachingFilePath(self.root.path)
# self.path = CachingFilePath(self.path.path)
class EINVALTestCase(TestCase):
"""
Sometimes, L{os.listdir} will raise C{EINVAL}. This is a transient error,
and L{CachingFilePath.listdir} should work around it by retrying the
C{listdir} operation until it succeeds.
"""
def setUp(self):
"""
Create a L{CachingFilePath} for the test to use.
"""
self.cfp = CachingFilePath(self.mktemp())
self.clock = Clock()
self.cfp._sleep = self.clock.advance
def test_testValidity(self):
"""
If C{listdir} is replaced on a L{CachingFilePath}, we should be able to
observe exceptions raised by the replacement. This verifies that the
test patching done here is actually testing something.
"""
class CustomException(Exception):
"Just for testing."
def blowUp(dirname):
raise CustomException()
self.cfp._listdir = blowUp
self.assertRaises(CustomException, self.cfp.listdir)
self.assertRaises(CustomException, self.cfp.children)
def test_retryLoop(self):
"""
L{CachingFilePath} should catch C{EINVAL} and respond by retrying the
C{listdir} operation until it succeeds.
"""
calls = []
def raiseEINVAL(dirname):
calls.append(dirname)
if len(calls) < 5:
raise OSError(EINVAL, "This should be caught by the test.")
return ['a', 'b', 'c']
self.cfp._listdir = raiseEINVAL
self.assertEquals(self.cfp.listdir(), ['a', 'b', 'c'])
self.assertEquals(self.cfp.children(), [
CachingFilePath(pathjoin(self.cfp.path, 'a')),
CachingFilePath(pathjoin(self.cfp.path, 'b')),
CachingFilePath(pathjoin(self.cfp.path, 'c')),
])
def requireTimePassed(self, filenames):
"""
Create a replacement for listdir() which only fires after a certain
amount of time.
"""
self.calls = []
def thunk(dirname):
now = self.clock.seconds()
if now < 20.0:
self.calls.append(now)
raise OSError(EINVAL, "Not enough time has passed yet.")
else:
return filenames
self.cfp._listdir = thunk
def assertRequiredTimePassed(self):
"""
Assert that calls to the simulated time.sleep() installed by
C{requireTimePassed} have been invoked the required number of times.
"""
# Waiting should be growing by *2 each time until the additional wait
# exceeds BACKOFF_MAX (5), at which point we should wait for 5s each
# time.
def cumulative(values):
current = 0.0
for value in values:
current += value
yield current
self.assertEquals(
self.calls,
list(cumulative(
[0.0, 0.1, 0.2, 0.4, 0.8, 1.6, 3.2, 5.0, 5.0]
))
)
def test_backoff(self):
"""
L{CachingFilePath} will wait for an increasing interval up to
C{BACKOFF_MAX} between calls to listdir().
"""
self.requireTimePassed(['a', 'b', 'c'])
self.assertEquals(self.cfp.listdir(), ['a', 'b', 'c'])
# def test_siblingExtensionSearch(self):
# """
# L{FilePath.siblingExtensionSearch} is unfortunately not implemented in
# terms of L{FilePath.listdir}, so we need to verify that it will also
# retry.
# """
# filenames = [self.cfp.basename() + '.a',
# self.cfp.basename() + '.b',
# self.cfp.basename() + '.c']
# siblings = map(self.cfp.sibling, filenames)
# for sibling in siblings:
# sibling.touch()
# self.requireTimePassed(filenames)
# self.assertEquals(self.cfp.siblingExtensionSearch("*"),
# siblings[0])
# self.assertRequiredTimePassed()
|