← Back to branch summary

~ntt-pf-lab/nova/monkey_patch_notification

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/test/test_fdesc.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/Twisted-10.0.0/twisted/test/test_fdesc.py
 
1
# Copyright (c) 2007-2009 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.assertEquals(len(s), n)
 
81
        self.assertEquals("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.assertEquals(len(result), written)
 
102
        self.assertEquals(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.assertEquals(l, [])
 
113
        self.assertEquals(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.assertEquals(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.assertEquals(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.assertEquals(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.assertEquals(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.assertEquals(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.assertEquals(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)