<s> import asyncore <EOL> import unittest <EOL> import select <EOL> import os <EOL> import socket <EOL> import sys <EOL> import time <EOL> import warnings <EOL> import errno <EOL> import struct <EOL> from test import support <EOL> from test . support import TESTFN , run_unittest , unlink , HOST , HOSTv6 <EOL> from io import BytesIO <EOL> from io import StringIO <EOL> try : <EOL> import threading <EOL> except ImportError : <EOL> threading = None <EOL> HAS_UNIX_SOCKETS = hasattr ( socket , '<STR_LIT>' ) <EOL> class dummysocket : <EOL> def __init__ ( self ) : <EOL> self . closed = False <EOL> def close ( self ) : <EOL> self . closed = True <EOL> def fileno ( self ) : <EOL> return <NUM_LIT> <EOL> class dummychannel : <EOL> def __init__ ( self ) : <EOL> self . socket = dummysocket ( ) <EOL> def close ( self ) : <EOL> self . socket . close ( ) <EOL> class exitingdummy : <EOL> def __init__ ( self ) : <EOL> pass <EOL> def handle_read_event ( self ) : <EOL> raise asyncore . ExitNow ( ) <EOL> handle_write_event = handle_read_event <EOL> handle_close = handle_read_event <EOL> handle_expt_event = handle_read_event <EOL> class crashingdummy : <EOL> def __init__ ( self ) : <EOL> self . error_handled = False <EOL> def handle_read_event ( self ) : <EOL> raise Exception ( ) <EOL> handle_write_event = handle_read_event <EOL> handle_close = handle_read_event <EOL> handle_expt_event = handle_read_event <EOL> def handle_error ( self ) : <EOL> self . error_handled = True <EOL> def capture_server ( evt , buf , serv ) : <EOL> try : <EOL> serv . listen ( <NUM_LIT:5> ) <EOL> conn , addr = serv . accept ( ) <EOL> except socket . timeout : <EOL> pass <EOL> else : <EOL> n = <NUM_LIT:200> <EOL> start = time . time ( ) <EOL> while n > <NUM_LIT:0> and time . time ( ) - start < <NUM_LIT> : <EOL> r , w , e = select . select ( [ conn ] , [ ] , [ ] , <NUM_LIT:0.1> ) <EOL> if r : <EOL> n -= <NUM_LIT:1> <EOL> data = conn . recv ( <NUM_LIT:10> ) <EOL> buf . write ( data . replace ( b'<STR_LIT:\n>' , b'<STR_LIT>' ) ) <EOL> if b'<STR_LIT:\n>' in data : <EOL> break <EOL> time . sleep ( <NUM_LIT> ) <EOL> conn . close ( ) <EOL> finally : <EOL> serv . close ( ) <EOL> evt . set ( ) <EOL> def bind_af_aware ( sock , addr ) : <EOL> """<STR_LIT>""" <EOL> if HAS_UNIX_SOCKETS and sock . family == socket . AF_UNIX : <EOL> unlink ( addr ) <EOL> sock . bind ( addr ) <EOL> class HelperFunctionTests ( unittest . TestCase ) : <EOL> def test_readwriteexc ( self ) : <EOL> tr1 = exitingdummy ( ) <EOL> self . assertRaises ( asyncore . ExitNow , asyncore . read , tr1 ) <EOL> self . assertRaises ( asyncore . ExitNow , asyncore . write , tr1 ) <EOL> self . assertRaises ( asyncore . ExitNow , asyncore . _exception , tr1 ) <EOL> tr2 = crashingdummy ( ) <EOL> asyncore . read ( tr2 ) <EOL> self . assertEqual ( tr2 . error_handled , True ) <EOL> tr2 = crashingdummy ( ) <EOL> asyncore . write ( tr2 ) <EOL> self . assertEqual ( tr2 . error_handled , True ) <EOL> tr2 = crashingdummy ( ) <EOL> asyncore . _exception ( tr2 ) <EOL> self . assertEqual ( tr2 . error_handled , True ) <EOL> @ unittest . skipUnless ( hasattr ( select , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> def test_readwrite ( self ) : <EOL> attributes = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> expected = ( <EOL> ( select . POLLIN , '<STR_LIT>' ) , <EOL> ( select . POLLPRI , '<STR_LIT>' ) , <EOL> ( select . POLLOUT , '<STR_LIT>' ) , <EOL> ( select . POLLERR , '<STR_LIT>' ) , <EOL> ( select . POLLHUP , '<STR_LIT>' ) , <EOL> ( select . POLLNVAL , '<STR_LIT>' ) , <EOL> ) <EOL> class testobj : <EOL> def __init__ ( self ) : <EOL> self . read = False <EOL> self . write = False <EOL> self . closed = False <EOL> self . expt = False <EOL> self . error_handled = False <EOL> def handle_read_event ( self ) : <EOL> self . read = True <EOL> def handle_write_event ( self ) : <EOL> self . write = True <EOL> def handle_close ( self ) : <EOL> self . closed = True <EOL> def handle_expt_event ( self ) : <EOL> self . expt = True <EOL> def handle_error ( self ) : <EOL> self . error_handled = True <EOL> for flag , expectedattr in expected : <EOL> tobj = testobj ( ) <EOL> self . assertEqual ( getattr ( tobj , expectedattr ) , False ) <EOL> asyncore . readwrite ( tobj , flag ) <EOL> for attr in attributes : <EOL> self . assertEqual ( getattr ( tobj , attr ) , attr == expectedattr ) <EOL> tr1 = exitingdummy ( ) <EOL> self . assertRaises ( asyncore . ExitNow , asyncore . readwrite , tr1 , flag ) <EOL> tr2 = crashingdummy ( ) <EOL> self . assertEqual ( tr2 . error_handled , False ) <EOL> asyncore . readwrite ( tr2 , flag ) <EOL> self . assertEqual ( tr2 . error_handled , True ) <EOL> def test_closeall ( self ) : <EOL> self . closeall_check ( False ) <EOL> def test_closeall_default ( self ) : <EOL> self . closeall_check ( True ) <EOL> def closeall_check ( self , usedefault ) : <EOL> l = [ ] <EOL> testmap = { } <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> c = dummychannel ( ) <EOL> l . append ( c ) <EOL> self . assertEqual ( c . socket . closed , False ) <EOL> testmap [ i ] = c <EOL> if usedefault : <EOL> socketmap = asyncore . socket_map <EOL> try : <EOL> asyncore . socket_map = testmap <EOL> asyncore . close_all ( ) <EOL> finally : <EOL> testmap , asyncore . socket_map = asyncore . socket_map , socketmap <EOL> else : <EOL> asyncore . close_all ( testmap ) <EOL> self . assertEqual ( len ( testmap ) , <NUM_LIT:0> ) <EOL> for c in l : <EOL> self . assertEqual ( c . socket . closed , True ) <EOL> def test_compact_traceback ( self ) : <EOL> try : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> except : <EOL> real_t , real_v , real_tb = sys . exc_info ( ) <EOL> r = asyncore . compact_traceback ( ) <EOL> else : <EOL> self . fail ( "<STR_LIT>" ) <EOL> ( f , function , line ) , t , v , info = r <EOL> self . assertEqual ( os . path . split ( f ) [ - <NUM_LIT:1> ] , '<STR_LIT>' ) <EOL> self . assertEqual ( function , '<STR_LIT>' ) <EOL> self . assertEqual ( t , real_t ) <EOL> self . assertEqual ( v , real_v ) <EOL> self . assertEqual ( info , '<STR_LIT>' % ( f , function , line ) ) <EOL> class DispatcherTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> pass <EOL> def tearDown ( self ) : <EOL> asyncore . close_all ( ) <EOL> def test_basic ( self ) : <EOL> d = asyncore . dispatcher ( ) <EOL> self . assertEqual ( d . readable ( ) , True ) <EOL> self . assertEqual ( d . writable ( ) , True ) <EOL> def test_repr ( self ) : <EOL> d = asyncore . dispatcher ( ) <EOL> self . assertEqual ( repr ( d ) , '<STR_LIT>' % id ( d ) ) <EOL> def test_log ( self ) : <EOL> d = asyncore . dispatcher ( ) <EOL> fp = StringIO ( ) <EOL> stderr = sys . stderr <EOL> l1 = "<STR_LIT>" <EOL> l2 = "<STR_LIT>" <EOL> try : <EOL> sys . stderr = fp <EOL> d . log ( l1 ) <EOL> d . log ( l2 ) <EOL> finally : <EOL> sys . stderr = stderr <EOL> lines = fp . getvalue ( ) . splitlines ( ) <EOL> self . assertEqual ( lines , [ '<STR_LIT>' % l1 , '<STR_LIT>' % l2 ] ) <EOL> def test_log_info ( self ) : <EOL> d = asyncore . dispatcher ( ) <EOL> fp = StringIO ( ) <EOL> stdout = sys . stdout <EOL> l1 = "<STR_LIT>" <EOL> l2 = "<STR_LIT>" <EOL> l3 = "<STR_LIT>" <EOL> try : <EOL> sys . stdout = fp <EOL> d . log_info ( l1 , '<STR_LIT>' ) <EOL> d . log_info ( l2 ) <EOL> d . log_info ( l3 , '<STR_LIT>' ) <EOL> finally : <EOL> sys . stdout = stdout <EOL> lines = fp . getvalue ( ) . splitlines ( ) <EOL> expected = [ '<STR_LIT>' % l1 , '<STR_LIT>' % l2 , '<STR_LIT>' % l3 ] <EOL> self . assertEqual ( lines , expected ) <EOL> def test_unhandled ( self ) : <EOL> d = asyncore . dispatcher ( ) <EOL> d . ignore_log_types = ( ) <EOL> fp = StringIO ( ) <EOL> stdout = sys . stdout <EOL> try : <EOL> sys . stdout = fp <EOL> d . handle_expt ( ) <EOL> d . handle_read ( ) <EOL> d . handle_write ( ) <EOL> d . handle_connect ( ) <EOL> finally : <EOL> sys . stdout = stdout <EOL> lines = fp . getvalue ( ) . splitlines ( ) <EOL> expected = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> self . assertEqual ( lines , expected ) <EOL> def test_issue_8594 ( self ) : <EOL> d = asyncore . dispatcher ( socket . socket ( ) ) <EOL> self . assertRaisesRegex ( AttributeError , '<STR_LIT>' , <EOL> getattr , d , '<STR_LIT:foo>' ) <EOL> with warnings . catch_warnings ( record = True ) as w : <EOL> warnings . simplefilter ( "<STR_LIT>" ) <EOL> family = d . family <EOL> self . assertEqual ( family , socket . AF_INET ) <EOL> self . assertEqual ( len ( w ) , <NUM_LIT:1> ) <EOL> self . assertTrue ( issubclass ( w [ <NUM_LIT:0> ] . category , DeprecationWarning ) ) <EOL> def test_strerror ( self ) : <EOL> err = asyncore . _strerror ( errno . EPERM ) <EOL> if hasattr ( os , '<STR_LIT>' ) : <EOL> self . assertEqual ( err , os . strerror ( errno . EPERM ) ) <EOL> err = asyncore . _strerror ( - <NUM_LIT:1> ) <EOL> self . assertTrue ( err != "<STR_LIT>" ) <EOL> class dispatcherwithsend_noread ( asyncore . dispatcher_with_send ) : <EOL> def readable ( self ) : <EOL> return False <EOL> def handle_connect ( self ) : <EOL> pass <EOL> class DispatcherWithSendTests ( unittest . TestCase ) : <EOL> usepoll = False <EOL> def setUp ( self ) : <EOL> pass <EOL> def tearDown ( self ) : <EOL> asyncore . close_all ( ) <EOL> @ unittest . skipUnless ( threading , '<STR_LIT>' ) <EOL> @ support . reap_threads <EOL> def test_send ( self ) : <EOL> evt = threading . Event ( ) <EOL> sock = socket . socket ( ) <EOL> sock . settimeout ( <NUM_LIT:3> ) <EOL> port = support . bind_port ( sock ) <EOL> cap = BytesIO ( ) <EOL> args = ( evt , cap , sock ) <EOL> t = threading . Thread ( target = capture_server , args = args ) <EOL> t . start ( ) <EOL> try : <EOL> time . sleep ( <NUM_LIT> ) <EOL> data = b"<STR_LIT>" <EOL> d = dispatcherwithsend_noread ( ) <EOL> d . create_socket ( ) <EOL> d . connect ( ( HOST , port ) ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> d . send ( data ) <EOL> d . send ( data ) <EOL> d . send ( b'<STR_LIT:\n>' ) <EOL> n = <NUM_LIT:1000> <EOL> while d . out_buffer and n > <NUM_LIT:0> : <EOL> asyncore . poll ( ) <EOL> n -= <NUM_LIT:1> <EOL> evt . wait ( ) <EOL> self . assertEqual ( cap . getvalue ( ) , data * <NUM_LIT:2> ) <EOL> finally : <EOL> t . join ( ) <EOL> class DispatcherWithSendTests_UsePoll ( DispatcherWithSendTests ) : <EOL> usepoll = True <EOL> @ unittest . skipUnless ( hasattr ( asyncore , '<STR_LIT>' ) , <EOL> '<STR_LIT>' ) <EOL> class FileWrapperTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . d = b"<STR_LIT>" <EOL> with open ( TESTFN , '<STR_LIT:wb>' ) as file : <EOL> file . write ( self . d ) <EOL> def tearDown ( self ) : <EOL> unlink ( TESTFN ) <EOL> def test_recv ( self ) : <EOL> fd = os . open ( TESTFN , os . O_RDONLY ) <EOL> w = asyncore . file_wrapper ( fd ) <EOL> os . close ( fd ) <EOL> self . assertNotEqual ( w . fd , fd ) <EOL> self . assertNotEqual ( w . fileno ( ) , fd ) <EOL> self . assertEqual ( w . recv ( <NUM_LIT> ) , b"<STR_LIT>" ) <EOL> self . assertEqual ( w . read ( <NUM_LIT:6> ) , b"<STR_LIT>" ) <EOL> w . close ( ) <EOL> self . assertRaises ( OSError , w . read , <NUM_LIT:1> ) <EOL> def test_send ( self ) : <EOL> d1 = b"<STR_LIT>" <EOL> d2 = b"<STR_LIT>" <EOL> fd = os . open ( TESTFN , os . O_WRONLY | os . O_APPEND ) <EOL> w = asyncore . file_wrapper ( fd ) <EOL> os . close ( fd ) <EOL> w . write ( d1 ) <EOL> w . send ( d2 ) <EOL> w . close ( ) <EOL> with open ( TESTFN , '<STR_LIT:rb>' ) as file : <EOL> self . assertEqual ( file . read ( ) , self . d + d1 + d2 ) <EOL> @ unittest . skipUnless ( hasattr ( asyncore , '<STR_LIT>' ) , <EOL> '<STR_LIT>' ) <EOL> def test_dispatcher ( self ) : <EOL> fd = os . open ( TESTFN , os . O_RDONLY ) <EOL> data = [ ] <EOL> class FileDispatcher ( asyncore . file_dispatcher ) : <EOL> def handle_read ( self ) : <EOL> data . append ( self . recv ( <NUM_LIT> ) ) <EOL> s = FileDispatcher ( fd ) <EOL> os . close ( fd ) <EOL> asyncore . loop ( timeout = <NUM_LIT> , use_poll = True , count = <NUM_LIT:2> ) <EOL> self . assertEqual ( b"<STR_LIT>" . join ( data ) , self . d ) <EOL> class BaseTestHandler ( asyncore . dispatcher ) : <EOL> def __init__ ( self , sock = None ) : <EOL> asyncore . dispatcher . __init__ ( self , sock ) <EOL> self . flag = False <EOL> def handle_accept ( self ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> def handle_accepted ( self ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> def handle_connect ( self ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> def handle_expt ( self ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> def handle_close ( self ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> def handle_error ( self ) : <EOL> raise <EOL> class BaseServer ( asyncore . dispatcher ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , family , addr , handler = BaseTestHandler ) : <EOL> asyncore . dispatcher . __init__ ( self ) <EOL> self . create_socket ( family ) <EOL> self . set_reuse_addr ( ) <EOL> bind_af_aware ( self . socket , addr ) <EOL> self . listen ( <NUM_LIT:5> ) <EOL> self . handler = handler <EOL> @ property <EOL> def address ( self ) : <EOL> return self . socket . getsockname ( ) <EOL> def handle_accepted ( self , sock , addr ) : <EOL> self . handler ( sock ) <EOL> def handle_error ( self ) : <EOL> raise <EOL> class BaseClient ( BaseTestHandler ) : <EOL> def __init__ ( self , family , address ) : <EOL> BaseTestHandler . __init__ ( self ) <EOL> self . create_socket ( family ) <EOL> self . connect ( address ) <EOL> def handle_connect ( self ) : <EOL> pass <EOL> class BaseTestAPI : <EOL> def tearDown ( self ) : <EOL> asyncore . close_all ( ) <EOL> def loop_waiting_for_flag ( self , instance , timeout = <NUM_LIT:5> ) : <EOL> timeout = float ( timeout ) / <NUM_LIT:100> <EOL> count = <NUM_LIT:100> <EOL> while asyncore . socket_map and count > <NUM_LIT:0> : <EOL> asyncore . loop ( timeout = <NUM_LIT> , count = <NUM_LIT:1> , use_poll = self . use_poll ) <EOL> if instance . flag : <EOL> return <EOL> count -= <NUM_LIT:1> <EOL> time . sleep ( timeout ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> def test_handle_connect ( self ) : <EOL> class TestClient ( BaseClient ) : <EOL> def handle_connect ( self ) : <EOL> self . flag = True <EOL> server = BaseServer ( self . family , self . addr ) <EOL> client = TestClient ( self . family , server . address ) <EOL> self . loop_waiting_for_flag ( client ) <EOL> def test_handle_accept ( self ) : <EOL> class TestListener ( BaseTestHandler ) : <EOL> def __init__ ( self , family , addr ) : <EOL> BaseTestHandler . __init__ ( self ) <EOL> self . create_socket ( family ) <EOL> bind_af_aware ( self . socket , addr ) <EOL> self . listen ( <NUM_LIT:5> ) <EOL> self . address = self . socket . getsockname ( ) <EOL> def handle_accept ( self ) : <EOL> self . flag = True <EOL> server = TestListener ( self . family , self . addr ) <EOL> client = BaseClient ( self . family , server . address ) <EOL> self . loop_waiting_for_flag ( server ) <EOL> def test_handle_accepted ( self ) : <EOL> class TestListener ( BaseTestHandler ) : <EOL> def __init__ ( self , family , addr ) : <EOL> BaseTestHandler . __init__ ( self ) <EOL> self . create_socket ( family ) <EOL> bind_af_aware ( self . socket , addr ) <EOL> self . listen ( <NUM_LIT:5> ) <EOL> self . address = self . socket . getsockname ( ) <EOL> def handle_accept ( self ) : <EOL> asyncore . dispatcher . handle_accept ( self ) <EOL> def handle_accepted ( self , sock , addr ) : <EOL> sock . close ( ) <EOL> self . flag = True <EOL> server = TestListener ( self . family , self . addr ) <EOL> client = BaseClient ( self . family , server . address ) <EOL> self . loop_waiting_for_flag ( server ) <EOL> def test_handle_read ( self ) : <EOL> class TestClient ( BaseClient ) : <EOL> def handle_read ( self ) : <EOL> self . flag = True <EOL> class TestHandler ( BaseTestHandler ) : <EOL> def __init__ ( self , conn ) : <EOL> BaseTestHandler . __init__ ( self , conn ) <EOL> self . send ( b'<STR_LIT:x>' * <NUM_LIT> ) <EOL> server = BaseServer ( self . family , self . addr , TestHandler ) <EOL> client = TestClient ( self . family , server . address ) <EOL> self . loop_waiting_for_flag ( client ) <EOL> def test_handle_write ( self ) : <EOL> class TestClient ( BaseClient ) : <EOL> def handle_write ( self ) : <EOL> self . flag = True <EOL> server = BaseServer ( self . family , self . addr ) <EOL> client = TestClient ( self . family , server . address ) <EOL> self . loop_waiting_for_flag ( client ) <EOL> def test_handle_close ( self ) : <EOL> class TestClient ( BaseClient ) : <EOL> def handle_read ( self ) : <EOL> self . recv ( <NUM_LIT> ) <EOL> def handle_close ( self ) : <EOL> self . flag = True <EOL> self . close ( ) <EOL> class TestHandler ( BaseTestHandler ) : <EOL> def __init__ ( self , conn ) : <EOL> BaseTestHandler . __init__ ( self , conn ) <EOL> self . close ( ) <EOL> server = BaseServer ( self . family , self . addr , TestHandler ) <EOL> client = TestClient ( self . family , server . address ) <EOL> self . loop_waiting_for_flag ( client ) <EOL> def test_handle_close_after_conn_broken ( self ) : <EOL> data = b'<STR_LIT>' * <NUM_LIT> <EOL> class TestClient ( BaseClient ) : <EOL> def handle_write ( self ) : <EOL> self . send ( data ) <EOL> def handle_close ( self ) : <EOL> self . flag = True <EOL> self . close ( ) <EOL> def handle_expt ( self ) : <EOL> self . flag = True <EOL> self . close ( ) <EOL> class TestHandler ( BaseTestHandler ) : <EOL> def handle_read ( self ) : <EOL> self . recv ( len ( data ) ) <EOL> self . close ( ) <EOL> def writable ( self ) : <EOL> return False <EOL> server = BaseServer ( self . family , self . addr , TestHandler ) <EOL> client = TestClient ( self . family , server . address ) <EOL> self . loop_waiting_for_flag ( client ) <EOL> @ unittest . skipIf ( sys . platform . startswith ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" ) <EOL> def test_handle_expt ( self ) : <EOL> if HAS_UNIX_SOCKETS and self . family == socket . AF_UNIX : <EOL> self . skipTest ( "<STR_LIT>" ) <EOL> class TestClient ( BaseClient ) : <EOL> def handle_expt ( self ) : <EOL> self . socket . recv ( <NUM_LIT> , socket . MSG_OOB ) <EOL> self . flag = True <EOL> class TestHandler ( BaseTestHandler ) : <EOL> def __init__ ( self , conn ) : <EOL> BaseTestHandler . __init__ ( self , conn ) <EOL> self . socket . send ( bytes ( chr ( <NUM_LIT> ) , '<STR_LIT>' ) , socket . MSG_OOB ) <EOL> server = BaseServer ( self . family , self . addr , TestHandler ) <EOL> client = TestClient ( self . family , server . address ) <EOL> self . loop_waiting_for_flag ( client ) <EOL> def test_handle_error ( self ) : <EOL> class TestClient ( BaseClient ) : <EOL> def handle_write ( self ) : <EOL> <NUM_LIT:1.0> / <NUM_LIT:0> <EOL> def handle_error ( self ) : <EOL> self . flag = True <EOL> try : <EOL> raise <EOL> except ZeroDivisionError : <EOL> pass <EOL> else : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> server = BaseServer ( self . family , self . addr ) <EOL> client = TestClient ( self . family , server . address ) <EOL> self . loop_waiting_for_flag ( client ) <EOL> def test_connection_attributes ( self ) : <EOL> server = BaseServer ( self . family , self . addr ) <EOL> client = BaseClient ( self . family , server . address ) <EOL> self . assertFalse ( server . connected ) <EOL> self . assertTrue ( server . accepting ) <EOL> self . assertFalse ( client . accepting ) <EOL> asyncore . loop ( timeout = <NUM_LIT> , use_poll = self . use_poll , count = <NUM_LIT:100> ) <EOL> self . assertFalse ( server . connected ) <EOL> self . assertTrue ( server . accepting ) <EOL> self . assertTrue ( client . connected ) <EOL> self . assertFalse ( client . accepting ) <EOL> client . close ( ) <EOL> self . assertFalse ( server . connected ) <EOL> self . assertTrue ( server . accepting ) <EOL> self . assertFalse ( client . connected ) <EOL> self . assertFalse ( client . accepting ) <EOL> server . close ( ) <EOL> self . assertFalse ( server . connected ) <EOL> self . assertFalse ( server . accepting ) <EOL> def test_create_socket ( self ) : <EOL> s = asyncore . dispatcher ( ) <EOL> s . create_socket ( self . family ) <EOL> self . assertEqual ( s . socket . family , self . family ) <EOL> SOCK_NONBLOCK = getattr ( socket , '<STR_LIT>' , <NUM_LIT:0> ) <EOL> self . assertEqual ( s . socket . type , socket . SOCK_STREAM | SOCK_NONBLOCK ) <EOL> def test_bind ( self ) : <EOL> if HAS_UNIX_SOCKETS and self . family == socket . AF_UNIX : <EOL> self . skipTest ( "<STR_LIT>" ) <EOL> s1 = asyncore . dispatcher ( ) <EOL> s1 . create_socket ( self . family ) <EOL> s1 . bind ( self . addr ) <EOL> s1 . listen ( <NUM_LIT:5> ) <EOL> port = s1 . socket . getsockname ( ) [ <NUM_LIT:1> ] <EOL> s2 = asyncore . dispatcher ( ) <EOL> s2 . create_socket ( self . family ) <EOL> self . assertRaises ( socket . error , s2 . bind , ( self . addr [ <NUM_LIT:0> ] , port ) ) <EOL> def test_set_reuse_addr ( self ) : <EOL> if HAS_UNIX_SOCKETS and self . family == socket . AF_UNIX : <EOL> self . skipTest ( "<STR_LIT>" ) <EOL> sock = socket . socket ( self . family ) <EOL> try : <EOL> sock . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , <NUM_LIT:1> ) <EOL> except socket . error : <EOL> unittest . skip ( "<STR_LIT>" ) <EOL> else : <EOL> s = asyncore . dispatcher ( socket . socket ( self . family ) ) <EOL> self . assertFalse ( s . socket . getsockopt ( socket . SOL_SOCKET , <EOL> socket . SO_REUSEADDR ) ) <EOL> s . socket . close ( ) <EOL> s . create_socket ( self . family ) <EOL> s . set_reuse_addr ( ) <EOL> self . assertTrue ( s . socket . getsockopt ( socket . SOL_SOCKET , <EOL> socket . SO_REUSEADDR ) ) <EOL> finally : <EOL> sock . close ( ) <EOL> @ unittest . skipUnless ( threading , '<STR_LIT>' ) <EOL> @ support . reap_threads <EOL> def test_quick_connect ( self ) : <EOL> if self . family in ( socket . AF_INET , getattr ( socket , "<STR_LIT>" , object ( ) ) ) : <EOL> server = BaseServer ( self . family , self . addr ) <EOL> t = threading . Thread ( target = lambda : asyncore . loop ( timeout = <NUM_LIT:0.1> , <EOL> count = <NUM_LIT> ) ) <EOL> t . start ( ) <EOL> self . addCleanup ( t . join ) <EOL> s = socket . socket ( self . family , socket . SOCK_STREAM ) <EOL> s . settimeout ( <NUM_LIT> ) <EOL> s . setsockopt ( socket . SOL_SOCKET , socket . SO_LINGER , <EOL> struct . pack ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:0> ) ) <EOL> try : <EOL> s . connect ( server . address ) <EOL> except socket . error : <EOL> pass <EOL> finally : <EOL> s . close ( ) <EOL> class TestAPI_UseIPv4Sockets ( BaseTestAPI ) : <EOL> family = socket . AF_INET <EOL> addr = ( HOST , <NUM_LIT:0> ) <EOL> @ unittest . skipUnless ( support . IPV6_ENABLED , '<STR_LIT>' ) <EOL> class TestAPI_UseIPv6Sockets ( BaseTestAPI ) : <EOL> family = socket . AF_INET6 <EOL> addr = ( HOSTv6 , <NUM_LIT:0> ) <EOL> @ unittest . skipUnless ( HAS_UNIX_SOCKETS , '<STR_LIT>' ) <EOL> class TestAPI_UseUnixSockets ( BaseTestAPI ) : <EOL> if HAS_UNIX_SOCKETS : <EOL> family = socket . AF_UNIX <EOL> addr = support . TESTFN <EOL> def tearDown ( self ) : <EOL> unlink ( self . addr ) <EOL> BaseTestAPI . tearDown ( self ) <EOL> class TestAPI_UseIPv4Select ( TestAPI_UseIPv4Sockets , unittest . TestCase ) : <EOL> use_poll = False <EOL> @ unittest . skipUnless ( hasattr ( select , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> class TestAPI_UseIPv4Poll ( TestAPI_UseIPv4Sockets , unittest . TestCase ) : <EOL> use_poll = True <EOL> class TestAPI_UseIPv6Select ( TestAPI_UseIPv6Sockets , unittest . TestCase ) : <EOL> use_poll = False <EOL> @ unittest . skipUnless ( hasattr ( select , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> class TestAPI_UseIPv6Poll ( TestAPI_UseIPv6Sockets , unittest . TestCase ) : <EOL> use_poll = True <EOL> class TestAPI_UseUnixSocketsSelect ( TestAPI_UseUnixSockets , unittest . TestCase ) : <EOL> use_poll = False <EOL> @ unittest . skipUnless ( hasattr ( select , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> class TestAPI_UseUnixSocketsPoll ( TestAPI_UseUnixSockets , unittest . TestCase ) : <EOL> use_poll = True <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> unittest . main ( ) </s>
