← Back to branch summary

~landscape/zope3/newer-from-ztk

« back to all changes in this revision

Viewing changes to src/twisted/test/test_epoll.py

  • Committer: Thomas Hervé
  • Date: 2009-07-08 13:52:04 UTC
  • Revision ID: thomas@canonical.com-20090708135204-df5eesrthifpylf8
Remove twisted copy

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/twisted/test/test_epoll.py
1
 
# Copyright (c) 2001-2006 Twisted Matrix Laboratories.
2
 
# See LICENSE for details.
3
 
 
4
 
"""
5
 
Tests for epoll wrapper.
6
 
"""
7
 
 
8
 
import socket, errno, time
9
 
 
10
 
from twisted.trial import unittest
11
 
 
12
 
try:
13
 
    from twisted.python import _epoll
14
 
except ImportError:
15
 
    _epoll = None
16
 
 
17
 
 
18
 
class EPoll(unittest.TestCase):
19
 
    """
20
 
    Tests for the low-level epoll bindings.
21
 
    """
22
 
    def setUp(self):
23
 
        """
24
 
        Create a listening server port and a list with which to keep track
25
 
        of created sockets.
26
 
        """
27
 
        self.serverSocket = socket.socket()
28
 
        self.serverSocket.bind(('127.0.0.1', 0))
29
 
        self.serverSocket.listen(1)
30
 
        self.connections = [self.serverSocket]
31
 
 
32
 
 
33
 
    def tearDown(self):
34
 
        """
35
 
        Close any sockets which were opened by the test.
36
 
        """
37
 
        for skt in self.connections:
38
 
            skt.close()
39
 
 
40
 
 
41
 
    def _connectedPair(self):
42
 
        """
43
 
        Return the two sockets which make up a new TCP connection.
44
 
        """
45
 
        client = socket.socket()
46
 
        client.setblocking(False)
47
 
        try:
48
 
            client.connect(('127.0.0.1', self.serverSocket.getsockname()[1]))
49
 
        except socket.error, e:
50
 
            self.assertEquals(e.args[0], errno.EINPROGRESS)
51
 
        else:
52
 
            raise unittest.FailTest("Connect should have raised EINPROGRESS")
53
 
        server, addr = self.serverSocket.accept()
54
 
 
55
 
        self.connections.extend((client, server))
56
 
        return client, server
57
 
 
58
 
 
59
 
    def test_create(self):
60
 
        """
61
 
        Test the creation of an epoll object.
62
 
        """
63
 
        try:
64
 
            p = _epoll.epoll(16)
65
 
        except OSError, e:
66
 
            raise unittest.FailTest(str(e))
67
 
        else:
68
 
            p.close()
69
 
 
70
 
 
71
 
    def test_badCreate(self):
72
 
        """
73
 
        Test that attempting to create an epoll object with some random
74
 
        objects raises a TypeError.
75
 
        """
76
 
        self.assertRaises(TypeError, _epoll.epoll, 1, 2, 3)
77
 
        self.assertRaises(TypeError, _epoll.epoll, 'foo')
78
 
        self.assertRaises(TypeError, _epoll.epoll, None)
79
 
        self.assertRaises(TypeError, _epoll.epoll, ())
80
 
        self.assertRaises(TypeError, _epoll.epoll, ['foo'])
81
 
        self.assertRaises(TypeError, _epoll.epoll, {})
82
 
        self.assertRaises(TypeError, _epoll.epoll)
83
 
 
84
 
 
85
 
    def test_add(self):
86
 
        """
87
 
        Test adding a socket to an epoll object.
88
 
        """
89
 
        server, client = self._connectedPair()
90
 
 
91
 
        p = _epoll.epoll(2)
92
 
        try:
93
 
            p._control(_epoll.CTL_ADD, server.fileno(), _epoll.IN | _epoll.OUT)
94
 
            p._control(_epoll.CTL_ADD, client.fileno(), _epoll.IN | _epoll.OUT)
95
 
        finally:
96
 
            p.close()
97
 
 
98
 
 
99
 
    def test_controlAndWait(self):
100
 
        """
101
 
        Test waiting on an epoll object which has had some sockets added to
102
 
        it.
103
 
        """
104
 
        client, server = self._connectedPair()
105
 
 
106
 
        p = _epoll.epoll(16)
107
 
        p._control(_epoll.CTL_ADD, client.fileno(), _epoll.IN | _epoll.OUT |
108
 
                   _epoll.ET)
109
 
        p._control(_epoll.CTL_ADD, server.fileno(), _epoll.IN | _epoll.OUT |
110
 
                   _epoll.ET)
111
 
 
112
 
        now = time.time()
113
 
        events = p.wait(4, 1000)
114
 
        then = time.time()
115
 
        self.failIf(then - now > 0.01)
116
 
 
117
 
        events.sort()
118
 
        expected = [(client.fileno(), _epoll.OUT),
119
 
                    (server.fileno(), _epoll.OUT)]
120
 
        expected.sort()
121
 
 
122
 
        self.assertEquals(events, expected)
123
 
 
124
 
        now = time.time()
125
 
        events = p.wait(4, 200)
126
 
        then = time.time()
127
 
        self.failUnless(then - now > 0.1)
128
 
        self.failIf(events)
129
 
 
130
 
        client.send("Hello!")
131
 
        server.send("world!!!")
132
 
 
133
 
        now = time.time()
134
 
        events = p.wait(4, 1000)
135
 
        then = time.time()
136
 
        self.failIf(then - now > 0.01)
137
 
 
138
 
        events.sort()
139
 
        expected = [(client.fileno(), _epoll.IN | _epoll.OUT),
140
 
                    (server.fileno(), _epoll.IN | _epoll.OUT)]
141
 
        expected.sort()
142
 
 
143
 
        self.assertEquals(events, expected)
144
 
 
145
 
if _epoll is None:
146
 
    EPoll.skip = "_epoll module unavailable"
147
 
else:
148
 
    try:
149
 
        e = _epoll.epoll(16)
150
 
    except IOError, exc:
151
 
        if exc.errno == errno.ENOSYS:
152
 
            del exc
153
 
            EPoll.skip = "epoll support missing from platform"
154
 
        else:
155
 
            raise
156
 
    else:
157
 
        e.close()
158
 
        del e