cpython/Lib/test/test_kqueue.py
Christian Heimes 4fbc72b4ac Merged revisions 61687-61688,61696,61700,61704-61705,61707-61709,61711-61712,61714-61716,61718-61722 via svnmerge from
svn+ssh://pythondev@svn.python.org/python/trunk

........
  r61687 | jeffrey.yasskin | 2008-03-21 06:02:44 +0100 (Fri, 21 Mar 2008) | 2 lines

  Speed up test_signal from ~24s to 4s by avoiding nearly all of the sleep calls.
........
  r61688 | jeffrey.yasskin | 2008-03-21 06:51:37 +0100 (Fri, 21 Mar 2008) | 5 lines

  Try to fix test_signal breakages on Linux due to r61687. It appears that at
  least two of the linux build bots aren't leaving zombie processes around for
  os.waitpid to wait for, causing ECHILD errors. This would be a symptom of a bug
  somewhere, but probably not in signal itself.
........
  r61696 | georg.brandl | 2008-03-21 15:32:33 +0100 (Fri, 21 Mar 2008) | 2 lines

  Mark the descitems in the tutorial as "noindex" so that :meth: cross-refs don't link to them.
........
  r61700 | georg.brandl | 2008-03-21 18:19:29 +0100 (Fri, 21 Mar 2008) | 2 lines

  Fix markup.
........
  r61704 | jeffrey.yasskin | 2008-03-21 19:25:06 +0100 (Fri, 21 Mar 2008) | 3 lines

  Try to fix test_signal on FreeBSD. I'm assuming that os.kill is failing to
  raise a signal, but switching to subprocess makes the code cleaner anyway.
........
  r61705 | jeffrey.yasskin | 2008-03-21 19:48:04 +0100 (Fri, 21 Mar 2008) | 7 lines

  Speed test_threading up from 14s to .5s, and avoid a deadlock on certain
  failures. The test for enumerate-after-join is now a little less rigorous, but
  the bug it references says the error happened in the first couple iterations,
  so 100 iterations should still be enough.

  cProfile was useful for identifying the slow tests here.
........
  r61707 | georg.brandl | 2008-03-21 20:14:38 +0100 (Fri, 21 Mar 2008) | 2 lines

  Fix a code block in __future__ docs.
........
  r61708 | georg.brandl | 2008-03-21 20:20:21 +0100 (Fri, 21 Mar 2008) | 2 lines

  Add docs for __func__ and __self__ on methods.
........
  r61709 | georg.brandl | 2008-03-21 20:37:57 +0100 (Fri, 21 Mar 2008) | 2 lines

  Add docs for print_function and future_builtins. Fixes #2442.
........
  r61711 | georg.brandl | 2008-03-21 20:54:00 +0100 (Fri, 21 Mar 2008) | 2 lines

  #2136: allow single quotes in realm spec.
........
  r61712 | georg.brandl | 2008-03-21 21:01:51 +0100 (Fri, 21 Mar 2008) | 3 lines

  Issue #2432: give DictReader the dialect and line_num attributes
  advertised in the docs.
........
  r61714 | georg.brandl | 2008-03-21 21:11:46 +0100 (Fri, 21 Mar 2008) | 2 lines

  #2358: add py3k warning to sys.exc_clear().
........
  r61715 | georg.brandl | 2008-03-21 21:21:46 +0100 (Fri, 21 Mar 2008) | 2 lines

  #2346/#2347: add py3k warning for __methods__ and __members__. Patch by Jack Diederich.
........
  r61716 | georg.brandl | 2008-03-21 21:38:24 +0100 (Fri, 21 Mar 2008) | 2 lines

  #2348: add py3k warning for file.softspace.
........
  r61718 | georg.brandl | 2008-03-21 21:55:20 +0100 (Fri, 21 Mar 2008) | 2 lines

  #2160: document PyImport_GetImporter.
........
  r61719 | georg.brandl | 2008-03-21 21:55:51 +0100 (Fri, 21 Mar 2008) | 2 lines

  Update doc ACKS.
........
  r61720 | steven.bethard | 2008-03-21 22:01:18 +0100 (Fri, 21 Mar 2008) | 1 line

  Replace hack in regrtest.py with use of sys.py3kwarning.
........
  r61721 | georg.brandl | 2008-03-21 22:05:03 +0100 (Fri, 21 Mar 2008) | 2 lines

  Add missing versionadded tag.
........
  r61722 | christian.heimes | 2008-03-22 00:49:44 +0100 (Sat, 22 Mar 2008) | 3 lines

  Applied patch #1657 epoll and kqueue wrappers for the select module
  The patch adds wrappers for the Linux epoll syscalls and the BSD kqueue syscalls. Thanks to Thomas Herve and the Twisted people for their support and help.
  TODO: Finish documentation documentation
........
2008-03-22 00:47:35 +00:00

167 lines
5.8 KiB
Python

"""
Tests for kqueue wrapper.
"""
import socket
import errno
import time
import select
import sys
import unittest
from test import test_support
if not hasattr(select, "kqueue"):
raise test_support.TestSkipped("test works only on BSD")
class TestKQueue(unittest.TestCase):
def test_create_queue(self):
kq = select.kqueue()
self.assert_(kq.fileno() > 0, kq.fileno())
self.assert_(not kq.closed)
kq.close()
self.assert_(kq.closed)
self.assertRaises(ValueError, kq.fileno)
def test_create_event(self):
fd = sys.stderr.fileno()
ev = select.kevent(fd)
other = select.kevent(1000)
self.assertEqual(ev.ident, fd)
self.assertEqual(ev.filter, select.KQ_FILTER_READ)
self.assertEqual(ev.flags, select.KQ_EV_ADD)
self.assertEqual(ev.fflags, 0)
self.assertEqual(ev.data, 0)
self.assertEqual(ev.udata, 0)
self.assertEqual(ev, ev)
self.assertNotEqual(ev, other)
self.assertEqual(cmp(ev, other), -1)
self.assert_(ev < other)
self.assert_(other >= ev)
self.assertRaises(TypeError, cmp, ev, None)
self.assertRaises(TypeError, cmp, ev, 1)
self.assertRaises(TypeError, cmp, ev, "ev")
ev = select.kevent(fd, select.KQ_FILTER_WRITE)
self.assertEqual(ev.ident, fd)
self.assertEqual(ev.filter, select.KQ_FILTER_WRITE)
self.assertEqual(ev.flags, select.KQ_EV_ADD)
self.assertEqual(ev.fflags, 0)
self.assertEqual(ev.data, 0)
self.assertEqual(ev.udata, 0)
self.assertEqual(ev, ev)
self.assertNotEqual(ev, other)
ev = select.kevent(fd, select.KQ_FILTER_WRITE, select.KQ_EV_ONESHOT)
self.assertEqual(ev.ident, fd)
self.assertEqual(ev.filter, select.KQ_FILTER_WRITE)
self.assertEqual(ev.flags, select.KQ_EV_ONESHOT)
self.assertEqual(ev.fflags, 0)
self.assertEqual(ev.data, 0)
self.assertEqual(ev.udata, 0)
self.assertEqual(ev, ev)
self.assertNotEqual(ev, other)
ev = select.kevent(1, 2, 3, 4, 5, 6)
self.assertEqual(ev.ident, 1)
self.assertEqual(ev.filter, 2)
self.assertEqual(ev.flags, 3)
self.assertEqual(ev.fflags, 4)
self.assertEqual(ev.data, 5)
self.assertEqual(ev.udata, 6)
self.assertEqual(ev, ev)
self.assertNotEqual(ev, other)
def test_queue_event(self):
serverSocket = socket.socket()
serverSocket.bind(('127.0.0.1', 0))
serverSocket.listen(1)
client = socket.socket()
client.setblocking(False)
try:
client.connect(('127.0.0.1', serverSocket.getsockname()[1]))
except socket.error as e:
self.assertEquals(e.args[0], errno.EINPROGRESS)
else:
#raise AssertionError("Connect should have raised EINPROGRESS")
pass # FreeBSD doesn't raise an exception here
server, addr = serverSocket.accept()
if sys.platform.startswith("darwin"):
flags = select.KQ_EV_ADD | select.KQ_EV_ENABLE
else:
flags = 0
kq = select.kqueue()
kq2 = select.kqueue.fromfd(kq.fileno())
ev = select.kevent(server.fileno(),
select.KQ_FILTER_WRITE,
select.KQ_EV_ADD | select.KQ_EV_ENABLE)
kq.control([ev], 0)
ev = select.kevent(server.fileno(),
select.KQ_FILTER_READ,
select.KQ_EV_ADD | select.KQ_EV_ENABLE)
kq.control([ev], 0)
ev = select.kevent(client.fileno(),
select.KQ_FILTER_WRITE,
select.KQ_EV_ADD | select.KQ_EV_ENABLE)
kq2.control([ev], 0)
ev = select.kevent(client.fileno(),
select.KQ_FILTER_READ,
select.KQ_EV_ADD | select.KQ_EV_ENABLE)
kq2.control([ev], 0)
events = kq.control(None, 4, 1)
events = [(e.ident, e.filter, e.flags) for e in events]
events.sort()
self.assertEquals(events, [
(client.fileno(), select.KQ_FILTER_WRITE, flags),
(server.fileno(), select.KQ_FILTER_WRITE, flags)])
client.send(b"Hello!")
server.send(b"world!!!")
events = kq.control(None, 4, 1)
# We may need to call it several times
for i in range(5):
if len(events) == 4:
break
events = kq.control(None, 4, 1)
events = [(e.ident, e.filter, e.flags) for e in events]
events.sort()
self.assertEquals(events, [
(client.fileno(), select.KQ_FILTER_WRITE, flags),
(client.fileno(), select.KQ_FILTER_READ, flags),
(server.fileno(), select.KQ_FILTER_WRITE, flags),
(server.fileno(), select.KQ_FILTER_READ, flags)])
# Remove completely client, and server read part
ev = select.kevent(client.fileno(),
select.KQ_FILTER_WRITE,
select.KQ_EV_DELETE)
kq.control([ev], 0)
ev = select.kevent(client.fileno(),
select.KQ_FILTER_READ,
select.KQ_EV_DELETE)
kq.control([ev], 0)
ev = select.kevent(server.fileno(),
select.KQ_FILTER_READ,
select.KQ_EV_DELETE)
kq.control([ev], 0, 0)
events = kq.control([], 4, 0.99)
events = [(e.ident, e.filter, e.flags) for e in events]
events.sort()
self.assertEquals(events, [
(server.fileno(), select.KQ_FILTER_WRITE, flags)])
client.close()
server.close()
serverSocket.close()
def test_main():
test_support.run_unittest(TestKQueue)
if __name__ == "__main__":
test_main()