<s> """<STR_LIT>""" <EOL> import contextlib <EOL> import imp <EOL> import os <EOL> import select <EOL> import signal <EOL> import socket <EOL> import tempfile <EOL> import unittest <EOL> import SocketServer <EOL> import test . test_support <EOL> from test . test_support import reap_children , reap_threads , verbose <EOL> try : <EOL> import threading <EOL> except ImportError : <EOL> threading = None <EOL> test . test_support . requires ( "<STR_LIT>" ) <EOL> TEST_STR = "<STR_LIT>" <EOL> HOST = test . test_support . HOST <EOL> HAVE_UNIX_SOCKETS = hasattr ( socket , "<STR_LIT>" ) <EOL> HAVE_FORKING = hasattr ( os , "<STR_LIT>" ) and os . name != "<STR_LIT>" <EOL> def signal_alarm ( n ) : <EOL> """<STR_LIT>""" <EOL> if hasattr ( signal , '<STR_LIT>' ) and not test . test_support . is_jython : <EOL> signal . alarm ( n ) <EOL> select_fn = select . select <EOL> if test . test_support . is_jython : <EOL> select_fn = select . cpython_compatible_select <EOL> def receive ( sock , n , timeout = <NUM_LIT:20> ) : <EOL> r , w , x = select_fn ( [ sock ] , [ ] , [ ] , timeout ) <EOL> if sock in r : <EOL> return sock . recv ( n ) <EOL> else : <EOL> raise RuntimeError , "<STR_LIT>" % ( sock , ) <EOL> if HAVE_UNIX_SOCKETS : <EOL> class ForkingUnixStreamServer ( SocketServer . ForkingMixIn , <EOL> SocketServer . UnixStreamServer ) : <EOL> pass <EOL> class ForkingUnixDatagramServer ( SocketServer . ForkingMixIn , <EOL> SocketServer . UnixDatagramServer ) : <EOL> pass <EOL> @ contextlib . contextmanager <EOL> def simple_subprocess ( testcase ) : <EOL> pid = os . fork ( ) <EOL> if pid == <NUM_LIT:0> : <EOL> os . _exit ( <NUM_LIT> ) <EOL> yield None <EOL> pid2 , status = os . waitpid ( pid , <NUM_LIT:0> ) <EOL> testcase . assertEqual ( pid2 , pid ) <EOL> testcase . assertEqual ( <NUM_LIT> << <NUM_LIT:8> , status ) <EOL> @ unittest . skipUnless ( threading , '<STR_LIT>' ) <EOL> class SocketServerTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> signal_alarm ( <NUM_LIT:20> ) <EOL> self . port_seed = <NUM_LIT:0> <EOL> self . test_files = [ ] <EOL> def tearDown ( self ) : <EOL> signal_alarm ( <NUM_LIT:0> ) <EOL> reap_children ( ) <EOL> for fn in self . test_files : <EOL> try : <EOL> os . remove ( fn ) <EOL> except os . error : <EOL> pass <EOL> self . test_files [ : ] = [ ] <EOL> def pickaddr ( self , proto ) : <EOL> if proto == socket . AF_INET : <EOL> return ( HOST , <NUM_LIT:0> ) <EOL> else : <EOL> dir = None <EOL> if os . name == '<STR_LIT>' : <EOL> dir = '<STR_LIT>' <EOL> fn = tempfile . mktemp ( prefix = '<STR_LIT>' , dir = dir ) <EOL> if os . name == '<STR_LIT>' : <EOL> if fn [ <NUM_LIT:1> ] == '<STR_LIT::>' : <EOL> fn = fn [ <NUM_LIT:2> : ] <EOL> if fn [ <NUM_LIT:0> ] in ( os . sep , os . altsep ) : <EOL> fn = fn [ <NUM_LIT:1> : ] <EOL> if os . sep == '<STR_LIT:/>' : <EOL> fn = fn . replace ( os . sep , os . altsep ) <EOL> else : <EOL> fn = fn . replace ( os . altsep , os . sep ) <EOL> self . test_files . append ( fn ) <EOL> return fn <EOL> def make_server ( self , addr , svrcls , hdlrbase ) : <EOL> class MyServer ( svrcls ) : <EOL> def handle_error ( self , request , client_address ) : <EOL> self . close_request ( request ) <EOL> self . server_close ( ) <EOL> raise <EOL> class MyHandler ( hdlrbase ) : <EOL> def handle ( self ) : <EOL> line = self . rfile . readline ( ) <EOL> self . wfile . write ( line ) <EOL> if verbose : print "<STR_LIT>" <EOL> server = MyServer ( addr , MyHandler ) <EOL> self . assertEqual ( server . server_address , server . socket . getsockname ( ) ) <EOL> return server <EOL> @ unittest . skipUnless ( threading , '<STR_LIT>' ) <EOL> @ reap_threads <EOL> def run_server ( self , svrcls , hdlrbase , testfunc ) : <EOL> server = self . make_server ( self . pickaddr ( svrcls . address_family ) , <EOL> svrcls , hdlrbase ) <EOL> addr = server . server_address <EOL> if verbose : <EOL> print "<STR_LIT>" <EOL> print "<STR_LIT>" , addr <EOL> print "<STR_LIT>" , svrcls <EOL> t = threading . Thread ( <EOL> name = '<STR_LIT>' % svrcls , <EOL> target = server . serve_forever , <EOL> kwargs = { '<STR_LIT>' : <NUM_LIT> } ) <EOL> t . daemon = True <EOL> t . start ( ) <EOL> if verbose : print "<STR_LIT>" <EOL> for i in range ( <NUM_LIT:3> ) : <EOL> if verbose : print "<STR_LIT>" , i <EOL> testfunc ( svrcls . address_family , addr ) <EOL> if verbose : print "<STR_LIT>" <EOL> server . shutdown ( ) <EOL> t . join ( ) <EOL> if verbose : print "<STR_LIT>" <EOL> def stream_examine ( self , proto , addr ) : <EOL> s = socket . socket ( proto , socket . SOCK_STREAM ) <EOL> s . connect ( addr ) <EOL> s . sendall ( TEST_STR ) <EOL> buf = data = receive ( s , <NUM_LIT:100> ) <EOL> while data and '<STR_LIT:\n>' not in buf : <EOL> data = receive ( s , <NUM_LIT:100> ) <EOL> buf += data <EOL> self . assertEqual ( buf , TEST_STR ) <EOL> s . close ( ) <EOL> def dgram_examine ( self , proto , addr ) : <EOL> s = socket . socket ( proto , socket . SOCK_DGRAM ) <EOL> s . sendto ( TEST_STR , addr ) <EOL> buf = data = receive ( s , <NUM_LIT:100> ) <EOL> while data and '<STR_LIT:\n>' not in buf : <EOL> data = receive ( s , <NUM_LIT:100> ) <EOL> buf += data <EOL> self . assertEqual ( buf , TEST_STR ) <EOL> s . close ( ) <EOL> def test_TCPServer ( self ) : <EOL> self . run_server ( SocketServer . TCPServer , <EOL> SocketServer . StreamRequestHandler , <EOL> self . stream_examine ) <EOL> def test_ThreadingTCPServer ( self ) : <EOL> self . run_server ( SocketServer . ThreadingTCPServer , <EOL> SocketServer . StreamRequestHandler , <EOL> self . stream_examine ) <EOL> if HAVE_FORKING : <EOL> def test_ForkingTCPServer ( self ) : <EOL> with simple_subprocess ( self ) : <EOL> self . run_server ( SocketServer . ForkingTCPServer , <EOL> SocketServer . StreamRequestHandler , <EOL> self . stream_examine ) <EOL> if HAVE_UNIX_SOCKETS : <EOL> def test_UnixStreamServer ( self ) : <EOL> self . run_server ( SocketServer . UnixStreamServer , <EOL> SocketServer . StreamRequestHandler , <EOL> self . stream_examine ) <EOL> def test_ThreadingUnixStreamServer ( self ) : <EOL> self . run_server ( SocketServer . ThreadingUnixStreamServer , <EOL> SocketServer . StreamRequestHandler , <EOL> self . stream_examine ) <EOL> if HAVE_FORKING : <EOL> def test_ForkingUnixStreamServer ( self ) : <EOL> with simple_subprocess ( self ) : <EOL> self . run_server ( ForkingUnixStreamServer , <EOL> SocketServer . StreamRequestHandler , <EOL> self . stream_examine ) <EOL> def test_UDPServer ( self ) : <EOL> self . run_server ( SocketServer . UDPServer , <EOL> SocketServer . DatagramRequestHandler , <EOL> self . dgram_examine ) <EOL> def test_ThreadingUDPServer ( self ) : <EOL> self . run_server ( SocketServer . ThreadingUDPServer , <EOL> SocketServer . DatagramRequestHandler , <EOL> self . dgram_examine ) <EOL> if HAVE_FORKING : <EOL> def test_ForkingUDPServer ( self ) : <EOL> with simple_subprocess ( self ) : <EOL> self . run_server ( SocketServer . ForkingUDPServer , <EOL> SocketServer . DatagramRequestHandler , <EOL> self . dgram_examine ) <EOL> @ reap_threads <EOL> def test_shutdown ( self ) : <EOL> class MyServer ( SocketServer . TCPServer ) : <EOL> pass <EOL> class MyHandler ( SocketServer . StreamRequestHandler ) : <EOL> pass <EOL> threads = [ ] <EOL> for i in range ( <NUM_LIT:20> ) : <EOL> s = MyServer ( ( HOST , <NUM_LIT:0> ) , MyHandler ) <EOL> t = threading . Thread ( <EOL> name = '<STR_LIT>' , <EOL> target = s . serve_forever , <EOL> kwargs = { '<STR_LIT>' : <NUM_LIT> } ) <EOL> t . daemon = True <EOL> threads . append ( ( t , s ) ) <EOL> for t , s in threads : <EOL> t . start ( ) <EOL> s . shutdown ( ) <EOL> for t , s in threads : <EOL> t . join ( ) <EOL> def test_main ( ) : <EOL> if imp . lock_held ( ) : <EOL> raise unittest . SkipTest ( "<STR_LIT>" ) <EOL> test . test_support . run_unittest ( SocketServerTest ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> test_main ( ) <EOL> signal_alarm ( <NUM_LIT:3> ) </s>
