<s> from cherrypy . _cpcompat import BadStatusLine , ntob <EOL> import os <EOL> import sys <EOL> import threading <EOL> import time <EOL> import cherrypy <EOL> engine = cherrypy . engine <EOL> thisdir = os . path . join ( os . getcwd ( ) , os . path . dirname ( __file__ ) ) <EOL> class Dependency : <EOL> def __init__ ( self , bus ) : <EOL> self . bus = bus <EOL> self . running = False <EOL> self . startcount = <NUM_LIT:0> <EOL> self . gracecount = <NUM_LIT:0> <EOL> self . threads = { } <EOL> def subscribe ( self ) : <EOL> self . bus . subscribe ( '<STR_LIT:start>' , self . start ) <EOL> self . bus . subscribe ( '<STR_LIT>' , self . stop ) <EOL> self . bus . subscribe ( '<STR_LIT>' , self . graceful ) <EOL> self . bus . subscribe ( '<STR_LIT>' , self . startthread ) <EOL> self . bus . subscribe ( '<STR_LIT>' , self . stopthread ) <EOL> def start ( self ) : <EOL> self . running = True <EOL> self . startcount += <NUM_LIT:1> <EOL> def stop ( self ) : <EOL> self . running = False <EOL> def graceful ( self ) : <EOL> self . gracecount += <NUM_LIT:1> <EOL> def startthread ( self , thread_id ) : <EOL> self . threads [ thread_id ] = None <EOL> def stopthread ( self , thread_id ) : <EOL> del self . threads [ thread_id ] <EOL> db_connection = Dependency ( engine ) <EOL> def setup_server ( ) : <EOL> class Root : <EOL> def index ( self ) : <EOL> return "<STR_LIT>" <EOL> index . exposed = True <EOL> def ctrlc ( self ) : <EOL> raise KeyboardInterrupt ( ) <EOL> ctrlc . exposed = True <EOL> def graceful ( self ) : <EOL> engine . graceful ( ) <EOL> return "<STR_LIT>" <EOL> graceful . exposed = True <EOL> def block_explicit ( self ) : <EOL> while True : <EOL> if cherrypy . response . timed_out : <EOL> cherrypy . response . timed_out = False <EOL> return "<STR_LIT>" <EOL> time . sleep ( <NUM_LIT> ) <EOL> block_explicit . exposed = True <EOL> def block_implicit ( self ) : <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> return "<STR_LIT>" % cherrypy . response . timeout <EOL> block_implicit . exposed = True <EOL> cherrypy . tree . mount ( Root ( ) ) <EOL> cherrypy . config . update ( { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:0.1> , <EOL> } ) <EOL> db_connection . subscribe ( ) <EOL> from cherrypy . test import helper <EOL> class ServerStateTests ( helper . CPWebCase ) : <EOL> setup_server = staticmethod ( setup_server ) <EOL> def setUp ( self ) : <EOL> cherrypy . server . socket_timeout = <NUM_LIT:0.1> <EOL> self . do_gc_test = False <EOL> def test_0_NormalStateFlow ( self ) : <EOL> engine . stop ( ) <EOL> self . assertEqual ( db_connection . running , False ) <EOL> self . assertEqual ( db_connection . startcount , <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( db_connection . threads ) , <NUM_LIT:0> ) <EOL> engine . start ( ) <EOL> self . assertEqual ( engine . state , engine . states . STARTED ) <EOL> host = cherrypy . server . socket_host <EOL> port = cherrypy . server . socket_port <EOL> self . assertRaises ( IOError , cherrypy . _cpserver . check_port , host , port ) <EOL> self . assertEqual ( db_connection . running , True ) <EOL> self . assertEqual ( db_connection . startcount , <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( db_connection . threads ) , <NUM_LIT:0> ) <EOL> self . getPage ( "<STR_LIT:/>" ) <EOL> self . assertBody ( "<STR_LIT>" ) <EOL> self . assertEqual ( len ( db_connection . threads ) , <NUM_LIT:1> ) <EOL> engine . stop ( ) <EOL> self . assertEqual ( engine . state , engine . states . STOPPED ) <EOL> self . assertEqual ( db_connection . running , False ) <EOL> self . assertEqual ( len ( db_connection . threads ) , <NUM_LIT:0> ) <EOL> def exittest ( ) : <EOL> self . getPage ( "<STR_LIT:/>" ) <EOL> self . assertBody ( "<STR_LIT>" ) <EOL> engine . exit ( ) <EOL> cherrypy . server . start ( ) <EOL> engine . start_with_callback ( exittest ) <EOL> engine . block ( ) <EOL> self . assertEqual ( engine . state , engine . states . EXITING ) <EOL> def test_1_Restart ( self ) : <EOL> cherrypy . server . start ( ) <EOL> engine . start ( ) <EOL> self . assertEqual ( db_connection . running , True ) <EOL> grace = db_connection . gracecount <EOL> self . getPage ( "<STR_LIT:/>" ) <EOL> self . assertBody ( "<STR_LIT>" ) <EOL> self . assertEqual ( len ( db_connection . threads ) , <NUM_LIT:1> ) <EOL> engine . graceful ( ) <EOL> self . assertEqual ( engine . state , engine . states . STARTED ) <EOL> self . getPage ( "<STR_LIT:/>" ) <EOL> self . assertBody ( "<STR_LIT>" ) <EOL> self . assertEqual ( db_connection . running , True ) <EOL> self . assertEqual ( db_connection . gracecount , grace + <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( db_connection . threads ) , <NUM_LIT:1> ) <EOL> self . getPage ( "<STR_LIT>" ) <EOL> self . assertEqual ( engine . state , engine . states . STARTED ) <EOL> self . assertBody ( "<STR_LIT>" ) <EOL> self . assertEqual ( db_connection . running , True ) <EOL> self . assertEqual ( db_connection . gracecount , grace + <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( db_connection . threads ) , <NUM_LIT:0> ) <EOL> engine . stop ( ) <EOL> self . assertEqual ( engine . state , engine . states . STOPPED ) <EOL> self . assertEqual ( db_connection . running , False ) <EOL> self . assertEqual ( len ( db_connection . threads ) , <NUM_LIT:0> ) <EOL> def test_2_KeyboardInterrupt ( self ) : <EOL> engine . start ( ) <EOL> cherrypy . server . start ( ) <EOL> self . persistent = True <EOL> try : <EOL> self . getPage ( "<STR_LIT:/>" ) <EOL> self . assertStatus ( '<STR_LIT>' ) <EOL> self . assertBody ( "<STR_LIT>" ) <EOL> self . assertNoHeader ( "<STR_LIT>" ) <EOL> cherrypy . server . httpserver . interrupt = KeyboardInterrupt <EOL> engine . block ( ) <EOL> self . assertEqual ( db_connection . running , False ) <EOL> self . assertEqual ( len ( db_connection . threads ) , <NUM_LIT:0> ) <EOL> self . assertEqual ( engine . state , engine . states . EXITING ) <EOL> finally : <EOL> self . persistent = False <EOL> engine . start ( ) <EOL> cherrypy . server . start ( ) <EOL> try : <EOL> self . getPage ( "<STR_LIT>" ) <EOL> except BadStatusLine : <EOL> pass <EOL> else : <EOL> print ( self . body ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> engine . block ( ) <EOL> self . assertEqual ( db_connection . running , False ) <EOL> self . assertEqual ( len ( db_connection . threads ) , <NUM_LIT:0> ) <EOL> def test_3_Deadlocks ( self ) : <EOL> cherrypy . config . update ( { '<STR_LIT>' : <NUM_LIT> } ) <EOL> engine . start ( ) <EOL> cherrypy . server . start ( ) <EOL> try : <EOL> self . assertNotEqual ( engine . timeout_monitor . thread , None ) <EOL> self . assertEqual ( engine . timeout_monitor . servings , [ ] ) <EOL> self . getPage ( "<STR_LIT:/>" ) <EOL> self . assertBody ( "<STR_LIT>" ) <EOL> while engine . timeout_monitor . servings : <EOL> sys . stdout . write ( "<STR_LIT:.>" ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . getPage ( "<STR_LIT>" ) <EOL> self . assertBody ( "<STR_LIT>" ) <EOL> self . getPage ( "<STR_LIT>" ) <EOL> self . assertStatus ( <NUM_LIT> ) <EOL> self . assertInBody ( "<STR_LIT>" ) <EOL> finally : <EOL> engine . exit ( ) <EOL> def test_4_Autoreload ( self ) : <EOL> p = helper . CPProcess ( ssl = ( self . scheme . lower ( ) == '<STR_LIT>' ) ) <EOL> p . write_conf ( <EOL> extra = '<STR_LIT>' ) <EOL> p . start ( imports = '<STR_LIT>' ) <EOL> try : <EOL> self . getPage ( "<STR_LIT>" ) <EOL> start = float ( self . body ) <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> os . utime ( os . path . join ( thisdir , "<STR_LIT>" ) , None ) <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> host = cherrypy . server . socket_host <EOL> port = cherrypy . server . socket_port <EOL> cherrypy . _cpserver . wait_for_occupied_port ( host , port ) <EOL> self . getPage ( "<STR_LIT>" ) <EOL> if not ( float ( self . body ) > start ) : <EOL> raise AssertionError ( "<STR_LIT>" % <EOL> ( float ( self . body ) , start ) ) <EOL> finally : <EOL> self . getPage ( "<STR_LIT>" ) <EOL> p . join ( ) <EOL> def test_5_Start_Error ( self ) : <EOL> p = helper . CPProcess ( ssl = ( self . scheme . lower ( ) == '<STR_LIT>' ) , <EOL> wait = True ) <EOL> p . write_conf ( <EOL> extra = """<STR_LIT>""" <EOL> ) <EOL> p . start ( imports = '<STR_LIT>' ) <EOL> if p . exit_code == <NUM_LIT:0> : <EOL> self . fail ( "<STR_LIT>" ) <EOL> class PluginTests ( helper . CPWebCase ) : <EOL> def test_daemonize ( self ) : <EOL> if os . name not in [ '<STR_LIT>' ] : <EOL> return self . skip ( "<STR_LIT>" ) <EOL> self . HOST = '<STR_LIT:127.0.0.1>' <EOL> self . PORT = <NUM_LIT> <EOL> p = helper . CPProcess ( ssl = ( self . scheme . lower ( ) == '<STR_LIT>' ) , <EOL> wait = True , daemonize = True , <EOL> socket_host = '<STR_LIT:127.0.0.1>' , <EOL> socket_port = <NUM_LIT> ) <EOL> p . write_conf ( <EOL> extra = '<STR_LIT>' ) <EOL> p . start ( imports = '<STR_LIT>' ) <EOL> try : <EOL> self . getPage ( "<STR_LIT>" ) <EOL> self . assertStatus ( <NUM_LIT:200> ) <EOL> page_pid = int ( self . body ) <EOL> self . assertEqual ( page_pid , p . get_pid ( ) ) <EOL> finally : <EOL> self . getPage ( "<STR_LIT>" ) <EOL> p . join ( ) <EOL> if p . exit_code != <NUM_LIT:0> : <EOL> self . fail ( "<STR_LIT>" ) <EOL> class SignalHandlingTests ( helper . CPWebCase ) : <EOL> def test_SIGHUP_tty ( self ) : <EOL> try : <EOL> from signal import SIGHUP <EOL> except ImportError : <EOL> return self . skip ( "<STR_LIT>" ) <EOL> p = helper . CPProcess ( ssl = ( self . scheme . lower ( ) == '<STR_LIT>' ) ) <EOL> p . write_conf ( <EOL> extra = '<STR_LIT>' ) <EOL> p . start ( imports = '<STR_LIT>' ) <EOL> os . kill ( p . get_pid ( ) , SIGHUP ) <EOL> p . join ( ) <EOL> def test_SIGHUP_daemonized ( self ) : <EOL> try : <EOL> from signal import SIGHUP <EOL> except ImportError : <EOL> return self . skip ( "<STR_LIT>" ) <EOL> if os . name not in [ '<STR_LIT>' ] : <EOL> return self . skip ( "<STR_LIT>" ) <EOL> p = helper . CPProcess ( ssl = ( self . scheme . lower ( ) == '<STR_LIT>' ) , <EOL> wait = True , daemonize = True ) <EOL> p . write_conf ( <EOL> extra = '<STR_LIT>' ) <EOL> p . start ( imports = '<STR_LIT>' ) <EOL> pid = p . get_pid ( ) <EOL> try : <EOL> os . kill ( pid , SIGHUP ) <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> self . getPage ( "<STR_LIT>" ) <EOL> self . assertStatus ( <NUM_LIT:200> ) <EOL> new_pid = int ( self . body ) <EOL> self . assertNotEqual ( new_pid , pid ) <EOL> finally : <EOL> self . getPage ( "<STR_LIT>" ) <EOL> p . join ( ) <EOL> def test_SIGTERM ( self ) : <EOL> try : <EOL> from signal import SIGTERM <EOL> except ImportError : <EOL> return self . skip ( "<STR_LIT>" ) <EOL> try : <EOL> from os import kill <EOL> except ImportError : <EOL> return self . skip ( "<STR_LIT>" ) <EOL> p = helper . CPProcess ( ssl = ( self . scheme . lower ( ) == '<STR_LIT>' ) ) <EOL> p . write_conf ( <EOL> extra = '<STR_LIT>' ) <EOL> p . start ( imports = '<STR_LIT>' ) <EOL> os . kill ( p . get_pid ( ) , SIGTERM ) <EOL> p . join ( ) <EOL> if os . name in [ '<STR_LIT>' ] : <EOL> p = helper . CPProcess ( ssl = ( self . scheme . lower ( ) == '<STR_LIT>' ) , <EOL> wait = True , daemonize = True ) <EOL> p . write_conf ( <EOL> extra = '<STR_LIT>' ) <EOL> p . start ( imports = '<STR_LIT>' ) <EOL> os . kill ( p . get_pid ( ) , SIGTERM ) <EOL> p . join ( ) <EOL> def test_signal_handler_unsubscribe ( self ) : <EOL> try : <EOL> from signal import SIGTERM <EOL> except ImportError : <EOL> return self . skip ( "<STR_LIT>" ) <EOL> try : <EOL> from os import kill <EOL> except ImportError : <EOL> return self . skip ( "<STR_LIT>" ) <EOL> p = helper . CPProcess ( ssl = ( self . scheme . lower ( ) == '<STR_LIT>' ) ) <EOL> p . write_conf ( <EOL> extra = """<STR_LIT>""" ) <EOL> p . start ( imports = '<STR_LIT>' ) <EOL> os . kill ( p . get_pid ( ) , SIGTERM ) <EOL> p . join ( ) <EOL> target_line = open ( p . error_log , '<STR_LIT:rb>' ) . readlines ( ) [ - <NUM_LIT:10> ] <EOL> if not ntob ( "<STR_LIT>" ) in target_line : <EOL> self . fail ( "<STR_LIT>" % target_line ) </s>
