twisted/test/test_fdesc.py

   1 # Copyright (c) Twisted Matrix Laboratories.
   2 # See LICENSE for details.
   3
   4 """
   5 Tests for L{twisted.internet.fdesc}.
   6 """
   7
   8 import os, sys
   9 import errno
  10
  11 try:
  12     import fcntl
  13 except ImportError:
  14     skip = "not supported on this platform"
  15 else:
  16     from twisted.internet import fdesc
  17
  18 from twisted.python.util import untilConcludes
  19 from twisted.trial import unittest
  20
  21
  22 class ReadWriteTestCase(unittest.TestCase):
  23     """
  24     Tests for fdesc.readFromFD, fdesc.writeToFD.
  25     """
  26
  27     def setUp(self):
  28         """
  29         Create two non-blocking pipes that can be used in tests.
  30         """
  31         self.r, self.w = os.pipe()
  32         fdesc.setNonBlocking(self.r)
  33         fdesc.setNonBlocking(self.w)
  34
  35
  36     def tearDown(self):
  37         """
  38         Close pipes.
  39         """
  40         try:
  41             os.close(self.w)
  42         except OSError:
  43             pass
  44         try:
  45             os.close(self.r)
  46         except OSError:
  47             pass
  48
  49
  50     def write(self, d):
  51         """
  52         Write data to the pipe.
  53         """
  54         return fdesc.writeToFD(self.w, d)
  55
  56
  57     def read(self):
  58         """
  59         Read data from the pipe.
  60         """
  61         l = []
  62         res = fdesc.readFromFD(self.r, l.append)
  63         if res is None:
  64             if l:
  65                 return l[0]
  66             else:
  67                 return ""
  68         else:
  69             return res
  70
  71
  72     def test_writeAndRead(self):
  73         """
  74         Test that the number of bytes L{fdesc.writeToFD} reports as written
  75         with its return value are seen by L{fdesc.readFromFD}.
  76         """
  77         n = self.write("hello")
  78         self.failUnless(n > 0)
  79         s = self.read()
  80         self.assertEqual(len(s), n)
  81         self.assertEqual("hello"[:n], s)
  82
  83
  84     def test_writeAndReadLarge(self):
  85         """
  86         Similar to L{test_writeAndRead}, but use a much larger string to verify
  87         the behavior for that case.
  88         """
  89         orig = "0123456879" * 10000
  90         written = self.write(orig)
  91         self.failUnless(written > 0)
  92         result = []
  93         resultlength = 0
  94         i = 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
  99             i += 1
 100         result = "".join(result)
 101         self.assertEqual(len(result), written)
 102         self.assertEqual(orig[:written], result)
 103
 104
 105     def test_readFromEmpty(self):
 106         """
 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.
 109         """
 110         l = []
 111         result = fdesc.readFromFD(self.r, l.append)
 112         self.assertEqual(l, [])
 113         self.assertEqual(result, None)
 114
 115
 116     def test_readFromCleanClose(self):
 117         """
 118         Test that using L{fdesc.readFromFD} on a cleanly closed file descriptor
 119         returns a connection done indicator.
 120         """
 121         os.close(self.w)
 122         self.assertEqual(self.read(), fdesc.CONNECTION_DONE)
 123
 124
 125     def test_writeToClosed(self):
 126         """
 127         Verify that writing with L{fdesc.writeToFD} when the read end is closed
 128         results in a connection lost indicator.
 129         """
 130         os.close(self.r)
 131         self.assertEqual(self.write("s"), fdesc.CONNECTION_LOST)
 132
 133
 134     def test_readFromInvalid(self):
 135         """
 136         Verify that reading with L{fdesc.readFromFD} when the read end is
 137         closed results in a connection lost indicator.
 138         """
 139         os.close(self.r)
 140         self.assertEqual(self.read(), fdesc.CONNECTION_LOST)
 141
 142
 143     def test_writeToInvalid(self):
 144         """
 145         Verify that writing with L{fdesc.writeToFD} when the write end is
 146         closed results in a connection lost indicator.
 147         """
 148         os.close(self.w)
 149         self.assertEqual(self.write("s"), fdesc.CONNECTION_LOST)
 150
 151
 152     def test_writeErrors(self):
 153         """
 154         Test error path for L{fdesc.writeTod}.
 155         """
 156         oldOsWrite = os.write
 157         def eagainWrite(fd, data):
 158             err = OSError()
 159             err.errno = errno.EAGAIN
 160             raise err
 161         os.write = eagainWrite
 162         try:
 163             self.assertEqual(self.write("s"), 0)
 164         finally:
 165             os.write = oldOsWrite
 166
 167         def eintrWrite(fd, data):
 168             err = OSError()
 169             err.errno = errno.EINTR
 170             raise err
 171         os.write = eintrWrite
 172         try:
 173             self.assertEqual(self.write("s"), 0)
 174         finally:
 175             os.write = oldOsWrite
 176
 177
 178
 179 class CloseOnExecTests(unittest.TestCase):
 180     """
 181     Tests for L{fdesc._setCloseOnExec} and L{fdesc._unsetCloseOnExec}.
 182     """
 183     program = '''
 184 import os, errno
 185 try:
 186     os.write(%d, 'lul')
 187 except OSError, e:
 188     if e.errno == errno.EBADF:
 189         os._exit(0)
 190     os._exit(5)
 191 except:
 192     os._exit(10)
 193 else:
 194     os._exit(20)
 195 '''
 196
 197     def _execWithFileDescriptor(self, fObj):
 198         pid = os.fork()
 199         if pid == 0:
 200             try:
 201                 os.execv(sys.executable, [sys.executable, '-c', self.program % (fObj.fileno(),)])
 202             except:
 203                 import traceback
 204                 traceback.print_exc()
 205                 os._exit(30)
 206         else:
 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]
 210
 211
 212     def test_setCloseOnExec(self):
 213         """
 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
 216         functions.
 217         """
 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)
 223
 224
 225     def test_unsetCloseOnExec(self):
 226         """
 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.
 229         """
 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)