1 # Copyright (c) Twisted Matrix Laboratories.
2 # See LICENSE for details.
5 Tests for POSIX-based L{IReactorProcess} implementations.
13 platformSkip = "non-POSIX platform"
15 from twisted.internet import process
18 from twisted.trial.unittest import TestCase
21 class FakeFile(object):
23 A dummy file object which records when it is closed.
25 def __init__(self, testcase, fd):
26 self.testcase = testcase
31 self.testcase._files.remove(self.fd)
35 class FakeResourceModule(object):
37 Fake version of L{resource} which hard-codes a particular rlimit for maximum
40 @ivar _limit: The value to return for the hard limit of number of open files.
44 def __init__(self, limit):
48 def getrlimit(self, no):
50 A fake of L{resource.getrlimit} which returns a pre-determined result.
52 if no == self.RLIMIT_NOFILE:
53 return [0, self._limit]
58 class FDDetectorTests(TestCase):
60 Tests for _FDDetector class in twisted.internet.process, which detects
61 which function to drop in place for the _listOpenFDs method.
63 @ivar devfs: A flag indicating whether the filesystem fake will indicate
66 @ivar accurateDevFDResults: A flag indicating whether the /dev/fd fake
67 returns accurate open file information.
69 @ivar procfs: A flag indicating whether the filesystem fake will indicate
70 that /proc/<pid>/fd exists.
75 accurateDevFDResults = False
81 Fake os.getpid, always return the same thing
86 def listdir(self, arg):
88 Fake os.listdir, depending on what mode we're in to simulate behaviour.
90 @param arg: the directory to list
92 accurate = map(str, self._files)
93 if self.procfs and arg == ('/proc/%d/fd' % (self.getpid(),)):
95 if self.devfs and arg == '/dev/fd':
96 if self.accurateDevFDResults:
98 return ["0", "1", "2"]
102 def openfile(self, fname, mode):
104 This is a mock for L{open}. It keeps track of opened files so extra
105 descriptors can be returned from the mock for L{os.listdir} when used on
106 one of the list-of-filedescriptors directories.
108 A L{FakeFile} is returned which can be closed to remove the new
109 descriptor from the open list.
111 # Find the smallest unused file descriptor and give it to the new file.
112 f = FakeFile(self, min(set(range(1024)) - set(self._files)))
113 self._files.append(f.fd)
117 def hideResourceModule(self):
119 Make the L{resource} module unimportable for the remainder of the
122 sys.modules['resource'] = None
125 def revealResourceModule(self, limit):
127 Make a L{FakeResourceModule} instance importable at the L{resource}
130 @param limit: The value which will be returned for the hard limit of
131 number of open files by the fake resource module's C{getrlimit}
134 sys.modules['resource'] = FakeResourceModule(limit)
137 def replaceResourceModule(self, value):
139 Restore the original resource module to L{sys.modules}.
143 del sys.modules['resource']
147 sys.modules['resource'] = value
152 Set up the tests, giving ourselves a detector object to play with and
153 setting up its testable knobs to refer to our mocked versions.
155 self.detector = process._FDDetector()
156 self.detector.listdir = self.listdir
157 self.detector.getpid = self.getpid
158 self.detector.openfile = self.openfile
159 self._files = [0, 1, 2]
161 self.replaceResourceModule, sys.modules.get('resource'))
164 def test_selectFirstWorking(self):
166 L{FDDetector._getImplementation} returns the first method from its
167 C{_implementations} list which returns results which reflect a newly
168 opened file descriptor.
170 def failWithException():
171 raise ValueError("This does not work")
173 def failWithWrongResults():
177 return self._files[:]
179 self.detector._implementations = [
180 failWithException, failWithWrongResults, correct]
182 self.assertIdentical(correct, self.detector._getImplementation())
185 def test_selectLast(self):
187 L{FDDetector._getImplementation} returns the last method from its
188 C{_implementations} list if none of the implementations manage to return
189 results which reflect a newly opened file descriptor.
191 def failWithWrongResults():
194 def failWithOtherWrongResults():
197 self.detector._implementations = [
198 failWithWrongResults, failWithOtherWrongResults]
200 self.assertIdentical(
201 failWithOtherWrongResults, self.detector._getImplementation())
204 def test_identityOfListOpenFDsChanges(self):
206 Check that the identity of _listOpenFDs changes after running
207 _listOpenFDs the first time, but not after the second time it's run.
209 In other words, check that the monkey patching actually works.
211 # Create a new instance
212 detector = process._FDDetector()
214 first = detector._listOpenFDs.func_name
215 detector._listOpenFDs()
216 second = detector._listOpenFDs.func_name
217 detector._listOpenFDs()
218 third = detector._listOpenFDs.func_name
220 self.assertNotEqual(first, second)
221 self.assertEqual(second, third)
224 def test_devFDImplementation(self):
226 L{_FDDetector._devFDImplementation} raises L{OSError} if there is no
227 I{/dev/fd} directory, otherwise it returns the basenames of its children
228 interpreted as integers.
231 self.assertRaises(OSError, self.detector._devFDImplementation)
233 self.accurateDevFDResults = False
234 self.assertEqual([0, 1, 2], self.detector._devFDImplementation())
237 def test_procFDImplementation(self):
239 L{_FDDetector._procFDImplementation} raises L{OSError} if there is no
240 I{/proc/<pid>/fd} directory, otherwise it returns the basenames of its
241 children interpreted as integers.
244 self.assertRaises(OSError, self.detector._procFDImplementation)
246 self.assertEqual([0, 1, 2], self.detector._procFDImplementation())
249 def test_resourceFDImplementation(self):
251 L{_FDDetector._fallbackFDImplementation} uses the L{resource} module if
252 it is available, returning a range of integers from 0 to the the
253 minimum of C{1024} and the hard I{NOFILE} limit.
255 # When the resource module is here, use its value.
256 self.revealResourceModule(512)
258 range(512), self.detector._fallbackFDImplementation())
260 # But limit its value to the arbitrarily selected value 1024.
261 self.revealResourceModule(2048)
263 range(1024), self.detector._fallbackFDImplementation())
266 def test_fallbackFDImplementation(self):
268 L{_FDDetector._fallbackFDImplementation}, the implementation of last
269 resort, succeeds with a fixed range of integers from 0 to 1024 when the
270 L{resource} module is not importable.
272 self.hideResourceModule()
273 self.assertEqual(range(1024), self.detector._fallbackFDImplementation())
277 class FileDescriptorTests(TestCase):
279 Tests for L{twisted.internet.process._listOpenFDs}
283 def test_openFDs(self):
285 File descriptors returned by L{_listOpenFDs} are mostly open.
287 This test assumes that zero-legth writes fail with EBADF on closed
290 for fd in process._listOpenFDs():
292 fcntl.fcntl(fd, fcntl.F_GETFL)
295 errno.EBADF, err.errno,
296 "fcntl(%d, F_GETFL) failed with unexpected errno %d" % (
300 def test_expectedFDs(self):
302 L{_listOpenFDs} lists expected file descriptors.
304 # This is a tricky test. A priori, there is no way to know what file
305 # descriptors are open now, so there is no way to know what _listOpenFDs
306 # should return. Work around this by creating some new file descriptors
307 # which we can know the state of and then just making assertions about
308 # their presence or absence in the result.
310 # Expect a file we just opened to be listed.
312 openfds = process._listOpenFDs()
313 self.assertIn(f.fileno(), openfds)
315 # Expect a file we just closed not to be listed - with a caveat. The
316 # implementation may need to open a file to discover the result. That
317 # open file descriptor will be allocated the same number as the one we
318 # just closed. So, instead, create a hole in the file descriptor space
319 # to catch that internal descriptor and make the assertion about a
320 # different closed file descriptor.
322 # This gets allocated a file descriptor larger than f's, since nothing
323 # has been closed since we opened f.
324 fd = os.dup(f.fileno())
326 # But sanity check that; if it fails the test is invalid.
329 "Expected duplicate file descriptor to be greater than original")
332 # Get rid of the original, creating the hole. The copy should still
333 # be open, of course.
335 self.assertIn(fd, process._listOpenFDs())
337 # Get rid of the copy now
339 # And it should not appear in the result.
340 self.assertNotIn(fd, process._listOpenFDs())