<s> """<STR_LIT>""" <EOL> import datetime <EOL> import logging <EOL> log = logging . getLogger ( __name__ ) <EOL> import os <EOL> thisdir = os . path . abspath ( os . path . dirname ( __file__ ) ) <EOL> serverpem = os . path . join ( os . getcwd ( ) , thisdir , '<STR_LIT>' ) <EOL> import re <EOL> import sys <EOL> import time <EOL> import warnings <EOL> import cherrypy <EOL> from cherrypy . _cpcompat import basestring , copyitems , HTTPSConnection , ntob <EOL> from cherrypy . lib import httputil <EOL> from cherrypy . lib import gctools <EOL> from cherrypy . lib . reprconf import unrepr <EOL> from cherrypy . test import webtest <EOL> import nose <EOL> _testconfig = None <EOL> def get_tst_config ( overconf = { } ) : <EOL> global _testconfig <EOL> if _testconfig is None : <EOL> conf = { <EOL> '<STR_LIT>' : '<STR_LIT:http>' , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT:port>' : <NUM_LIT> , <EOL> '<STR_LIT:host>' : '<STR_LIT:127.0.0.1>' , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> try : <EOL> import testconfig <EOL> _conf = testconfig . config . get ( '<STR_LIT>' , None ) <EOL> if _conf is not None : <EOL> for k , v in _conf . items ( ) : <EOL> if isinstance ( v , basestring ) : <EOL> _conf [ k ] = unrepr ( v ) <EOL> conf . update ( _conf ) <EOL> except ImportError : <EOL> pass <EOL> _testconfig = conf <EOL> conf = _testconfig . copy ( ) <EOL> conf . update ( overconf ) <EOL> return conf <EOL> class Supervisor ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , ** kwargs ) : <EOL> for k , v in kwargs . items ( ) : <EOL> if k == '<STR_LIT:port>' : <EOL> setattr ( self , k , int ( v ) ) <EOL> setattr ( self , k , v ) <EOL> log_to_stderr = lambda msg , level : sys . stderr . write ( msg + os . linesep ) <EOL> class LocalSupervisor ( Supervisor ) : <EOL> """<STR_LIT>""" <EOL> using_apache = False <EOL> using_wsgi = False <EOL> def __init__ ( self , ** kwargs ) : <EOL> for k , v in kwargs . items ( ) : <EOL> setattr ( self , k , v ) <EOL> cherrypy . server . httpserver = self . httpserver_class <EOL> cherrypy . config . update ( { '<STR_LIT>' : False } ) <EOL> engine = cherrypy . engine <EOL> if hasattr ( engine , "<STR_LIT>" ) : <EOL> engine . signal_handler . subscribe ( ) <EOL> if hasattr ( engine , "<STR_LIT>" ) : <EOL> engine . console_control_handler . subscribe ( ) <EOL> def start ( self , modulename = None ) : <EOL> """<STR_LIT>""" <EOL> if modulename : <EOL> cherrypy . server . httpserver = None <EOL> cherrypy . engine . start ( ) <EOL> self . sync_apps ( ) <EOL> def sync_apps ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def stop ( self ) : <EOL> td = getattr ( self , '<STR_LIT>' , None ) <EOL> if td : <EOL> td ( ) <EOL> cherrypy . engine . exit ( ) <EOL> for name , server in copyitems ( getattr ( cherrypy , '<STR_LIT>' , { } ) ) : <EOL> server . unsubscribe ( ) <EOL> del cherrypy . servers [ name ] <EOL> class NativeServerSupervisor ( LocalSupervisor ) : <EOL> """<STR_LIT>""" <EOL> httpserver_class = "<STR_LIT>" <EOL> using_apache = False <EOL> using_wsgi = False <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" % ( self . host , self . port ) <EOL> class LocalWSGISupervisor ( LocalSupervisor ) : <EOL> """<STR_LIT>""" <EOL> httpserver_class = "<STR_LIT>" <EOL> using_apache = False <EOL> using_wsgi = True <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" % ( self . host , self . port ) <EOL> def sync_apps ( self ) : <EOL> """<STR_LIT>""" <EOL> cherrypy . server . httpserver . wsgi_app = self . get_app ( ) <EOL> def get_app ( self , app = None ) : <EOL> """<STR_LIT>""" <EOL> if app is None : <EOL> app = cherrypy . tree <EOL> if self . conquer : <EOL> try : <EOL> import wsgiconq <EOL> except ImportError : <EOL> warnings . warn ( "<STR_LIT>" ) <EOL> else : <EOL> app = wsgiconq . WSGILogger ( app , c_calls = True ) <EOL> if self . validate : <EOL> try : <EOL> from wsgiref import validate <EOL> except ImportError : <EOL> warnings . warn ( "<STR_LIT>" ) <EOL> else : <EOL> app = validate . validator ( app ) <EOL> return app <EOL> def get_cpmodpy_supervisor ( ** options ) : <EOL> from cherrypy . test import modpy <EOL> sup = modpy . ModPythonSupervisor ( ** options ) <EOL> sup . template = modpy . conf_cpmodpy <EOL> return sup <EOL> def get_modpygw_supervisor ( ** options ) : <EOL> from cherrypy . test import modpy <EOL> sup = modpy . ModPythonSupervisor ( ** options ) <EOL> sup . template = modpy . conf_modpython_gateway <EOL> sup . using_wsgi = True <EOL> return sup <EOL> def get_modwsgi_supervisor ( ** options ) : <EOL> from cherrypy . test import modwsgi <EOL> return modwsgi . ModWSGISupervisor ( ** options ) <EOL> def get_modfcgid_supervisor ( ** options ) : <EOL> from cherrypy . test import modfcgid <EOL> return modfcgid . ModFCGISupervisor ( ** options ) <EOL> def get_modfastcgi_supervisor ( ** options ) : <EOL> from cherrypy . test import modfastcgi <EOL> return modfastcgi . ModFCGISupervisor ( ** options ) <EOL> def get_wsgi_u_supervisor ( ** options ) : <EOL> cherrypy . server . wsgi_version = ( '<STR_LIT:u>' , <NUM_LIT:0> ) <EOL> return LocalWSGISupervisor ( ** options ) <EOL> class CPWebCase ( webtest . WebCase ) : <EOL> script_name = "<STR_LIT>" <EOL> scheme = "<STR_LIT:http>" <EOL> available_servers = { '<STR_LIT>' : LocalWSGISupervisor , <EOL> '<STR_LIT>' : get_wsgi_u_supervisor , <EOL> '<STR_LIT>' : NativeServerSupervisor , <EOL> '<STR_LIT>' : get_cpmodpy_supervisor , <EOL> '<STR_LIT>' : get_modpygw_supervisor , <EOL> '<STR_LIT>' : get_modwsgi_supervisor , <EOL> '<STR_LIT>' : get_modfcgid_supervisor , <EOL> '<STR_LIT>' : get_modfastcgi_supervisor , <EOL> } <EOL> default_server = "<STR_LIT>" <EOL> def _setup_server ( cls , supervisor , conf ) : <EOL> v = sys . version . split ( ) [ <NUM_LIT:0> ] <EOL> log . info ( "<STR_LIT>" % v ) <EOL> log . info ( "<STR_LIT>" % cherrypy . __version__ ) <EOL> if supervisor . scheme == "<STR_LIT>" : <EOL> ssl = "<STR_LIT>" <EOL> else : <EOL> ssl = "<STR_LIT>" <EOL> log . info ( "<STR_LIT>" % ( supervisor . protocol , ssl ) ) <EOL> log . info ( "<STR_LIT>" % os . getpid ( ) ) <EOL> cherrypy . server . using_apache = supervisor . using_apache <EOL> cherrypy . server . using_wsgi = supervisor . using_wsgi <EOL> if sys . platform [ : <NUM_LIT:4> ] == '<STR_LIT>' : <EOL> cherrypy . config . update ( { '<STR_LIT>' : False } ) <EOL> if isinstance ( conf , basestring ) : <EOL> parser = cherrypy . lib . reprconf . Parser ( ) <EOL> conf = parser . dict_from_file ( conf ) . get ( '<STR_LIT>' , { } ) <EOL> else : <EOL> conf = conf or { } <EOL> baseconf = conf . copy ( ) <EOL> baseconf . update ( { '<STR_LIT>' : supervisor . host , <EOL> '<STR_LIT>' : supervisor . port , <EOL> '<STR_LIT>' : supervisor . protocol , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> } ) <EOL> if supervisor . scheme == "<STR_LIT>" : <EOL> baseconf [ '<STR_LIT>' ] = serverpem <EOL> baseconf [ '<STR_LIT>' ] = serverpem <EOL> if supervisor . scheme == "<STR_LIT>" : <EOL> webtest . WebCase . HTTP_CONN = HTTPSConnection <EOL> return baseconf <EOL> _setup_server = classmethod ( _setup_server ) <EOL> def setup_class ( cls ) : <EOL> '<STR_LIT>' <EOL> conf = get_tst_config ( ) <EOL> supervisor_factory = cls . available_servers . get ( conf . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> if supervisor_factory is None : <EOL> raise RuntimeError ( '<STR_LIT>' % conf [ '<STR_LIT>' ] ) <EOL> supervisor = supervisor_factory ( ** conf ) <EOL> cherrypy . config . reset ( ) <EOL> baseconf = cls . _setup_server ( supervisor , conf ) <EOL> cherrypy . config . update ( baseconf ) <EOL> setup_client ( ) <EOL> if hasattr ( cls , '<STR_LIT>' ) : <EOL> cherrypy . tree = cherrypy . _cptree . Tree ( ) <EOL> cherrypy . server . httpserver = None <EOL> cls . setup_server ( ) <EOL> cherrypy . tree . mount ( gctools . GCRoot ( ) , '<STR_LIT>' ) <EOL> cls . do_gc_test = True <EOL> supervisor . start ( cls . __module__ ) <EOL> cls . supervisor = supervisor <EOL> setup_class = classmethod ( setup_class ) <EOL> def teardown_class ( cls ) : <EOL> '<STR_LIT>' <EOL> if hasattr ( cls , '<STR_LIT>' ) : <EOL> cls . supervisor . stop ( ) <EOL> teardown_class = classmethod ( teardown_class ) <EOL> do_gc_test = False <EOL> def test_gc ( self ) : <EOL> if self . do_gc_test : <EOL> self . getPage ( "<STR_LIT>" ) <EOL> self . assertBody ( "<STR_LIT>" ) <EOL> test_gc . compat_co_firstlineno = getattr ( sys , '<STR_LIT>' , None ) or float ( '<STR_LIT>' ) <EOL> def prefix ( self ) : <EOL> return self . script_name . rstrip ( "<STR_LIT:/>" ) <EOL> def base ( self ) : <EOL> if ( ( self . scheme == "<STR_LIT:http>" and self . PORT == <NUM_LIT> ) or <EOL> ( self . scheme == "<STR_LIT>" and self . PORT == <NUM_LIT> ) ) : <EOL> port = "<STR_LIT>" <EOL> else : <EOL> port = "<STR_LIT>" % self . PORT <EOL> return "<STR_LIT>" % ( self . scheme , self . HOST , port , <EOL> self . script_name . rstrip ( "<STR_LIT:/>" ) ) <EOL> def exit ( self ) : <EOL> sys . exit ( ) <EOL> def getPage ( self , url , headers = None , method = "<STR_LIT:GET>" , body = None , protocol = None ) : <EOL> """<STR_LIT>""" <EOL> if self . script_name : <EOL> url = httputil . urljoin ( self . script_name , url ) <EOL> return webtest . WebCase . getPage ( self , url , headers , method , body , protocol ) <EOL> def skip ( self , msg = '<STR_LIT>' ) : <EOL> raise nose . SkipTest ( msg ) <EOL> def assertErrorPage ( self , status , message = None , pattern = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> page = cherrypy . _cperror . get_error_page ( status , message = message ) <EOL> esc = re . escape <EOL> epage = esc ( page ) <EOL> epage = epage . replace ( esc ( '<STR_LIT>' ) , <EOL> esc ( '<STR_LIT>' ) + '<STR_LIT>' + esc ( '<STR_LIT>' ) ) <EOL> m = re . match ( ntob ( epage , self . encoding ) , self . body , re . DOTALL ) <EOL> if not m : <EOL> self . _handlewebError ( '<STR_LIT>' + page ) <EOL> return <EOL> if pattern is None : <EOL> if m and m . group ( <NUM_LIT:1> ) : <EOL> self . _handlewebError ( '<STR_LIT>' ) <EOL> else : <EOL> if ( m is None ) or ( <EOL> not re . search ( ntob ( re . escape ( pattern ) , self . encoding ) , <EOL> m . group ( <NUM_LIT:1> ) ) ) : <EOL> msg = '<STR_LIT>' <EOL> self . _handlewebError ( msg % repr ( pattern ) ) <EOL> date_tolerance = <NUM_LIT:2> <EOL> def assertEqualDates ( self , dt1 , dt2 , seconds = None ) : <EOL> """<STR_LIT>""" <EOL> if seconds is None : <EOL> seconds = self . date_tolerance <EOL> if dt1 > dt2 : <EOL> diff = dt1 - dt2 <EOL> else : <EOL> diff = dt2 - dt1 <EOL> if not diff < datetime . timedelta ( seconds = seconds ) : <EOL> raise AssertionError ( '<STR_LIT>' % <EOL> ( dt1 , dt2 , seconds ) ) <EOL> def setup_client ( ) : <EOL> """<STR_LIT>""" <EOL> webtest . WebCase . PORT = cherrypy . server . socket_port <EOL> webtest . WebCase . HOST = cherrypy . server . socket_host <EOL> if cherrypy . server . ssl_certificate : <EOL> CPWebCase . scheme = '<STR_LIT>' <EOL> class CPProcess ( object ) : <EOL> pid_file = os . path . join ( thisdir , '<STR_LIT>' ) <EOL> config_file = os . path . join ( thisdir , '<STR_LIT>' ) <EOL> config_template = """<STR_LIT>""" <EOL> error_log = os . path . join ( thisdir , '<STR_LIT>' ) <EOL> access_log = os . path . join ( thisdir , '<STR_LIT>' ) <EOL> def __init__ ( self , wait = False , daemonize = False , ssl = False , socket_host = None , socket_port = None ) : <EOL> self . wait = wait <EOL> self . daemonize = daemonize <EOL> self . ssl = ssl <EOL> self . host = socket_host or cherrypy . server . socket_host <EOL> self . port = socket_port or cherrypy . server . socket_port <EOL> def write_conf ( self , extra = "<STR_LIT>" ) : <EOL> if self . ssl : <EOL> serverpem = os . path . join ( thisdir , '<STR_LIT>' ) <EOL> ssl = """<STR_LIT>""" % ( serverpem , serverpem ) <EOL> else : <EOL> ssl = "<STR_LIT>" <EOL> conf = self . config_template % { <EOL> '<STR_LIT:host>' : self . host , <EOL> '<STR_LIT:port>' : self . port , <EOL> '<STR_LIT>' : self . error_log , <EOL> '<STR_LIT>' : self . access_log , <EOL> '<STR_LIT>' : ssl , <EOL> '<STR_LIT>' : extra , <EOL> } <EOL> f = open ( self . config_file , '<STR_LIT:wb>' ) <EOL> f . write ( ntob ( conf , '<STR_LIT:utf-8>' ) ) <EOL> f . close ( ) <EOL> def start ( self , imports = None ) : <EOL> """<STR_LIT>""" <EOL> cherrypy . _cpserver . wait_for_free_port ( self . host , self . port ) <EOL> args = [ sys . executable , os . path . join ( thisdir , '<STR_LIT:..>' , '<STR_LIT>' ) , <EOL> '<STR_LIT:-c>' , self . config_file , '<STR_LIT>' , self . pid_file ] <EOL> if not isinstance ( imports , ( list , tuple ) ) : <EOL> imports = [ imports ] <EOL> for i in imports : <EOL> if i : <EOL> args . append ( '<STR_LIT>' ) <EOL> args . append ( i ) <EOL> if self . daemonize : <EOL> args . append ( '<STR_LIT>' ) <EOL> env = os . environ . copy ( ) <EOL> grandparentdir = os . path . abspath ( os . path . join ( thisdir , '<STR_LIT:..>' , '<STR_LIT:..>' ) ) <EOL> if env . get ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> env [ '<STR_LIT>' ] = os . pathsep . join ( ( grandparentdir , env [ '<STR_LIT>' ] ) ) <EOL> else : <EOL> env [ '<STR_LIT>' ] = grandparentdir <EOL> if self . wait : <EOL> self . exit_code = os . spawnve ( os . P_WAIT , sys . executable , args , env ) <EOL> else : <EOL> os . spawnve ( os . P_NOWAIT , sys . executable , args , env ) <EOL> cherrypy . _cpserver . wait_for_occupied_port ( self . host , self . port ) <EOL> if self . daemonize : <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> else : <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> def get_pid ( self ) : <EOL> return int ( open ( self . pid_file , '<STR_LIT:rb>' ) . read ( ) ) <EOL> def join ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> try : <EOL> os . wait ( ) <EOL> except AttributeError : <EOL> try : <EOL> pid = self . get_pid ( ) <EOL> except IOError : <EOL> pass <EOL> else : <EOL> os . waitpid ( pid , <NUM_LIT:0> ) <EOL> except OSError : <EOL> x = sys . exc_info ( ) [ <NUM_LIT:1> ] <EOL> if x . args != ( <NUM_LIT:10> , '<STR_LIT>' ) : <EOL> raise </s>
