1 # Copyright (c) Twisted Matrix Laboratories.
2 # See LICENSE for details.
5 Tests for L{twisted.internet.fdesc}.
14 skip = "not supported on this platform"
16 from twisted.internet import fdesc
18 from twisted.python.util import untilConcludes
19 from twisted.trial import unittest
22 class ReadWriteTestCase(unittest.TestCase):
24 Tests for fdesc.readFromFD, fdesc.writeToFD.
29 Create two non-blocking pipes that can be used in tests.
31 self.r, self.w = os.pipe()
32 fdesc.setNonBlocking(self.r)
33 fdesc.setNonBlocking(self.w)
52 Write data to the pipe.
54 return fdesc.writeToFD(self.w, d)
59 Read data from the pipe.
62 res = fdesc.readFromFD(self.r, l.append)
72 def test_writeAndRead(self):
74 Test that the number of bytes L{fdesc.writeToFD} reports as written
75 with its return value are seen by L{fdesc.readFromFD}.
77 n = self.write("hello")
78 self.failUnless(n > 0)
80 self.assertEqual(len(s), n)
81 self.assertEqual("hello"[:n], s)
84 def test_writeAndReadLarge(self):
86 Similar to L{test_writeAndRead}, but use a much larger string to verify
87 the behavior for that case.
89 orig = "0123456879" * 10000
90 written = self.write(orig)
91 self.failUnless(written > 0)
95 while resultlength < written or i < 50:
96 result.append(self.read())
97 resultlength += len(result[-1])
98 # Increment a counter to be sure we'll exit at some point
100 result = "".join(result)
101 self.assertEqual(len(result), written)
102 self.assertEqual(orig[:written], result)
105 def test_readFromEmpty(self):
107 Verify that reading from a file descriptor with no data does not raise
108 an exception and does not result in the callback function being called.
111 result = fdesc.readFromFD(self.r, l.append)
112 self.assertEqual(l, [])
113 self.assertEqual(result, None)
116 def test_readFromCleanClose(self):
118 Test that using L{fdesc.readFromFD} on a cleanly closed file descriptor
119 returns a connection done indicator.
122 self.assertEqual(self.read(), fdesc.CONNECTION_DONE)
125 def test_writeToClosed(self):
127 Verify that writing with L{fdesc.writeToFD} when the read end is closed
128 results in a connection lost indicator.
131 self.assertEqual(self.write("s"), fdesc.CONNECTION_LOST)
134 def test_readFromInvalid(self):
136 Verify that reading with L{fdesc.readFromFD} when the read end is
137 closed results in a connection lost indicator.
140 self.assertEqual(self.read(), fdesc.CONNECTION_LOST)
143 def test_writeToInvalid(self):
145 Verify that writing with L{fdesc.writeToFD} when the write end is
146 closed results in a connection lost indicator.
149 self.assertEqual(self.write("s"), fdesc.CONNECTION_LOST)
152 def test_writeErrors(self):
154 Test error path for L{fdesc.writeTod}.
156 oldOsWrite = os.write
157 def eagainWrite(fd, data):
159 err.errno = errno.EAGAIN
161 os.write = eagainWrite
163 self.assertEqual(self.write("s"), 0)
165 os.write = oldOsWrite
167 def eintrWrite(fd, data):
169 err.errno = errno.EINTR
171 os.write = eintrWrite
173 self.assertEqual(self.write("s"), 0)
175 os.write = oldOsWrite
179 class CloseOnExecTests(unittest.TestCase):
181 Tests for L{fdesc._setCloseOnExec} and L{fdesc._unsetCloseOnExec}.
188 if e.errno == errno.EBADF:
197 def _execWithFileDescriptor(self, fObj):
201 os.execv(sys.executable, [sys.executable, '-c', self.program % (fObj.fileno(),)])
204 traceback.print_exc()
207 # On Linux wait(2) doesn't seem ever able to fail with EINTR but
208 # POSIX seems to allow it and on OS X it happens quite a lot.
209 return untilConcludes(os.waitpid, pid, 0)[1]
212 def test_setCloseOnExec(self):
214 A file descriptor passed to L{fdesc._setCloseOnExec} is not inherited
215 by a new process image created with one of the exec family of
218 fObj = file(self.mktemp(), 'w')
219 fdesc._setCloseOnExec(fObj.fileno())
220 status = self._execWithFileDescriptor(fObj)
221 self.assertTrue(os.WIFEXITED(status))
222 self.assertEqual(os.WEXITSTATUS(status), 0)
225 def test_unsetCloseOnExec(self):
227 A file descriptor passed to L{fdesc._unsetCloseOnExec} is inherited by
228 a new process image created with one of the exec family of functions.
230 fObj = file(self.mktemp(), 'w')
231 fdesc._setCloseOnExec(fObj.fileno())
232 fdesc._unsetCloseOnExec(fObj.fileno())
233 status = self._execWithFileDescriptor(fObj)
234 self.assertTrue(os.WIFEXITED(status))
235 self.assertEqual(os.WEXITSTATUS(status), 20)