mirror of
https://github.com/python/cpython.git
synced 2025-08-04 17:08:35 +00:00
Close #18794: Add a fileno() method and a closed attribute to select.devpoll
objects. Add also tests on fileno() method and closed attribute of select.epoll and select.kqueue.
This commit is contained in:
parent
43f80e6c1f
commit
13423c3726
6 changed files with 191 additions and 7 deletions
|
@ -87,6 +87,31 @@ class DevPollTests(unittest.TestCase):
|
|||
self.assertRaises(OverflowError, pollster.poll, 1 << 63)
|
||||
self.assertRaises(OverflowError, pollster.poll, 1 << 64)
|
||||
|
||||
def test_close(self):
|
||||
open_file = open(__file__, "rb")
|
||||
self.addCleanup(open_file.close)
|
||||
fd = open_file.fileno()
|
||||
devpoll = select.devpoll()
|
||||
|
||||
# test fileno() method and closed attribute
|
||||
self.assertIsInstance(devpoll.fileno(), int)
|
||||
self.assertFalse(devpoll.closed)
|
||||
|
||||
# test close()
|
||||
devpoll.close()
|
||||
self.assertTrue(devpoll.closed)
|
||||
self.assertRaises(ValueError, devpoll.fileno)
|
||||
|
||||
# close() can be called more than once
|
||||
devpoll.close()
|
||||
|
||||
# operations must fail with ValueError("I/O operation on closed ...")
|
||||
self.assertRaises(ValueError, devpoll.modify, fd, select.POLLIN)
|
||||
self.assertRaises(ValueError, devpoll.poll)
|
||||
self.assertRaises(ValueError, devpoll.register, fd, fd, select.POLLIN)
|
||||
self.assertRaises(ValueError, devpoll.unregister, fd)
|
||||
|
||||
|
||||
def test_main():
|
||||
run_unittest(DevPollTests)
|
||||
|
||||
|
|
|
@ -225,6 +225,31 @@ class TestEPoll(unittest.TestCase):
|
|||
server.close()
|
||||
ep.unregister(fd)
|
||||
|
||||
def test_close(self):
|
||||
open_file = open(__file__, "rb")
|
||||
self.addCleanup(open_file.close)
|
||||
fd = open_file.fileno()
|
||||
epoll = select.epoll()
|
||||
|
||||
# test fileno() method and closed attribute
|
||||
self.assertIsInstance(epoll.fileno(), int)
|
||||
self.assertFalse(epoll.closed)
|
||||
|
||||
# test close()
|
||||
epoll.close()
|
||||
self.assertTrue(epoll.closed)
|
||||
self.assertRaises(ValueError, epoll.fileno)
|
||||
|
||||
# close() can be called more than once
|
||||
epoll.close()
|
||||
|
||||
# operations must fail with ValueError("I/O operation on closed ...")
|
||||
self.assertRaises(ValueError, epoll.modify, fd, select.EPOLLIN)
|
||||
self.assertRaises(ValueError, epoll.poll, 1.0)
|
||||
self.assertRaises(ValueError, epoll.register, fd, select.EPOLLIN)
|
||||
self.assertRaises(ValueError, epoll.unregister, fd)
|
||||
|
||||
|
||||
def test_main():
|
||||
support.run_unittest(TestEPoll)
|
||||
|
||||
|
|
|
@ -185,6 +185,28 @@ class TestKQueue(unittest.TestCase):
|
|||
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_main():
|
||||
support.run_unittest(TestKQueue)
|
||||
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue