import errno import random import selectors import signal import socket from test import support from time import sleep import unittest try: from time import monotonic as time except ImportError: from time import time as time try: import resource except ImportError: resource = None if hasattr(socket, 'socketpair'): socketpair = socket.socketpair else: def socketpair(family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0): with socket.socket(family, type, proto) as l: l.bind((support.HOST, 0)) l.listen(3) c = socket.socket(family, type, proto) try: c.connect(l.getsockname()) caddr = c.getsockname() while True: a, addr = l.accept() # check that we've got the correct client if addr == caddr: return c, a a.close() except OSError: c.close() raise def find_ready_matching(ready, flag): match = [] for key, events in ready: if events & flag: match.append(key.fileobj) return match class BaseSelectorTestCase(unittest.TestCase): def test_register(self): s = self.SELECTOR() self.addCleanup(s.close) rd, wr = socketpair() self.addCleanup(rd.close) self.addCleanup(wr.close) key = s.register(rd, selectors.EVENT_READ, "data") self.assertIsInstance(key, selectors.SelectorKey) self.assertEqual(key.fileobj, rd) self.assertEqual(key.fd, rd.fileno()) self.assertEqual(key.events, selectors.EVENT_READ) self.assertEqual(key.data, "data") # register an unknown event self.assertRaises(ValueError, s.register, 0, 999999) # register an invalid FD self.assertRaises(ValueError, s.register, -10, selectors.EVENT_READ) # register twice self.assertRaises(KeyError, s.register, rd, selectors.EVENT_READ) # register the same FD, but with a different object self.assertRaises(KeyError, s.register, rd.fileno(), selectors.EVENT_READ) def test_unregister(self): s = self.SELECTOR() self.addCleanup(s.close) rd, wr = socketpair() self.addCleanup(rd.close) self.addCleanup(wr.close) s.register(rd, selectors.EVENT_READ) s.unregister(rd) # unregister an unknown file obj self.assertRaises(KeyError, s.unregister, 999999) # unregister twice self.assertRaises(KeyError, s.unregister, rd) def test_modify(self): s = self.SELECTOR() self.addCleanup(s.close) rd, wr = socketpair() self.addCleanup(rd.close) self.addCleanup(wr.close) key = s.register(rd, selectors.EVENT_READ) # modify events key2 = s.modify(rd, selectors.EVENT_WRITE) self.assertNotEqual(key.events, key2.events) self.assertEqual(key2, s.get_key(rd)) s.unregister(rd) # modify data d1 = object() d2 = object() key = s.register(rd, selectors.EVENT_READ, d1) key2 = s.modify(rd, selectors.EVENT_READ, d2) self.assertEqual(key.events, key2.events) self.assertNotEqual(key.data, key2.data) self.assertEqual(key2, s.get_key(rd)) self.assertEqual(key2.data, d2) # modify unknown file obj self.assertRaises(KeyError, s.modify, 999999, selectors.EVENT_READ) def test_close(self): s = self.SELECTOR() self.addCleanup(s.close) rd, wr = socketpair() self.addCleanup(rd.close) self.addCleanup(wr.close) s.register(rd, selectors.EVENT_READ) s.register(wr, selectors.EVENT_WRITE) s.close() self.assertRaises(KeyError, s.get_key, rd) self.assertRaises(KeyError, s.get_key, wr) def test_get_key(self): s = self.SELECTOR() self.addCleanup(s.close) rd, wr = socketpair() self.addCleanup(rd.close) self.addCleanup(wr.close) key = s.register(rd, selectors.EVENT_READ, "data") self.assertEqual(key, s.get_key(rd)) # unknown file obj self.assertRaises(KeyError, s.get_key, 999999) def test_select(self): s = self.SELECTOR() self.addCleanup(s.close) rd, wr = socketpair() self.addCleanup(rd.close) self.addCleanup(wr.close) s.register(rd, selectors.EVENT_READ) wr_key = s.register(wr, selectors.EVENT_WRITE) result = s.select() for key, events in result: self.assertTrue(isinstance(key, selectors.SelectorKey)) self.assertTrue(events) self.assertFalse(events & ~(selectors.EVENT_READ | selectors.EVENT_WRITE)) self.assertEqual([(wr_key, selectors.EVENT_WRITE)], result) def test_context_manager(self): s = self.SELECTOR() self.addCleanup(s.close) rd, wr = socketpair() self.addCleanup(rd.close) self.addCleanup(wr.close) with s as sel: sel.register(rd, selectors.EVENT_READ) sel.register(wr, selectors.EVENT_WRITE) self.assertRaises(KeyError, s.get_key, rd) self.assertRaises(KeyError, s.get_key, wr) def test_fileno(self): s = self.SELECTOR() self.addCleanup(s.close) if hasattr(s, 'fileno'): fd = s.fileno() self.assertTrue(isinstance(fd, int)) self.assertGreaterEqual(fd, 0) def test_selector(self): s = self.SELECTOR() self.addCleanup(s.close) NUM_SOCKETS = 12 MSG = b" This is a test." MSG_LEN = len(MSG) readers = [] writers = [] r2w = {} w2r = {} for i in range(NUM_SOCKETS): rd, wr = socketpair() self.addCleanup(rd.close) self.addCleanup(wr.close) s.register(rd, selectors.EVENT_READ) s.register(wr, selectors.EVENT_WRITE) readers.append(rd) writers.append(wr) r2w[rd] = wr w2r[wr] = rd bufs = [] while writers: ready = s.select() ready_writers = find_ready_matching(ready, selectors.EVENT_WRITE) if not ready_writers: self.fail("no sockets ready for writing") wr = random.choice(ready_writers) wr.send(MSG) for i in range(10): ready = s.select() ready_readers = find_ready_matching(ready, selectors.EVENT_READ) if ready_readers: break # there might be a delay between the write to the write end and # the read end is reported ready sleep(0.1) else: self.fail("no sockets ready for reading") self.assertEqual([w2r[wr]], ready_readers) rd = ready_readers[0] buf = rd.recv(MSG_LEN) self.assertEqual(len(buf), MSG_LEN) bufs.append(buf) s.unregister(r2w[rd]) s.unregister(rd) writers.remove(r2w[rd]) self.assertEqual(bufs, [MSG] * NUM_SOCKETS) def test_timeout(self): s = self.SELECTOR() self.addCleanup(s.close) rd, wr = socketpair() self.addCleanup(rd.close) self.addCleanup(wr.close) s.register(wr, selectors.EVENT_WRITE) t = time() self.assertEqual(1, len(s.select(0))) self.assertEqual(1, len(s.select(-1))) self.assertTrue(time() - t < 0.5) s.unregister(wr) s.register(rd, selectors.EVENT_READ) t = time() self.assertFalse(s.select(0)) self.assertFalse(s.select(-1)) self.assertTrue(time() - t < 0.5) t = time() self.assertFalse(s.select(1)) self.assertTrue(0.5 < time() - t < 1.5) @unittest.skipUnless(hasattr(signal, "alarm"), "signal.alarm() required for this test") def test_interrupted_retry(self): s = self.SELECTOR() self.addCleanup(s.close) rd, wr = socketpair() self.addCleanup(rd.close) self.addCleanup(wr.close) orig_alrm_handler = signal.signal(signal.SIGALRM, lambda *args: None) self.addCleanup(signal.signal, signal.SIGALRM, orig_alrm_handler) self.addCleanup(signal.alarm, 0) signal.alarm(1) s.register(rd, selectors.EVENT_READ) t = time() self.assertFalse(s.select(2)) self.assertLess(time() - t, 2.5) class ScalableSelectorMixIn: # see issue #18963 for why it's skipped on older OS X versions @support.requires_mac_ver(10, 5) @unittest.skipUnless(resource, "Test needs resource module") def test_above_fd_setsize(self): # A scalable implementation should have no problem with more than # FD_SETSIZE file descriptors. Since we don't know the value, we just # try to set the soft RLIMIT_NOFILE to the hard RLIMIT_NOFILE ceiling. soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE) try: resource.setrlimit(resource.RLIMIT_NOFILE, (hard, hard)) self.addCleanup(resource.setrlimit, resource.RLIMIT_NOFILE, (soft, hard)) NUM_FDS = hard except (OSError, ValueError): NUM_FDS = soft # guard for already allocated FDs (stdin, stdout...) NUM_FDS -= 32 s = self.SELECTOR() self.addCleanup(s.close) for i in range(NUM_FDS // 2): try: rd, wr = socketpair() except OSError: # too many FDs, skip - note that we should only catch EMFILE # here, but apparently *BSD and Solaris can fail upon connect() # or bind() with EADDRNOTAVAIL, so let's be safe self.skipTest("FD limit reached") self.addCleanup(rd.close) self.addCleanup(wr.close) try: s.register(rd, selectors.EVENT_READ) s.register(wr, selectors.EVENT_WRITE) except OSError as e: if e.errno == errno.ENOSPC: # this can be raised by epoll if we go over # fs.epoll.max_user_watches sysctl self.skipTest("FD limit reached") raise self.assertEqual(NUM_FDS // 2, len(s.select())) class DefaultSelectorTestCase(BaseSelectorTestCase): SELECTOR = selectors.DefaultSelector class SelectSelectorTestCase(BaseSelectorTestCase): SELECTOR = selectors.SelectSelector @unittest.skipUnless(hasattr(selectors, 'PollSelector'), "Test needs selectors.PollSelector") class PollSelectorTestCase(BaseSelectorTestCase, ScalableSelectorMixIn): SELECTOR = getattr(selectors, 'PollSelector', None) @unittest.skipUnless(hasattr(selectors, 'EpollSelector'), "Test needs selectors.EpollSelector") class EpollSelectorTestCase(BaseSelectorTestCase, ScalableSelectorMixIn): SELECTOR = getattr(selectors, 'EpollSelector', None) @unittest.skipUnless(hasattr(selectors, 'KqueueSelector'), "Test needs selectors.KqueueSelector)") class KqueueSelectorTestCase(BaseSelectorTestCase, ScalableSelectorMixIn): SELECTOR = getattr(selectors, 'KqueueSelector', None) def test_main(): tests = [DefaultSelectorTestCase, SelectSelectorTestCase, PollSelectorTestCase, EpollSelectorTestCase, KqueueSelectorTestCase] support.run_unittest(*tests) support.reap_children() if __name__ == "__main__": test_main()