<s> import errno <EOL> import os <EOL> import random <EOL> import selectors <EOL> import signal <EOL> import socket <EOL> import sys <EOL> from test import support <EOL> from time import sleep <EOL> import unittest <EOL> import unittest . mock <EOL> try : <EOL> from time import monotonic as time <EOL> except ImportError : <EOL> from time import time as time <EOL> try : <EOL> import resource <EOL> except ImportError : <EOL> resource = None <EOL> if hasattr ( socket , '<STR_LIT>' ) : <EOL> socketpair = socket . socketpair <EOL> else : <EOL> def socketpair ( family = socket . AF_INET , type = socket . SOCK_STREAM , proto = <NUM_LIT:0> ) : <EOL> with socket . socket ( family , type , proto ) as l : <EOL> l . bind ( ( support . HOST , <NUM_LIT:0> ) ) <EOL> l . listen ( <NUM_LIT:3> ) <EOL> c = socket . socket ( family , type , proto ) <EOL> try : <EOL> c . connect ( l . getsockname ( ) ) <EOL> caddr = c . getsockname ( ) <EOL> while True : <EOL> a , addr = l . accept ( ) <EOL> if addr == caddr : <EOL> return c , a <EOL> a . close ( ) <EOL> except OSError : <EOL> c . close ( ) <EOL> raise <EOL> def find_ready_matching ( ready , flag ) : <EOL> match = [ ] <EOL> for key , events in ready : <EOL> if events & flag : <EOL> match . append ( key . fileobj ) <EOL> return match <EOL> class BaseSelectorTestCase ( unittest . TestCase ) : <EOL> def make_socketpair ( self ) : <EOL> rd , wr = socketpair ( ) <EOL> self . addCleanup ( rd . close ) <EOL> self . addCleanup ( wr . close ) <EOL> return rd , wr <EOL> def test_register ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> rd , wr = self . make_socketpair ( ) <EOL> key = s . register ( rd , selectors . EVENT_READ , "<STR_LIT:data>" ) <EOL> self . assertIsInstance ( key , selectors . SelectorKey ) <EOL> self . assertEqual ( key . fileobj , rd ) <EOL> self . assertEqual ( key . fd , rd . fileno ( ) ) <EOL> self . assertEqual ( key . events , selectors . EVENT_READ ) <EOL> self . assertEqual ( key . data , "<STR_LIT:data>" ) <EOL> self . assertRaises ( ValueError , s . register , <NUM_LIT:0> , <NUM_LIT> ) <EOL> self . assertRaises ( ValueError , s . register , - <NUM_LIT:10> , selectors . EVENT_READ ) <EOL> self . assertRaises ( KeyError , s . register , rd , selectors . EVENT_READ ) <EOL> self . assertRaises ( KeyError , s . register , rd . fileno ( ) , <EOL> selectors . EVENT_READ ) <EOL> def test_unregister ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> rd , wr = self . make_socketpair ( ) <EOL> s . register ( rd , selectors . EVENT_READ ) <EOL> s . unregister ( rd ) <EOL> self . assertRaises ( KeyError , s . unregister , <NUM_LIT> ) <EOL> self . assertRaises ( KeyError , s . unregister , rd ) <EOL> def test_unregister_after_fd_close ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> rd , wr = self . make_socketpair ( ) <EOL> r , w = rd . fileno ( ) , wr . fileno ( ) <EOL> s . register ( r , selectors . EVENT_READ ) <EOL> s . register ( w , selectors . EVENT_WRITE ) <EOL> rd . close ( ) <EOL> wr . close ( ) <EOL> s . unregister ( r ) <EOL> s . unregister ( w ) <EOL> @ unittest . skipUnless ( os . name == '<STR_LIT>' , "<STR_LIT>" ) <EOL> def test_unregister_after_fd_close_and_reuse ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> rd , wr = self . make_socketpair ( ) <EOL> r , w = rd . fileno ( ) , wr . fileno ( ) <EOL> s . register ( r , selectors . EVENT_READ ) <EOL> s . register ( w , selectors . EVENT_WRITE ) <EOL> rd2 , wr2 = self . make_socketpair ( ) <EOL> rd . close ( ) <EOL> wr . close ( ) <EOL> os . dup2 ( rd2 . fileno ( ) , r ) <EOL> os . dup2 ( wr2 . fileno ( ) , w ) <EOL> self . addCleanup ( os . close , r ) <EOL> self . addCleanup ( os . close , w ) <EOL> s . unregister ( r ) <EOL> s . unregister ( w ) <EOL> def test_unregister_after_socket_close ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> rd , wr = self . make_socketpair ( ) <EOL> s . register ( rd , selectors . EVENT_READ ) <EOL> s . register ( wr , selectors . EVENT_WRITE ) <EOL> rd . close ( ) <EOL> wr . close ( ) <EOL> s . unregister ( rd ) <EOL> s . unregister ( wr ) <EOL> def test_modify ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> rd , wr = self . make_socketpair ( ) <EOL> key = s . register ( rd , selectors . EVENT_READ ) <EOL> key2 = s . modify ( rd , selectors . EVENT_WRITE ) <EOL> self . assertNotEqual ( key . events , key2 . events ) <EOL> self . assertEqual ( key2 , s . get_key ( rd ) ) <EOL> s . unregister ( rd ) <EOL> d1 = object ( ) <EOL> d2 = object ( ) <EOL> key = s . register ( rd , selectors . EVENT_READ , d1 ) <EOL> key2 = s . modify ( rd , selectors . EVENT_READ , d2 ) <EOL> self . assertEqual ( key . events , key2 . events ) <EOL> self . assertNotEqual ( key . data , key2 . data ) <EOL> self . assertEqual ( key2 , s . get_key ( rd ) ) <EOL> self . assertEqual ( key2 . data , d2 ) <EOL> self . assertRaises ( KeyError , s . modify , <NUM_LIT> , selectors . EVENT_READ ) <EOL> d3 = object ( ) <EOL> s . register = unittest . mock . Mock ( ) <EOL> s . unregister = unittest . mock . Mock ( ) <EOL> s . modify ( rd , selectors . EVENT_READ , d3 ) <EOL> self . assertFalse ( s . register . called ) <EOL> self . assertFalse ( s . unregister . called ) <EOL> def test_close ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> mapping = s . get_map ( ) <EOL> rd , wr = self . make_socketpair ( ) <EOL> s . register ( rd , selectors . EVENT_READ ) <EOL> s . register ( wr , selectors . EVENT_WRITE ) <EOL> s . close ( ) <EOL> self . assertRaises ( KeyError , s . get_key , rd ) <EOL> self . assertRaises ( KeyError , s . get_key , wr ) <EOL> self . assertRaises ( KeyError , mapping . __getitem__ , rd ) <EOL> self . assertRaises ( KeyError , mapping . __getitem__ , wr ) <EOL> def test_get_key ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> rd , wr = self . make_socketpair ( ) <EOL> key = s . register ( rd , selectors . EVENT_READ , "<STR_LIT:data>" ) <EOL> self . assertEqual ( key , s . get_key ( rd ) ) <EOL> self . assertRaises ( KeyError , s . get_key , <NUM_LIT> ) <EOL> def test_get_map ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> rd , wr = self . make_socketpair ( ) <EOL> keys = s . get_map ( ) <EOL> self . assertFalse ( keys ) <EOL> self . assertEqual ( len ( keys ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( list ( keys ) , [ ] ) <EOL> key = s . register ( rd , selectors . EVENT_READ , "<STR_LIT:data>" ) <EOL> self . assertIn ( rd , keys ) <EOL> self . assertEqual ( key , keys [ rd ] ) <EOL> self . assertEqual ( len ( keys ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( list ( keys ) , [ rd . fileno ( ) ] ) <EOL> self . assertEqual ( list ( keys . values ( ) ) , [ key ] ) <EOL> with self . assertRaises ( KeyError ) : <EOL> keys [ <NUM_LIT> ] <EOL> with self . assertRaises ( TypeError ) : <EOL> del keys [ rd ] <EOL> def test_select ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> rd , wr = self . make_socketpair ( ) <EOL> s . register ( rd , selectors . EVENT_READ ) <EOL> wr_key = s . register ( wr , selectors . EVENT_WRITE ) <EOL> result = s . select ( ) <EOL> for key , events in result : <EOL> self . assertTrue ( isinstance ( key , selectors . SelectorKey ) ) <EOL> self . assertTrue ( events ) <EOL> self . assertFalse ( events & ~ ( selectors . EVENT_READ | <EOL> selectors . EVENT_WRITE ) ) <EOL> self . assertEqual ( [ ( wr_key , selectors . EVENT_WRITE ) ] , result ) <EOL> def test_context_manager ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> rd , wr = self . make_socketpair ( ) <EOL> with s as sel : <EOL> sel . register ( rd , selectors . EVENT_READ ) <EOL> sel . register ( wr , selectors . EVENT_WRITE ) <EOL> self . assertRaises ( KeyError , s . get_key , rd ) <EOL> self . assertRaises ( KeyError , s . get_key , wr ) <EOL> def test_fileno ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> if hasattr ( s , '<STR_LIT>' ) : <EOL> fd = s . fileno ( ) <EOL> self . assertTrue ( isinstance ( fd , int ) ) <EOL> self . assertGreaterEqual ( fd , <NUM_LIT:0> ) <EOL> def test_selector ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> NUM_SOCKETS = <NUM_LIT:12> <EOL> MSG = b"<STR_LIT>" <EOL> MSG_LEN = len ( MSG ) <EOL> readers = [ ] <EOL> writers = [ ] <EOL> r2w = { } <EOL> w2r = { } <EOL> for i in range ( NUM_SOCKETS ) : <EOL> rd , wr = self . make_socketpair ( ) <EOL> s . register ( rd , selectors . EVENT_READ ) <EOL> s . register ( wr , selectors . EVENT_WRITE ) <EOL> readers . append ( rd ) <EOL> writers . append ( wr ) <EOL> r2w [ rd ] = wr <EOL> w2r [ wr ] = rd <EOL> bufs = [ ] <EOL> while writers : <EOL> ready = s . select ( ) <EOL> ready_writers = find_ready_matching ( ready , selectors . EVENT_WRITE ) <EOL> if not ready_writers : <EOL> self . fail ( "<STR_LIT>" ) <EOL> wr = random . choice ( ready_writers ) <EOL> wr . send ( MSG ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> ready = s . select ( ) <EOL> ready_readers = find_ready_matching ( ready , <EOL> selectors . EVENT_READ ) <EOL> if ready_readers : <EOL> break <EOL> sleep ( <NUM_LIT:0.1> ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> self . assertEqual ( [ w2r [ wr ] ] , ready_readers ) <EOL> rd = ready_readers [ <NUM_LIT:0> ] <EOL> buf = rd . recv ( MSG_LEN ) <EOL> self . assertEqual ( len ( buf ) , MSG_LEN ) <EOL> bufs . append ( buf ) <EOL> s . unregister ( r2w [ rd ] ) <EOL> s . unregister ( rd ) <EOL> writers . remove ( r2w [ rd ] ) <EOL> self . assertEqual ( bufs , [ MSG ] * NUM_SOCKETS ) <EOL> @ unittest . skipIf ( sys . platform == '<STR_LIT:win32>' , <EOL> '<STR_LIT>' ) <EOL> def test_empty_select ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> self . assertEqual ( s . select ( timeout = <NUM_LIT:0> ) , [ ] ) <EOL> def test_timeout ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> rd , wr = self . make_socketpair ( ) <EOL> s . register ( wr , selectors . EVENT_WRITE ) <EOL> t = time ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( s . select ( <NUM_LIT:0> ) ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( s . select ( - <NUM_LIT:1> ) ) ) <EOL> self . assertLess ( time ( ) - t , <NUM_LIT:0.5> ) <EOL> s . unregister ( wr ) <EOL> s . register ( rd , selectors . EVENT_READ ) <EOL> t = time ( ) <EOL> self . assertFalse ( s . select ( <NUM_LIT:0> ) ) <EOL> self . assertFalse ( s . select ( - <NUM_LIT:1> ) ) <EOL> self . assertLess ( time ( ) - t , <NUM_LIT:0.5> ) <EOL> t0 = time ( ) <EOL> self . assertFalse ( s . select ( <NUM_LIT:1> ) ) <EOL> t1 = time ( ) <EOL> dt = t1 - t0 <EOL> self . assertTrue ( <NUM_LIT> <= dt <= <NUM_LIT> , dt ) <EOL> @ unittest . skipUnless ( hasattr ( signal , "<STR_LIT>" ) , <EOL> "<STR_LIT>" ) <EOL> def test_select_interrupt ( self ) : <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> rd , wr = self . make_socketpair ( ) <EOL> orig_alrm_handler = signal . signal ( signal . SIGALRM , lambda * args : None ) <EOL> self . addCleanup ( signal . signal , signal . SIGALRM , orig_alrm_handler ) <EOL> self . addCleanup ( signal . alarm , <NUM_LIT:0> ) <EOL> signal . alarm ( <NUM_LIT:1> ) <EOL> s . register ( rd , selectors . EVENT_READ ) <EOL> t = time ( ) <EOL> self . assertFalse ( s . select ( <NUM_LIT:2> ) ) <EOL> self . assertLess ( time ( ) - t , <NUM_LIT> ) <EOL> class ScalableSelectorMixIn : <EOL> @ support . requires_mac_ver ( <NUM_LIT:10> , <NUM_LIT:5> ) <EOL> @ unittest . skipUnless ( resource , "<STR_LIT>" ) <EOL> def test_above_fd_setsize ( self ) : <EOL> soft , hard = resource . getrlimit ( resource . RLIMIT_NOFILE ) <EOL> try : <EOL> resource . setrlimit ( resource . RLIMIT_NOFILE , ( hard , hard ) ) <EOL> self . addCleanup ( resource . setrlimit , resource . RLIMIT_NOFILE , <EOL> ( soft , hard ) ) <EOL> NUM_FDS = min ( hard , <NUM_LIT:2> ** <NUM_LIT:16> ) <EOL> except ( OSError , ValueError ) : <EOL> NUM_FDS = soft <EOL> NUM_FDS -= <NUM_LIT:32> <EOL> s = self . SELECTOR ( ) <EOL> self . addCleanup ( s . close ) <EOL> for i in range ( NUM_FDS // <NUM_LIT:2> ) : <EOL> try : <EOL> rd , wr = self . make_socketpair ( ) <EOL> except OSError : <EOL> self . skipTest ( "<STR_LIT>" ) <EOL> try : <EOL> s . register ( rd , selectors . EVENT_READ ) <EOL> s . register ( wr , selectors . EVENT_WRITE ) <EOL> except OSError as e : <EOL> if e . errno == errno . ENOSPC : <EOL> self . skipTest ( "<STR_LIT>" ) <EOL> raise <EOL> self . assertEqual ( NUM_FDS // <NUM_LIT:2> , len ( s . select ( ) ) ) <EOL> class DefaultSelectorTestCase ( BaseSelectorTestCase ) : <EOL> SELECTOR = selectors . DefaultSelector <EOL> class SelectSelectorTestCase ( BaseSelectorTestCase ) : <EOL> SELECTOR = selectors . SelectSelector <EOL> @ unittest . skipUnless ( hasattr ( selectors , '<STR_LIT>' ) , <EOL> "<STR_LIT>" ) <EOL> class PollSelectorTestCase ( BaseSelectorTestCase , ScalableSelectorMixIn ) : <EOL> SELECTOR = getattr ( selectors , '<STR_LIT>' , None ) <EOL> @ unittest . skipUnless ( hasattr ( selectors , '<STR_LIT>' ) , <EOL> "<STR_LIT>" ) <EOL> class EpollSelectorTestCase ( BaseSelectorTestCase , ScalableSelectorMixIn ) : <EOL> SELECTOR = getattr ( selectors , '<STR_LIT>' , None ) <EOL> @ unittest . skipUnless ( hasattr ( selectors , '<STR_LIT>' ) , <EOL> "<STR_LIT>" ) <EOL> class KqueueSelectorTestCase ( BaseSelectorTestCase , ScalableSelectorMixIn ) : <EOL> SELECTOR = getattr ( selectors , '<STR_LIT>' , None ) <EOL> def test_main ( ) : <EOL> tests = [ DefaultSelectorTestCase , SelectSelectorTestCase , <EOL> PollSelectorTestCase , EpollSelectorTestCase , <EOL> KqueueSelectorTestCase ] <EOL> support . run_unittest ( * tests ) <EOL> support . reap_children ( ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> test_main ( ) </s>
