| Server IP : 79.136.114.73 / Your IP : 216.73.216.107 Web Server : Apache/2.4.7 (Ubuntu) PHP/5.5.9-1ubuntu4.29 OpenSSL/1.0.1f System : Linux b8009 3.13.0-170-generic #220-Ubuntu SMP Thu May 9 12:40:49 UTC 2019 x86_64 User : www-data ( 33) PHP Version : 5.5.9-1ubuntu4.29 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority, MySQL : ON | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : ON Directory : /proc/self/root/home/b8009/Python-3.6.3/Lib/test/ |
Upload File : |
"""
Tests for kqueue wrapper.
"""
import errno
import os
import select
import socket
import time
import unittest
if not hasattr(select, "kqueue"):
raise unittest.SkipTest("test works only on BSD")
class TestKQueue(unittest.TestCase):
def test_create_queue(self):
kq = select.kqueue()
self.assertTrue(kq.fileno() > 0, kq.fileno())
self.assertTrue(not kq.closed)
kq.close()
self.assertTrue(kq.closed)
self.assertRaises(ValueError, kq.fileno)
def test_create_event(self):
from operator import lt, le, gt, ge
fd = os.open(os.devnull, os.O_WRONLY)
self.addCleanup(os.close, fd)
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.assertTrue(ev < other)
self.assertTrue(other >= ev)
for op in lt, le, gt, ge:
self.assertRaises(TypeError, op, ev, None)
self.assertRaises(TypeError, op, ev, 1)
self.assertRaises(TypeError, op, 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)
bignum = 0x7fff
ev = select.kevent(bignum, 1, 2, 3, bignum - 1, bignum)
self.assertEqual(ev.ident, bignum)
self.assertEqual(ev.filter, 1)
self.assertEqual(ev.flags, 2)
self.assertEqual(ev.fflags, 3)
self.assertEqual(ev.data, bignum - 1)
self.assertEqual(ev.udata, bignum)
self.assertEqual(ev, ev)
self.assertNotEqual(ev, other)
# Issue 11973
bignum = 0xffff
ev = select.kevent(0, 1, bignum)
self.assertEqual(ev.ident, 0)
self.assertEqual(ev.filter, 1)
self.assertEqual(ev.flags, bignum)
self.assertEqual(ev.fflags, 0)
self.assertEqual(ev.data, 0)
self.assertEqual(ev.udata, 0)
self.assertEqual(ev, ev)
self.assertNotEqual(ev, other)
# Issue 11973
bignum = 0xffffffff
ev = select.kevent(0, 1, 2, bignum)
self.assertEqual(ev.ident, 0)
self.assertEqual(ev.filter, 1)
self.assertEqual(ev.flags, 2)
self.assertEqual(ev.fflags, bignum)
self.assertEqual(ev.data, 0)
self.assertEqual(ev.udata, 0)
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()
client = socket.socket()
client.setblocking(False)
try:
client.connect(('127.0.0.1', serverSocket.getsockname()[1]))
except OSError as e:
self.assertEqual(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()
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 = set((e.ident, e.filter) for e in events)
self.assertEqual(events, set([
(client.fileno(), select.KQ_FILTER_WRITE),
(server.fileno(), select.KQ_FILTER_WRITE)]))
client.send(b"Hello!")
server.send(b"world!!!")
# We may need to call it several times
for i in range(10):
events = kq.control(None, 4, 1)
if len(events) == 4:
break
time.sleep(1.0)
else:
self.fail('timeout waiting for event notifications')
events = set((e.ident, e.filter) for e in events)
self.assertEqual(events, set([
(client.fileno(), select.KQ_FILTER_WRITE),
(client.fileno(), select.KQ_FILTER_READ),
(server.fileno(), select.KQ_FILTER_WRITE),
(server.fileno(), select.KQ_FILTER_READ)]))
# 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 = set((e.ident, e.filter) for e in events)
self.assertEqual(events, set([
(server.fileno(), select.KQ_FILTER_WRITE)]))
client.close()
server.close()
serverSocket.close()
def testPair(self):
kq = select.kqueue()
a, b = socket.socketpair()
a.send(b'foo')
event1 = select.kevent(a, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE)
event2 = select.kevent(b, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE)
r = kq.control([event1, event2], 1, 1)
self.assertTrue(r)
self.assertFalse(r[0].flags & select.KQ_EV_ERROR)
self.assertEqual(b.recv(r[0].data), b'foo')
a.close()
b.close()
kq.close()
def test_close(self):
open_file = open(__file__, "rb")
self.addCleanup(open_file.close)
fd = open_file.fileno()
kqueue = select.kqueue()
# test fileno() method and closed attribute
self.assertIsInstance(kqueue.fileno(), int)
self.assertFalse(kqueue.closed)
# test close()
kqueue.close()
self.assertTrue(kqueue.closed)
self.assertRaises(ValueError, kqueue.fileno)
# close() can be called more than once
kqueue.close()
# operations must fail with ValueError("I/O operation on closed ...")
self.assertRaises(ValueError, kqueue.control, None, 4)
def test_fd_non_inheritable(self):
kqueue = select.kqueue()
self.addCleanup(kqueue.close)
self.assertEqual(os.get_inheritable(kqueue.fileno()), False)
if __name__ == "__main__":
unittest.main()