<s> """<STR_LIT>""" <EOL> import errno <EOL> import json <EOL> import select <EOL> import socket <EOL> import time <EOL> import unittest <EOL> import urllib2 <EOL> import google <EOL> from cherrypy import wsgiserver <EOL> import mox <EOL> from google . appengine . tools . devappserver2 import wsgi_server <EOL> class TestError ( Exception ) : <EOL> pass <EOL> class _SingleAddressWsgiServerTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( _SingleAddressWsgiServerTest , self ) . setUp ( ) <EOL> self . server = wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT:localhost>' , <NUM_LIT:0> ) , <EOL> self . wsgi_application ) <EOL> self . server . start ( ) <EOL> def tearDown ( self ) : <EOL> super ( _SingleAddressWsgiServerTest , self ) . tearDown ( ) <EOL> self . server . quit ( ) <EOL> def test_serve ( self ) : <EOL> result = urllib2 . urlopen ( '<STR_LIT>' % <EOL> self . server . port ) <EOL> body = result . read ( ) <EOL> environ = json . loads ( body ) <EOL> self . assertEqual ( <NUM_LIT:200> , result . code ) <EOL> self . assertEqual ( '<STR_LIT>' , environ [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , environ [ '<STR_LIT>' ] ) <EOL> def wsgi_application ( self , environ , start_response ) : <EOL> start_response ( '<STR_LIT>' , [ ( '<STR_LIT:Content-Type>' , '<STR_LIT:application/json>' ) ] ) <EOL> serializable_environ = environ . copy ( ) <EOL> del serializable_environ [ '<STR_LIT>' ] <EOL> del serializable_environ [ '<STR_LIT>' ] <EOL> return [ json . dumps ( serializable_environ ) ] <EOL> def other_wsgi_application ( self , environ , start_response ) : <EOL> start_response ( '<STR_LIT>' , [ ( '<STR_LIT:Content-Type>' , '<STR_LIT>' ) ] ) <EOL> return [ '<STR_LIT>' ] <EOL> def test_set_app ( self ) : <EOL> self . server . set_app ( self . other_wsgi_application ) <EOL> result = urllib2 . urlopen ( '<STR_LIT>' % <EOL> self . server . port ) <EOL> body = result . read ( ) <EOL> self . assertEqual ( <NUM_LIT:200> , result . code ) <EOL> self . assertEqual ( '<STR_LIT>' , body ) <EOL> def test_set_error ( self ) : <EOL> self . server . set_error ( <NUM_LIT> ) <EOL> result = urllib2 . urlopen ( '<STR_LIT>' % <EOL> self . server . port ) <EOL> self . assertEqual ( <NUM_LIT> , result . code ) <EOL> class SharedCherryPyThreadPoolTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . mox = mox . Mox ( ) <EOL> self . mox . StubOutWithMock ( wsgi_server . _THREAD_POOL , '<STR_LIT>' ) <EOL> self . thread_pool = wsgi_server . _SharedCherryPyThreadPool ( ) <EOL> def tearDown ( self ) : <EOL> self . mox . UnsetStubs ( ) <EOL> def test_put ( self ) : <EOL> connection = object ( ) <EOL> wsgi_server . _THREAD_POOL . submit ( self . thread_pool . _handle , connection ) <EOL> self . mox . ReplayAll ( ) <EOL> self . thread_pool . put ( connection ) <EOL> self . mox . VerifyAll ( ) <EOL> self . assertEqual ( set ( [ connection ] ) , self . thread_pool . _connections ) <EOL> def test_handle ( self ) : <EOL> connection = self . mox . CreateMock ( wsgiserver . HTTPConnection ) <EOL> self . mox . StubOutWithMock ( self . thread_pool . _condition , '<STR_LIT>' ) <EOL> self . thread_pool . _connections . add ( connection ) <EOL> connection . communicate ( ) <EOL> connection . close ( ) <EOL> self . thread_pool . _condition . notify ( ) <EOL> self . mox . ReplayAll ( ) <EOL> self . thread_pool . _handle ( connection ) <EOL> self . mox . VerifyAll ( ) <EOL> self . assertEqual ( set ( ) , self . thread_pool . _connections ) <EOL> def test_handle_with_exception ( self ) : <EOL> connection = self . mox . CreateMock ( wsgiserver . HTTPConnection ) <EOL> self . mox . StubOutWithMock ( self . thread_pool . _condition , '<STR_LIT>' ) <EOL> self . thread_pool . _connections . add ( connection ) <EOL> connection . communicate ( ) . AndRaise ( TestError ) <EOL> connection . close ( ) <EOL> self . thread_pool . _condition . notify ( ) <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( TestError , self . thread_pool . _handle , connection ) <EOL> self . mox . VerifyAll ( ) <EOL> self . assertEqual ( set ( ) , self . thread_pool . _connections ) <EOL> def test_stop ( self ) : <EOL> self . mox . ReplayAll ( ) <EOL> self . thread_pool . stop ( <NUM_LIT:3> ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_stop_no_connections ( self ) : <EOL> self . mox . ReplayAll ( ) <EOL> self . thread_pool . stop ( <NUM_LIT:0.1> ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_stop_with_connections ( self ) : <EOL> connection = self . mox . CreateMock ( wsgiserver . HTTPConnection ) <EOL> self . thread_pool . _connections . add ( connection ) <EOL> self . mox . StubOutWithMock ( self . thread_pool , '<STR_LIT>' ) <EOL> self . thread_pool . _shutdown_connection ( connection ) <EOL> self . mox . ReplayAll ( ) <EOL> self . thread_pool . stop ( <NUM_LIT:1> ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_shutdown_connection ( self ) : <EOL> class DummyObect ( object ) : <EOL> pass <EOL> connection = DummyObect ( ) <EOL> connection . rfile = DummyObect ( ) <EOL> connection . rfile . closed = False <EOL> connection . socket = self . mox . CreateMockAnything ( ) <EOL> connection . socket . shutdown ( socket . SHUT_RD ) <EOL> self . mox . ReplayAll ( ) <EOL> self . thread_pool . _shutdown_connection ( connection ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_shutdown_connection_rfile_already_close ( self ) : <EOL> class DummyObect ( object ) : <EOL> pass <EOL> connection = DummyObect ( ) <EOL> connection . rfile = DummyObect ( ) <EOL> connection . rfile . closed = True <EOL> connection . socket = self . mox . CreateMockAnything ( ) <EOL> self . mox . ReplayAll ( ) <EOL> self . thread_pool . _shutdown_connection ( connection ) <EOL> self . mox . VerifyAll ( ) <EOL> class SelectThreadTest ( unittest . TestCase ) : <EOL> class _MockSocket ( object ) : <EOL> def fileno ( self ) : <EOL> return id ( self ) <EOL> def setUp ( self ) : <EOL> self . select_thread = wsgi_server . SelectThread ( ) <EOL> self . original_has_poll = wsgi_server . _HAS_POLL <EOL> self . mox = mox . Mox ( ) <EOL> self . mox . StubOutWithMock ( select , '<STR_LIT>' ) <EOL> if hasattr ( select , '<STR_LIT>' ) : <EOL> self . mox . StubOutWithMock ( select , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( time , '<STR_LIT>' ) <EOL> def tearDown ( self ) : <EOL> self . mox . UnsetStubs ( ) <EOL> wsgi_server . _HAS_POLL = self . original_has_poll <EOL> def test_add_socket ( self ) : <EOL> file_descriptors = self . select_thread . _file_descriptors <EOL> file_descriptor_to_callback = ( <EOL> self . select_thread . _file_descriptor_to_callback ) <EOL> file_descriptors_copy = frozenset ( self . select_thread . _file_descriptors ) <EOL> file_descriptor_to_callback_copy = ( <EOL> self . select_thread . _file_descriptor_to_callback . copy ( ) ) <EOL> s = self . _MockSocket ( ) <EOL> callback = object ( ) <EOL> self . select_thread . add_socket ( s , callback ) <EOL> self . assertEqual ( file_descriptors_copy , file_descriptors ) <EOL> self . assertEqual ( file_descriptor_to_callback_copy , <EOL> file_descriptor_to_callback ) <EOL> self . assertEqual ( frozenset ( [ s . fileno ( ) ] ) , <EOL> self . select_thread . _file_descriptors ) <EOL> self . assertEqual ( { s . fileno ( ) : callback } , <EOL> self . select_thread . _file_descriptor_to_callback ) <EOL> def test_remove_socket ( self ) : <EOL> s1 = self . _MockSocket ( ) <EOL> callback1 = object ( ) <EOL> s2 = self . _MockSocket ( ) <EOL> callback2 = object ( ) <EOL> self . select_thread . _file_descriptors = frozenset ( [ s1 . fileno ( ) , s2 . fileno ( ) ] ) <EOL> self . select_thread . _file_descriptor_to_callback = { <EOL> s1 . fileno ( ) : callback1 , s2 . fileno ( ) : callback2 } <EOL> file_descriptors = self . select_thread . _file_descriptors <EOL> file_descriptor_to_callback = ( <EOL> self . select_thread . _file_descriptor_to_callback ) <EOL> file_descriptors_copy = frozenset ( self . select_thread . _file_descriptors ) <EOL> file_descriptor_to_callback_copy = ( <EOL> self . select_thread . _file_descriptor_to_callback . copy ( ) ) <EOL> self . select_thread . remove_socket ( s1 ) <EOL> self . assertEqual ( file_descriptors_copy , file_descriptors ) <EOL> self . assertEqual ( file_descriptor_to_callback_copy , <EOL> file_descriptor_to_callback ) <EOL> self . assertEqual ( frozenset ( [ s2 . fileno ( ) ] ) , <EOL> self . select_thread . _file_descriptors ) <EOL> self . assertEqual ( { s2 . fileno ( ) : callback2 } , <EOL> self . select_thread . _file_descriptor_to_callback ) <EOL> def test_select_no_sockets ( self ) : <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> self . mox . ReplayAll ( ) <EOL> self . select_thread . _select ( ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_select_no_poll ( self ) : <EOL> wsgi_server . _HAS_POLL = False <EOL> s = self . _MockSocket ( ) <EOL> callback = self . mox . CreateMockAnything ( ) <EOL> select . select ( frozenset ( [ s . fileno ( ) ] ) , [ ] , [ ] , <NUM_LIT:1> ) . AndReturn ( <EOL> ( [ s . fileno ( ) ] , [ ] , [ ] ) ) <EOL> callback ( ) <EOL> self . mox . ReplayAll ( ) <EOL> self . select_thread . add_socket ( s , callback ) <EOL> self . select_thread . _select ( ) <EOL> self . mox . VerifyAll ( ) <EOL> @ unittest . skipUnless ( wsgi_server . _HAS_POLL , '<STR_LIT>' ) <EOL> def test_select_with_poll ( self ) : <EOL> s = self . _MockSocket ( ) <EOL> callback = self . mox . CreateMockAnything ( ) <EOL> poll = self . mox . CreateMockAnything ( ) <EOL> select . poll ( ) . AndReturn ( poll ) <EOL> poll . register ( s . fileno ( ) , select . POLLIN ) <EOL> poll . poll ( <NUM_LIT:1000> ) . AndReturn ( [ ( s . fileno ( ) , select . POLLIN ) ] ) <EOL> callback ( ) <EOL> self . mox . ReplayAll ( ) <EOL> self . select_thread . add_socket ( s , callback ) <EOL> self . select_thread . _select ( ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_select_not_ready_no_poll ( self ) : <EOL> wsgi_server . _HAS_POLL = False <EOL> s = self . _MockSocket ( ) <EOL> callback = self . mox . CreateMockAnything ( ) <EOL> select . select ( frozenset ( [ s . fileno ( ) ] ) , [ ] , [ ] , <NUM_LIT:1> ) . AndReturn ( ( [ ] , [ ] , [ ] ) ) <EOL> self . mox . ReplayAll ( ) <EOL> self . select_thread . add_socket ( s , callback ) <EOL> self . select_thread . _select ( ) <EOL> self . mox . VerifyAll ( ) <EOL> @ unittest . skipUnless ( wsgi_server . _HAS_POLL , '<STR_LIT>' ) <EOL> def test_select_not_ready_with_poll ( self ) : <EOL> s = self . _MockSocket ( ) <EOL> callback = self . mox . CreateMockAnything ( ) <EOL> poll = self . mox . CreateMockAnything ( ) <EOL> select . poll ( ) . AndReturn ( poll ) <EOL> poll . register ( s . fileno ( ) , select . POLLIN ) <EOL> poll . poll ( <NUM_LIT:1000> ) . AndReturn ( [ ] ) <EOL> self . mox . ReplayAll ( ) <EOL> self . select_thread . add_socket ( s , callback ) <EOL> self . select_thread . _select ( ) <EOL> self . mox . VerifyAll ( ) <EOL> class WsgiServerStartupTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . mox = mox . Mox ( ) <EOL> self . server = wsgi_server . WsgiServer ( ( '<STR_LIT:localhost>' , <NUM_LIT> ) , None ) <EOL> def tearDown ( self ) : <EOL> self . mox . UnsetStubs ( ) <EOL> def test_start_some_fail_to_bind ( self ) : <EOL> failing_server = self . mox . CreateMock ( <EOL> wsgi_server . _SingleAddressWsgiServer ) <EOL> starting_server = self . mox . CreateMock ( <EOL> wsgi_server . _SingleAddressWsgiServer ) <EOL> another_starting_server = self . mox . CreateMock ( <EOL> wsgi_server . _SingleAddressWsgiServer ) <EOL> self . mox . StubOutWithMock ( wsgi_server , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( socket , '<STR_LIT>' ) <EOL> socket . getaddrinfo ( '<STR_LIT:localhost>' , <NUM_LIT> , socket . AF_UNSPEC , socket . SOCK_STREAM , <EOL> <NUM_LIT:0> , socket . AI_PASSIVE ) . AndReturn ( <EOL> [ ( None , None , None , None , ( '<STR_LIT:foo>' , '<STR_LIT:bar>' , '<STR_LIT>' ) ) , <EOL> ( None , None , None , None , ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:4> , <NUM_LIT:5> ) ) , <EOL> ( None , None , None , None , ( <NUM_LIT:3> , <NUM_LIT:4> ) ) ] ) <EOL> wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) , None ) . AndReturn ( <EOL> failing_server ) <EOL> wsgi_server . _SingleAddressWsgiServer ( ( <NUM_LIT:1> , <NUM_LIT:2> ) , None ) . AndReturn ( <EOL> starting_server ) <EOL> wsgi_server . _SingleAddressWsgiServer ( ( <NUM_LIT:3> , <NUM_LIT:4> ) , None ) . AndReturn ( <EOL> another_starting_server ) <EOL> starting_server . start ( ) <EOL> failing_server . start ( ) . AndRaise ( wsgi_server . BindError ) <EOL> another_starting_server . start ( ) <EOL> self . mox . ReplayAll ( ) <EOL> self . server . start ( ) <EOL> self . mox . VerifyAll ( ) <EOL> self . assertItemsEqual ( [ starting_server , another_starting_server ] , <EOL> self . server . _servers ) <EOL> def test_start_all_fail_to_bind ( self ) : <EOL> failing_server = self . mox . CreateMock ( <EOL> wsgi_server . _SingleAddressWsgiServer ) <EOL> self . mox . StubOutWithMock ( wsgi_server , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( socket , '<STR_LIT>' ) <EOL> socket . getaddrinfo ( '<STR_LIT:localhost>' , <NUM_LIT> , socket . AF_UNSPEC , socket . SOCK_STREAM , <EOL> <NUM_LIT:0> , socket . AI_PASSIVE ) . AndReturn ( <EOL> [ ( None , None , None , None , ( '<STR_LIT:foo>' , '<STR_LIT:bar>' , '<STR_LIT>' ) ) ] ) <EOL> wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) , None ) . AndReturn ( <EOL> failing_server ) <EOL> failing_server . start ( ) . AndRaise ( wsgi_server . BindError ) <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( wsgi_server . BindError , self . server . start ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_remove_duplicates ( self ) : <EOL> foo_server = self . mox . CreateMock ( wsgi_server . _SingleAddressWsgiServer ) <EOL> foo2_server = self . mox . CreateMock ( wsgi_server . _SingleAddressWsgiServer ) <EOL> self . mox . StubOutWithMock ( wsgi_server , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( socket , '<STR_LIT>' ) <EOL> socket . getaddrinfo ( '<STR_LIT:localhost>' , <NUM_LIT> , socket . AF_UNSPEC , socket . SOCK_STREAM , <EOL> <NUM_LIT:0> , socket . AI_PASSIVE ) . AndReturn ( <EOL> [ ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , '<STR_LIT>' , ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) ) , <EOL> ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , '<STR_LIT>' , ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , <EOL> ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , '<STR_LIT>' , ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) ) ] ) <EOL> wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) , None ) . AndReturn ( <EOL> foo_server ) <EOL> foo_server . start ( ) <EOL> wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT>' , <NUM_LIT> ) , None ) . AndReturn ( <EOL> foo2_server ) <EOL> foo2_server . start ( ) <EOL> self . mox . ReplayAll ( ) <EOL> self . server . start ( ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_quit ( self ) : <EOL> running_server = self . mox . CreateMock ( <EOL> wsgi_server . _SingleAddressWsgiServer ) <EOL> self . server . _servers = [ running_server ] <EOL> running_server . quit ( ) <EOL> self . mox . ReplayAll ( ) <EOL> self . server . quit ( ) <EOL> self . mox . VerifyAll ( ) <EOL> class WsgiServerPort0StartupTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . mox = mox . Mox ( ) <EOL> self . server = wsgi_server . WsgiServer ( ( '<STR_LIT:localhost>' , <NUM_LIT:0> ) , None ) <EOL> def tearDown ( self ) : <EOL> self . mox . UnsetStubs ( ) <EOL> def test_basic_behavior ( self ) : <EOL> inet4_server = self . mox . CreateMock ( wsgi_server . _SingleAddressWsgiServer ) <EOL> inet6_server = self . mox . CreateMock ( wsgi_server . _SingleAddressWsgiServer ) <EOL> self . mox . StubOutWithMock ( wsgi_server , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( socket , '<STR_LIT>' ) <EOL> socket . getaddrinfo ( '<STR_LIT:localhost>' , <NUM_LIT:0> , socket . AF_UNSPEC , socket . SOCK_STREAM , <NUM_LIT:0> , <EOL> socket . AI_PASSIVE ) . AndReturn ( <EOL> [ ( None , None , None , None , ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> , '<STR_LIT>' ) ) , <EOL> ( None , None , None , None , ( '<STR_LIT>' , <NUM_LIT:0> , '<STR_LIT>' ) ) ] ) <EOL> wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> ) , None ) . AndReturn ( <EOL> inet4_server ) <EOL> inet4_server . start ( ) <EOL> inet4_server . port = <NUM_LIT> <EOL> wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT>' , <NUM_LIT> ) , None ) . AndReturn ( <EOL> inet6_server ) <EOL> inet6_server . start ( ) <EOL> self . mox . ReplayAll ( ) <EOL> self . server . start ( ) <EOL> self . mox . VerifyAll ( ) <EOL> self . assertItemsEqual ( [ inet4_server , inet6_server ] , <EOL> self . server . _servers ) <EOL> def test_retry_eaddrinuse ( self ) : <EOL> inet4_server = self . mox . CreateMock ( wsgi_server . _SingleAddressWsgiServer ) <EOL> inet6_server = self . mox . CreateMock ( wsgi_server . _SingleAddressWsgiServer ) <EOL> inet4_server_retry = self . mox . CreateMock ( <EOL> wsgi_server . _SingleAddressWsgiServer ) <EOL> inet6_server_retry = self . mox . CreateMock ( <EOL> wsgi_server . _SingleAddressWsgiServer ) <EOL> self . mox . StubOutWithMock ( wsgi_server , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( socket , '<STR_LIT>' ) <EOL> socket . getaddrinfo ( '<STR_LIT:localhost>' , <NUM_LIT:0> , socket . AF_UNSPEC , socket . SOCK_STREAM , <NUM_LIT:0> , <EOL> socket . AI_PASSIVE ) . AndReturn ( <EOL> [ ( None , None , None , None , ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> , '<STR_LIT>' ) ) , <EOL> ( None , None , None , None , ( '<STR_LIT>' , <NUM_LIT:0> , '<STR_LIT>' ) ) ] ) <EOL> wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> ) , None ) . AndReturn ( <EOL> inet4_server ) <EOL> inet4_server . start ( ) <EOL> inet4_server . port = <NUM_LIT> <EOL> wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT>' , <NUM_LIT> ) , None ) . AndReturn ( <EOL> inet6_server ) <EOL> inet6_server . start ( ) . AndRaise ( <EOL> wsgi_server . BindError ( '<STR_LIT:message>' , ( errno . EADDRINUSE , '<STR_LIT>' ) ) ) <EOL> inet4_server . quit ( ) <EOL> wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> ) , None ) . AndReturn ( <EOL> inet4_server_retry ) <EOL> inet4_server_retry . start ( ) <EOL> inet4_server_retry . port = <NUM_LIT> <EOL> wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT>' , <NUM_LIT> ) , None ) . AndReturn ( <EOL> inet6_server_retry ) <EOL> inet6_server_retry . start ( ) <EOL> self . mox . ReplayAll ( ) <EOL> self . server . start ( ) <EOL> self . mox . VerifyAll ( ) <EOL> self . assertItemsEqual ( [ inet4_server_retry , inet6_server_retry ] , <EOL> self . server . _servers ) <EOL> def test_retry_limited ( self ) : <EOL> inet4_servers = [ self . mox . CreateMock ( wsgi_server . _SingleAddressWsgiServer ) <EOL> for _ in range ( wsgi_server . _PORT_0_RETRIES ) ] <EOL> inet6_servers = [ self . mox . CreateMock ( wsgi_server . _SingleAddressWsgiServer ) <EOL> for _ in range ( wsgi_server . _PORT_0_RETRIES ) ] <EOL> self . mox . StubOutWithMock ( wsgi_server , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( socket , '<STR_LIT>' ) <EOL> socket . getaddrinfo ( '<STR_LIT:localhost>' , <NUM_LIT:0> , socket . AF_UNSPEC , socket . SOCK_STREAM , <NUM_LIT:0> , <EOL> socket . AI_PASSIVE ) . AndReturn ( <EOL> [ ( None , None , None , None , ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> , '<STR_LIT>' ) ) , <EOL> ( None , None , None , None , ( '<STR_LIT>' , <NUM_LIT:0> , '<STR_LIT>' ) ) ] ) <EOL> for offset , ( inet4_server , inet6_server ) in enumerate ( zip ( <EOL> inet4_servers , inet6_servers ) ) : <EOL> wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> ) , None ) . AndReturn ( <EOL> inet4_server ) <EOL> inet4_server . start ( ) <EOL> inet4_server . port = offset + <NUM_LIT:1> <EOL> wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT>' , offset + <NUM_LIT:1> ) , None ) . AndReturn ( <EOL> inet6_server ) <EOL> inet6_server . start ( ) . AndRaise ( <EOL> wsgi_server . BindError ( '<STR_LIT:message>' , ( errno . EADDRINUSE , '<STR_LIT>' ) ) ) <EOL> inet4_server . quit ( ) <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( wsgi_server . BindError , self . server . start ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_ignore_other_errors ( self ) : <EOL> inet4_server = self . mox . CreateMock ( wsgi_server . _SingleAddressWsgiServer ) <EOL> inet6_server = self . mox . CreateMock ( wsgi_server . _SingleAddressWsgiServer ) <EOL> self . mox . StubOutWithMock ( wsgi_server , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( socket , '<STR_LIT>' ) <EOL> socket . getaddrinfo ( '<STR_LIT:localhost>' , <NUM_LIT:0> , socket . AF_UNSPEC , socket . SOCK_STREAM , <NUM_LIT:0> , <EOL> socket . AI_PASSIVE ) . AndReturn ( <EOL> [ ( None , None , None , None , ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> , '<STR_LIT>' ) ) , <EOL> ( None , None , None , None , ( '<STR_LIT>' , <NUM_LIT:0> , '<STR_LIT>' ) ) ] ) <EOL> wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT:127.0.0.1>' , <NUM_LIT:0> ) , None ) . AndReturn ( <EOL> inet4_server ) <EOL> inet4_server . start ( ) <EOL> inet4_server . port = <NUM_LIT> <EOL> wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT>' , <NUM_LIT> ) , None ) . AndReturn ( <EOL> inet6_server ) <EOL> inet6_server . start ( ) . AndRaise ( <EOL> wsgi_server . BindError ( '<STR_LIT:message>' , ( errno . ENOPROTOOPT , '<STR_LIT>' ) ) ) <EOL> self . mox . ReplayAll ( ) <EOL> self . server . start ( ) <EOL> self . mox . VerifyAll ( ) <EOL> self . assertItemsEqual ( [ inet4_server ] , <EOL> self . server . _servers ) <EOL> class _SingleAddressWsgiServerStartupTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . mox = mox . Mox ( ) <EOL> self . server = wsgi_server . _SingleAddressWsgiServer ( ( '<STR_LIT:localhost>' , <NUM_LIT:0> ) , None ) <EOL> def tearDown ( self ) : <EOL> self . mox . UnsetStubs ( ) <EOL> def test_start_port_in_use ( self ) : <EOL> self . mox . StubOutWithMock ( socket , '<STR_LIT>' ) <EOL> self . mox . StubOutWithMock ( self . server , '<STR_LIT>' ) <EOL> af = object ( ) <EOL> socktype = object ( ) <EOL> proto = object ( ) <EOL> socket . getaddrinfo ( '<STR_LIT:localhost>' , <NUM_LIT:0> , socket . AF_UNSPEC , socket . SOCK_STREAM , <NUM_LIT:0> , <EOL> socket . AI_PASSIVE ) . AndReturn ( <EOL> [ ( af , socktype , proto , None , None ) ] ) <EOL> self . server . bind ( af , socktype , proto ) . AndRaise ( socket . error ) <EOL> self . mox . ReplayAll ( ) <EOL> self . assertRaises ( wsgi_server . BindError , self . server . start ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_start ( self ) : <EOL> self . mox . StubOutWithMock ( wsgiserver . ThreadPool , '<STR_LIT:start>' ) <EOL> self . mox . StubOutWithMock ( wsgi_server . _SELECT_THREAD , '<STR_LIT>' ) <EOL> wsgi_server . _SELECT_THREAD . add_socket ( mox . IsA ( socket . socket ) , <EOL> self . server . tick ) <EOL> self . mox . ReplayAll ( ) <EOL> self . server . start ( ) <EOL> self . mox . VerifyAll ( ) <EOL> def test_quit ( self ) : <EOL> self . mox . StubOutWithMock ( wsgi_server . _SELECT_THREAD , '<STR_LIT>' ) <EOL> self . server . socket = object ( ) <EOL> self . server . requests = self . mox . CreateMock ( <EOL> wsgi_server . _SharedCherryPyThreadPool ) <EOL> wsgi_server . _SELECT_THREAD . remove_socket ( self . server . socket ) <EOL> self . server . requests . stop ( timeout = <NUM_LIT:1> ) <EOL> self . mox . ReplayAll ( ) <EOL> self . server . quit ( ) <EOL> self . mox . VerifyAll ( ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
