mirror of
https://github.com/python/cpython.git
synced 2026-10-29 02:45:15 +08:00
Serhiy Storchaka
cdca5021d5
Test setting a value of an accepted type, of a wrong type and an invalid value, and deleting the attribute, for the attributes defined with PyMemberDef and PyGetSetDef which were not covered.
331 lines
12 KiB
Python
331 lines
12 KiB
Python
"""
|
|
Tests for kqueue wrapper.
|
|
"""
|
|
import errno
|
|
import os
|
|
import select
|
|
import socket
|
|
from test import support
|
|
import time
|
|
import unittest
|
|
|
|
from test.support import warnings_helper
|
|
|
|
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_control_overflowing_timeout(self):
|
|
# gh-154836: out-of-range timeouts must raise OverflowError,
|
|
# not a (misleading) TypeError, like select(), poll() and
|
|
# epoll() do.
|
|
kq = select.kqueue()
|
|
self.addCleanup(kq.close)
|
|
for timeout in (1e300, float('inf'), 2**200):
|
|
with self.subTest(timeout=timeout):
|
|
with self.assertRaises(OverflowError):
|
|
kq.control(None, 0, timeout)
|
|
# Non-numbers still raise TypeError.
|
|
with self.assertRaises(TypeError):
|
|
kq.control(None, 0, "0.1")
|
|
|
|
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_event_attributes(self):
|
|
fd = os.open(os.devnull, os.O_WRONLY)
|
|
self.addCleanup(os.close, fd)
|
|
|
|
ev = select.kevent(fd)
|
|
# All attributes are numeric members: they can be set and cannot be
|
|
# deleted.
|
|
for name, value in (('ident', 1), ('filter', select.KQ_FILTER_WRITE),
|
|
('flags', select.KQ_EV_DELETE), ('fflags', 2),
|
|
('data', 3), ('udata', 4)):
|
|
with self.subTest(name=name):
|
|
setattr(ev, name, value)
|
|
self.assertEqual(getattr(ev, name), value)
|
|
with self.assertRaises(TypeError):
|
|
setattr(ev, name, 'not a number')
|
|
with self.assertRaises(OverflowError):
|
|
setattr(ev, name, 2**1000)
|
|
with self.assertRaises(OverflowError):
|
|
setattr(ev, name, -2**1000)
|
|
with self.assertRaisesRegex(
|
|
TypeError, "can't delete numeric/char attribute"):
|
|
delattr(ev, name)
|
|
# a failed assignment does not change the value
|
|
self.assertEqual(getattr(ev, name), value)
|
|
|
|
def test_queue_event(self):
|
|
serverSocket = socket.create_server(('127.0.0.1', 0))
|
|
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_issue30058(self):
|
|
# changelist must be an iterable
|
|
kq = select.kqueue()
|
|
a, b = socket.socketpair()
|
|
ev = select.kevent(a, select.KQ_FILTER_READ, select.KQ_EV_ADD | select.KQ_EV_ENABLE)
|
|
|
|
kq.control([ev], 0)
|
|
# not a list
|
|
kq.control((ev,), 0)
|
|
# __len__ is not consistent with __iter__
|
|
class BadList:
|
|
def __len__(self):
|
|
return 0
|
|
def __iter__(self):
|
|
for i in range(100):
|
|
yield ev
|
|
kq.control(BadList(), 0)
|
|
# doesn't have __len__
|
|
kq.control(iter([ev]), 0)
|
|
|
|
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_close_error(self):
|
|
# gh-146205: close() should raise OSError if underlying fd is invalid
|
|
kqueue = select.kqueue()
|
|
fd = kqueue.fileno()
|
|
os.close(fd)
|
|
with self.assertRaises(OSError) as cm:
|
|
kqueue.close()
|
|
self.assertEqual(cm.exception.errno, errno.EBADF)
|
|
|
|
def test_fd_non_inheritable(self):
|
|
kqueue = select.kqueue()
|
|
self.addCleanup(kqueue.close)
|
|
self.assertEqual(os.get_inheritable(kqueue.fileno()), False)
|
|
|
|
@warnings_helper.ignore_fork_in_thread_deprecation_warnings()
|
|
@support.requires_fork()
|
|
def test_fork(self):
|
|
# gh-110395: kqueue objects must be closed after fork
|
|
kqueue = select.kqueue()
|
|
if (pid := os.fork()) == 0:
|
|
try:
|
|
self.assertTrue(kqueue.closed)
|
|
with self.assertRaisesRegex(ValueError, "closed kqueue"):
|
|
kqueue.fileno()
|
|
except:
|
|
os._exit(1)
|
|
finally:
|
|
os._exit(0)
|
|
else:
|
|
support.wait_process(pid, exitcode=0)
|
|
self.assertFalse(kqueue.closed) # child done, we're still open.
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|