Rietveld Code Review Tool
Help | Bug tracker | Discussion group | Source code | Sign in
(253236)

Delta Between Two Patch Sets: Lib/test/test_kqueue.py

Issue 16510: Using appropriate checks in tests
Left Patch Set: Created 5 years, 11 months ago
Right Patch Set: Created 5 years, 6 months ago
Left:
Right:
Use n/p to move between diff chunks; N/P to move between comments. Please Sign in to add in-line comments.
Jump to:
Left: Side by side diff | Download
Right: Side by side diff | Download
« no previous file with change/comment | « Lib/test/test_json/test_decode.py ('k') | Lib/test/test_logging.py » ('j') | no next file with change/comment »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
LEFTRIGHT
1 """ 1 """
2 Tests for kqueue wrapper. 2 Tests for kqueue wrapper.
3 """ 3 """
4 import errno 4 import errno
5 import os 5 import os
6 import select 6 import select
7 import socket 7 import socket
8 import sys 8 import sys
9 import time 9 import time
10 import unittest 10 import unittest
11 11
12 from test import support 12 from test import support
13 if not hasattr(select, "kqueue"): 13 if not hasattr(select, "kqueue"):
14 raise unittest.SkipTest("test works only on BSD") 14 raise unittest.SkipTest("test works only on BSD")
15 15
16 class TestKQueue(unittest.TestCase): 16 class TestKQueue(unittest.TestCase):
17 def test_create_queue(self): 17 def test_create_queue(self):
18 kq = select.kqueue() 18 kq = select.kqueue()
19 self.assertTrue(kq.fileno() > 0, kq.fileno()) 19 self.assertGreater(kq.fileno(), 0)
20 self.assertFalse(kq.closed) 20 self.assertFalse(kq.closed)
21 kq.close() 21 kq.close()
22 self.assertTrue(kq.closed) 22 self.assertTrue(kq.closed)
23 self.assertRaises(ValueError, kq.fileno) 23 self.assertRaises(ValueError, kq.fileno)
24 24
25 def test_create_event(self): 25 def test_create_event(self):
26 from operator import lt, le, gt, ge 26 from operator import lt, le, gt, ge
27 27
28 fd = os.open(os.devnull, os.O_WRONLY) 28 fd = os.open(os.devnull, os.O_WRONLY)
29 self.addCleanup(os.close, fd) 29 self.addCleanup(os.close, fd)
(...skipping 38 matching lines...) Expand 10 before | Expand all | Expand 10 after
68 ev = select.kevent(1, 2, 3, 4, 5, 6) 68 ev = select.kevent(1, 2, 3, 4, 5, 6)
69 self.assertEqual(ev.ident, 1) 69 self.assertEqual(ev.ident, 1)
70 self.assertEqual(ev.filter, 2) 70 self.assertEqual(ev.filter, 2)
71 self.assertEqual(ev.flags, 3) 71 self.assertEqual(ev.flags, 3)
72 self.assertEqual(ev.fflags, 4) 72 self.assertEqual(ev.fflags, 4)
73 self.assertEqual(ev.data, 5) 73 self.assertEqual(ev.data, 5)
74 self.assertEqual(ev.udata, 6) 74 self.assertEqual(ev.udata, 6)
75 self.assertEqual(ev, ev) 75 self.assertEqual(ev, ev)
76 self.assertNotEqual(ev, other) 76 self.assertNotEqual(ev, other)
77 77
78 bignum = sys.maxsize * 2 + 1 78 bignum = 0x7fff
79 ev = select.kevent(bignum, 1, 2, 3, sys.maxsize, bignum) 79 ev = select.kevent(bignum, 1, 2, 3, bignum - 1, bignum)
80 self.assertEqual(ev.ident, bignum) 80 self.assertEqual(ev.ident, bignum)
81 self.assertEqual(ev.filter, 1) 81 self.assertEqual(ev.filter, 1)
82 self.assertEqual(ev.flags, 2) 82 self.assertEqual(ev.flags, 2)
83 self.assertEqual(ev.fflags, 3) 83 self.assertEqual(ev.fflags, 3)
84 self.assertEqual(ev.data, sys.maxsize) 84 self.assertEqual(ev.data, bignum - 1)
85 self.assertEqual(ev.udata, bignum) 85 self.assertEqual(ev.udata, bignum)
86 self.assertEqual(ev, ev) 86 self.assertEqual(ev, ev)
87 self.assertNotEqual(ev, other) 87 self.assertNotEqual(ev, other)
88 88
89 def test_queue_event(self): 89 def test_queue_event(self):
90 serverSocket = socket.socket() 90 serverSocket = socket.socket()
91 serverSocket.bind(('127.0.0.1', 0)) 91 serverSocket.bind(('127.0.0.1', 0))
92 serverSocket.listen(1) 92 serverSocket.listen(1)
93 client = socket.socket() 93 client = socket.socket()
94 client.setblocking(False) 94 client.setblocking(False)
95 try: 95 try:
96 client.connect(('127.0.0.1', serverSocket.getsockname()[1])) 96 client.connect(('127.0.0.1', serverSocket.getsockname()[1]))
97 except OSError as e: 97 except OSError as e:
98 self.assertEqual(e.args[0], errno.EINPROGRESS) 98 self.assertEqual(e.args[0], errno.EINPROGRESS)
99 else: 99 else:
100 #raise AssertionError("Connect should have raised EINPROGRESS") 100 #raise AssertionError("Connect should have raised EINPROGRESS")
101 pass # FreeBSD doesn't raise an exception here 101 pass # FreeBSD doesn't raise an exception here
102 server, addr = serverSocket.accept() 102 server, addr = serverSocket.accept()
103 103
104 if sys.platform.startswith("darwin"):
105 flags = select.KQ_EV_ADD | select.KQ_EV_ENABLE
106 else:
107 flags = 0
108
109 kq = select.kqueue() 104 kq = select.kqueue()
110 kq2 = select.kqueue.fromfd(kq.fileno()) 105 kq2 = select.kqueue.fromfd(kq.fileno())
111 106
112 ev = select.kevent(server.fileno(), 107 ev = select.kevent(server.fileno(),
113 select.KQ_FILTER_WRITE, 108 select.KQ_FILTER_WRITE,
114 select.KQ_EV_ADD | select.KQ_EV_ENABLE) 109 select.KQ_EV_ADD | select.KQ_EV_ENABLE)
115 kq.control([ev], 0) 110 kq.control([ev], 0)
116 ev = select.kevent(server.fileno(), 111 ev = select.kevent(server.fileno(),
117 select.KQ_FILTER_READ, 112 select.KQ_FILTER_READ,
118 select.KQ_EV_ADD | select.KQ_EV_ENABLE) 113 select.KQ_EV_ADD | select.KQ_EV_ENABLE)
119 kq.control([ev], 0) 114 kq.control([ev], 0)
120 ev = select.kevent(client.fileno(), 115 ev = select.kevent(client.fileno(),
121 select.KQ_FILTER_WRITE, 116 select.KQ_FILTER_WRITE,
122 select.KQ_EV_ADD | select.KQ_EV_ENABLE) 117 select.KQ_EV_ADD | select.KQ_EV_ENABLE)
123 kq2.control([ev], 0) 118 kq2.control([ev], 0)
124 ev = select.kevent(client.fileno(), 119 ev = select.kevent(client.fileno(),
125 select.KQ_FILTER_READ, 120 select.KQ_FILTER_READ,
126 select.KQ_EV_ADD | select.KQ_EV_ENABLE) 121 select.KQ_EV_ADD | select.KQ_EV_ENABLE)
127 kq2.control([ev], 0) 122 kq2.control([ev], 0)
128 123
129 events = kq.control(None, 4, 1) 124 events = kq.control(None, 4, 1)
130 events = [(e.ident, e.filter, e.flags) for e in events] 125 events = set((e.ident, e.filter) for e in events)
131 events.sort() 126 self.assertEqual(events, set([
132 self.assertEqual(events, [ 127 (client.fileno(), select.KQ_FILTER_WRITE),
133 (client.fileno(), select.KQ_FILTER_WRITE, flags), 128 (server.fileno(), select.KQ_FILTER_WRITE)]))
134 (server.fileno(), select.KQ_FILTER_WRITE, flags)])
135 129
136 client.send(b"Hello!") 130 client.send(b"Hello!")
137 server.send(b"world!!!") 131 server.send(b"world!!!")
138 132
139 # We may need to call it several times 133 # We may need to call it several times
140 for i in range(10): 134 for i in range(10):
141 events = kq.control(None, 4, 1) 135 events = kq.control(None, 4, 1)
142 if len(events) == 4: 136 if len(events) == 4:
143 break 137 break
144 time.sleep(1.0) 138 time.sleep(1.0)
145 else: 139 else:
146 self.fail('timeout waiting for event notifications') 140 self.fail('timeout waiting for event notifications')
147 141
148 events = [(e.ident, e.filter, e.flags) for e in events] 142 events = set((e.ident, e.filter) for e in events)
149 events.sort() 143 self.assertEqual(events, set([
150 144 (client.fileno(), select.KQ_FILTER_WRITE),
151 self.assertEqual(events, [ 145 (client.fileno(), select.KQ_FILTER_READ),
152 (client.fileno(), select.KQ_FILTER_WRITE, flags), 146 (server.fileno(), select.KQ_FILTER_WRITE),
153 (client.fileno(), select.KQ_FILTER_READ, flags), 147 (server.fileno(), select.KQ_FILTER_READ)]))
154 (server.fileno(), select.KQ_FILTER_WRITE, flags),
155 (server.fileno(), select.KQ_FILTER_READ, flags)])
156 148
157 # Remove completely client, and server read part 149 # Remove completely client, and server read part
158 ev = select.kevent(client.fileno(), 150 ev = select.kevent(client.fileno(),
159 select.KQ_FILTER_WRITE, 151 select.KQ_FILTER_WRITE,
160 select.KQ_EV_DELETE) 152 select.KQ_EV_DELETE)
161 kq.control([ev], 0) 153 kq.control([ev], 0)
162 ev = select.kevent(client.fileno(), 154 ev = select.kevent(client.fileno(),
163 select.KQ_FILTER_READ, 155 select.KQ_FILTER_READ,
164 select.KQ_EV_DELETE) 156 select.KQ_EV_DELETE)
165 kq.control([ev], 0) 157 kq.control([ev], 0)
166 ev = select.kevent(server.fileno(), 158 ev = select.kevent(server.fileno(),
167 select.KQ_FILTER_READ, 159 select.KQ_FILTER_READ,
168 select.KQ_EV_DELETE) 160 select.KQ_EV_DELETE)
169 kq.control([ev], 0, 0) 161 kq.control([ev], 0, 0)
170 162
171 events = kq.control([], 4, 0.99) 163 events = kq.control([], 4, 0.99)
172 events = [(e.ident, e.filter, e.flags) for e in events] 164 events = set((e.ident, e.filter) for e in events)
173 events.sort() 165 self.assertEqual(events, set([
174 self.assertEqual(events, [ 166 (server.fileno(), select.KQ_FILTER_WRITE)]))
175 (server.fileno(), select.KQ_FILTER_WRITE, flags)])
176 167
177 client.close() 168 client.close()
178 server.close() 169 server.close()
179 serverSocket.close() 170 serverSocket.close()
180 171
181 def testPair(self): 172 def testPair(self):
182 kq = select.kqueue() 173 kq = select.kqueue()
183 a, b = socket.socketpair() 174 a, b = socket.socketpair()
184 175
185 a.send(b'foo') 176 a.send(b'foo')
186 event1 = select.kevent(a, select.KQ_FILTER_READ, select.KQ_EV_ADD | sele ct.KQ_EV_ENABLE) 177 event1 = select.kevent(a, select.KQ_FILTER_READ, select.KQ_EV_ADD | sele ct.KQ_EV_ENABLE)
187 event2 = select.kevent(b, select.KQ_FILTER_READ, select.KQ_EV_ADD | sele ct.KQ_EV_ENABLE) 178 event2 = select.kevent(b, select.KQ_FILTER_READ, select.KQ_EV_ADD | sele ct.KQ_EV_ENABLE)
188 r = kq.control([event1, event2], 1, 1) 179 r = kq.control([event1, event2], 1, 1)
189 self.assertTrue(r) 180 self.assertTrue(r)
190 self.assertFalse(r[0].flags & select.KQ_EV_ERROR) 181 self.assertFalse(r[0].flags & select.KQ_EV_ERROR)
191 self.assertEqual(b.recv(r[0].data), b'foo') 182 self.assertEqual(b.recv(r[0].data), b'foo')
192 183
193 a.close() 184 a.close()
194 b.close() 185 b.close()
195 kq.close() 186 kq.close()
196 187
188 def test_close(self):
189 open_file = open(__file__, "rb")
190 self.addCleanup(open_file.close)
191 fd = open_file.fileno()
192 kqueue = select.kqueue()
193
194 # test fileno() method and closed attribute
195 self.assertIsInstance(kqueue.fileno(), int)
196 self.assertFalse(kqueue.closed)
197
198 # test close()
199 kqueue.close()
200 self.assertTrue(kqueue.closed)
201 self.assertRaises(ValueError, kqueue.fileno)
202
203 # close() can be called more than once
204 kqueue.close()
205
206 # operations must fail with ValueError("I/O operation on closed ...")
207 self.assertRaises(ValueError, kqueue.control, None, 4)
208
209 def test_fd_non_inheritable(self):
210 kqueue = select.kqueue()
211 self.addCleanup(kqueue.close)
212 self.assertEqual(os.get_inheritable(kqueue.fileno()), False)
213
214
197 def test_main(): 215 def test_main():
198 support.run_unittest(TestKQueue) 216 support.run_unittest(TestKQueue)
199 217
200 if __name__ == "__main__": 218 if __name__ == "__main__":
201 test_main() 219 test_main()
LEFTRIGHT

RSS Feeds Recent Issues | This issue
This is Rietveld 894c83f36cb7+