<s> import socket <EOL> from django . core . servers import basehttp <EOL> from django . core . handlers . wsgi import WSGIHandler <EOL> from django . contrib . staticfiles . handlers import StaticFilesHandler <EOL> import threading <EOL> from django_selenium import settings <EOL> try : <EOL> from django . core . servers . basehttp import WSGIServerException as wsgi_exec_error <EOL> except ImportError : <EOL> import socket <EOL> wsgi_exec_error = socket . error <EOL> class StoppableWSGIServer ( basehttp . WSGIServer ) : <EOL> """<STR_LIT>""" <EOL> def server_bind ( self ) : <EOL> """<STR_LIT>""" <EOL> basehttp . WSGIServer . server_bind ( self ) <EOL> self . socket . settimeout ( <NUM_LIT:1> ) <EOL> def get_request ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> sock , address = self . socket . accept ( ) <EOL> sock . settimeout ( None ) <EOL> return ( sock , address ) <EOL> except socket . timeout : <EOL> raise <EOL> class TestServerThread ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , address , port ) : <EOL> self . address = address <EOL> self . port = port <EOL> self . _start_event = threading . Event ( ) <EOL> self . _stop_event = threading . Event ( ) <EOL> self . _activate_event = threading . Event ( ) <EOL> self . _ready_event = threading . Event ( ) <EOL> self . _error = None <EOL> super ( TestServerThread , self ) . __init__ ( ) <EOL> def run ( self ) : <EOL> try : <EOL> handler = StaticFilesHandler ( WSGIHandler ( ) ) <EOL> def test_app ( environ , start_response ) : <EOL> if environ [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> start_response ( '<STR_LIT>' , [ ] ) <EOL> return '<STR_LIT>' <EOL> if environ [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> start_response ( '<STR_LIT>' , [ ] ) <EOL> return '<STR_LIT>' <EOL> return handler ( environ , start_response ) <EOL> server_address = ( self . address , self . port ) <EOL> httpd = StoppableWSGIServer ( server_address , basehttp . WSGIRequestHandler ) <EOL> httpd . set_app ( test_app ) <EOL> self . _start_event . set ( ) <EOL> except wsgi_exec_error as e : <EOL> self . error = e <EOL> self . _start_event . set ( ) <EOL> return <EOL> self . _activate_event . set ( ) <EOL> while not self . _stop_event . is_set ( ) : <EOL> if self . _activate_event . wait ( <NUM_LIT:5> ) : <EOL> httpd . handle_request ( ) <EOL> self . _ready_event . set ( ) <EOL> def stop ( self , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> self . _stop_event . set ( ) <EOL> self . join ( timeout ) <EOL> def activate ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _activate_event . set ( ) <EOL> self . _ready_event . clear ( ) <EOL> if not self . _ready_event . wait ( settings . SELENIUM_TEST_SERVER_TIMEOUT ) : <EOL> raise Exception ( '<STR_LIT>' % settings . SELENIUM_TEST_SERVER_TIMEOUT ) <EOL> def deactivate ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _activate_event . clear ( ) <EOL> self . _ready_event . clear ( ) <EOL> if not self . _ready_event . wait ( settings . SELENIUM_TEST_SERVER_TIMEOUT ) : <EOL> raise Exception ( '<STR_LIT>' % settings . SELENIUM_TEST_SERVER_TIMEOUT ) <EOL> def get_test_server ( ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT>' not in globals ( ) : <EOL> server_thread = TestServerThread ( settings . SELENIUM_TESTSERVER_HOST , settings . SELENIUM_TESTSERVER_PORT ) <EOL> server_thread . start ( ) <EOL> server_thread . _start_event . wait ( ) <EOL> if server_thread . _error : <EOL> raise server_thread . _error <EOL> globals ( ) [ '<STR_LIT>' ] = server_thread <EOL> return globals ( ) [ '<STR_LIT>' ] </s>
