<s> import logging <EOL> import multiprocessing <EOL> import select <EOL> import unittest <EOL> import collections <EOL> import os <EOL> import sys <EOL> import six <EOL> import multiprocessing . connection as connection <EOL> from nose2 import events , loader , result , runner , session , util <EOL> log = logging . getLogger ( __name__ ) <EOL> class MultiProcess ( events . Plugin ) : <EOL> configSection = '<STR_LIT>' <EOL> def __init__ ( self ) : <EOL> self . addArgument ( self . setProcs , '<STR_LIT:N>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . testRunTimeout = self . config . as_float ( '<STR_LIT>' , <NUM_LIT> ) <EOL> self . procs = self . config . as_int ( <EOL> '<STR_LIT>' , multiprocessing . cpu_count ( ) ) <EOL> self . setAddress ( self . config . as_str ( '<STR_LIT>' , None ) ) <EOL> self . cases = { } <EOL> def setProcs ( self , num ) : <EOL> self . procs = int ( num [ <NUM_LIT:0> ] ) <EOL> self . register ( ) <EOL> def setAddress ( self , address ) : <EOL> if address is None or address . strip ( ) == '<STR_LIT>' : <EOL> address = [ ] <EOL> else : <EOL> address = [ x . strip ( ) for x in address . split ( '<STR_LIT::>' ) [ : <NUM_LIT:2> ] ] <EOL> self . bind_host = None <EOL> self . bind_port = <NUM_LIT:0> <EOL> if sys . platform == "<STR_LIT:win32>" or address : <EOL> self . bind_host = '<STR_LIT>' <EOL> if address and address [ <NUM_LIT:0> ] : <EOL> self . bind_host = address [ <NUM_LIT:0> ] <EOL> self . bind_port = <NUM_LIT:0> <EOL> if len ( address ) >= <NUM_LIT:2> : <EOL> self . bind_port = int ( address [ <NUM_LIT:1> ] ) <EOL> def pluginsLoaded ( self , event ) : <EOL> self . addMethods ( '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> def startTestRun ( self , event ) : <EOL> event . executeTests = self . _runmp <EOL> def beforeInteraction ( self , event ) : <EOL> event . handled = True <EOL> return False <EOL> def _runmp ( self , test , result ) : <EOL> flat = list ( self . _flatten ( test ) ) <EOL> procs = self . _startProcs ( len ( flat ) ) <EOL> for proc , conn in procs : <EOL> if not flat : <EOL> break <EOL> caseid = flat . pop ( <NUM_LIT:0> ) <EOL> conn . send ( caseid ) <EOL> rdrs = [ conn for proc , conn in procs if proc . is_alive ( ) ] <EOL> while flat or rdrs : <EOL> ready , _ , _ = select . select ( rdrs , [ ] , [ ] , self . testRunTimeout ) <EOL> for conn in ready : <EOL> try : <EOL> remote_events = conn . recv ( ) <EOL> except EOFError : <EOL> log . warning ( "<STR_LIT>" ) <EOL> continue <EOL> if remote_events is None : <EOL> log . debug ( "<STR_LIT>" , conn ) <EOL> rdrs . remove ( conn ) <EOL> continue <EOL> testid , events = remote_events <EOL> log . debug ( "<STR_LIT>" , testid ) <EOL> for ( hook , event ) in events : <EOL> log . debug ( "<STR_LIT>" , hook , event ) <EOL> self . _localize ( event ) <EOL> getattr ( self . session . hooks , hook ) ( event ) <EOL> if not flat : <EOL> conn . send ( None ) <EOL> continue <EOL> caseid = flat . pop ( <NUM_LIT:0> ) <EOL> conn . send ( caseid ) <EOL> for _ , conn in procs : <EOL> conn . close ( ) <EOL> for proc , _ in procs : <EOL> proc . join ( ) <EOL> def _prepConns ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . bind_host is not None : <EOL> authkey = os . urandom ( <NUM_LIT:20> ) <EOL> address = ( self . bind_host , self . bind_port ) <EOL> listener = connection . Listener ( address , authkey = authkey ) <EOL> return ( listener , listener . address + ( authkey , ) ) <EOL> else : <EOL> return multiprocessing . Pipe ( ) <EOL> def _acceptConns ( self , parent_conn ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( parent_conn , connection . Listener ) : <EOL> rdrs = [ parent_conn . _listener . _socket ] <EOL> readable , _ , _ = select . select ( rdrs , [ ] , [ ] , <EOL> self . testRunTimeout ) <EOL> if readable : <EOL> return parent_conn . accept ( ) <EOL> else : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> else : <EOL> return parent_conn <EOL> def _startProcs ( self , test_count ) : <EOL> session_export = self . _exportSession ( ) <EOL> procs = [ ] <EOL> count = min ( test_count , self . procs ) <EOL> log . debug ( "<STR_LIT>" , count ) <EOL> for i in range ( <NUM_LIT:0> , count ) : <EOL> parent_conn , child_conn = self . _prepConns ( ) <EOL> proc = multiprocessing . Process ( <EOL> target = procserver , args = ( session_export , child_conn ) ) <EOL> proc . daemon = True <EOL> proc . start ( ) <EOL> parent_conn = self . _acceptConns ( parent_conn ) <EOL> procs . append ( ( proc , parent_conn ) ) <EOL> return procs <EOL> def _flatten ( self , suite ) : <EOL> log . debug ( "<STR_LIT>" ) <EOL> mods = { } <EOL> classes = { } <EOL> stack = [ suite ] <EOL> while stack : <EOL> suite = stack . pop ( ) <EOL> for test in suite : <EOL> if isinstance ( test , unittest . TestSuite ) : <EOL> stack . append ( test ) <EOL> else : <EOL> testid = util . test_name ( test ) <EOL> self . cases [ testid ] = test <EOL> if util . has_module_fixtures ( test ) : <EOL> mods . setdefault ( test . __class__ . __module__ , [ ] ) . append ( <EOL> testid ) <EOL> elif util . has_class_fixtures ( test ) : <EOL> classes . setdefault ( <EOL> "<STR_LIT>" % ( test . __class__ . __module__ , <EOL> test . __class__ . __name__ ) , <EOL> [ ] ) . append ( testid ) <EOL> else : <EOL> yield testid <EOL> for cls in sorted ( classes . keys ( ) ) : <EOL> yield cls <EOL> for mod in sorted ( mods . keys ( ) ) : <EOL> yield mod <EOL> def _localize ( self , event ) : <EOL> if hasattr ( event , '<STR_LIT:result>' ) : <EOL> event . result = self . session . testResult <EOL> if hasattr ( event , '<STR_LIT>' ) : <EOL> event . loader = self . session . testLoader <EOL> if hasattr ( event , '<STR_LIT>' ) : <EOL> event . runner = self . session . testRunner <EOL> if hasattr ( event , '<STR_LIT:test>' ) and isinstance ( event . test , six . string_types ) : <EOL> try : <EOL> event . test = self . cases [ event . test ] <EOL> except KeyError : <EOL> event . test = self . session . testLoader . failedLoadTests ( <EOL> '<STR_LIT>' , <EOL> RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" % event . test ) ) . _tests [ <NUM_LIT:0> ] <EOL> def _exportSession ( self ) : <EOL> export = { '<STR_LIT>' : self . session . config , <EOL> '<STR_LIT>' : self . session . verbosity , <EOL> '<STR_LIT>' : self . session . startDir , <EOL> '<STR_LIT>' : self . session . topLevelDir , <EOL> '<STR_LIT>' : self . session . logLevel , <EOL> '<STR_LIT>' : [ ] } <EOL> event = RegisterInSubprocessEvent ( ) <EOL> self . session . hooks . registerInSubprocess ( event ) <EOL> export [ '<STR_LIT>' ] . extend ( event . pluginClasses ) <EOL> return export <EOL> def procserver ( session_export , conn ) : <EOL> rlog = multiprocessing . log_to_stderr ( ) <EOL> rlog . setLevel ( session_export [ '<STR_LIT>' ] ) <EOL> ssn = session . Session ( ) <EOL> ssn . config = session_export [ '<STR_LIT>' ] <EOL> ssn . hooks = RecordingPluginInterface ( ) <EOL> ssn . verbosity = session_export [ '<STR_LIT>' ] <EOL> ssn . startDir = session_export [ '<STR_LIT>' ] <EOL> ssn . topLevelDir = session_export [ '<STR_LIT>' ] <EOL> ssn . prepareSysPath ( ) <EOL> loader_ = loader . PluggableTestLoader ( ssn ) <EOL> ssn . testLoader = loader_ <EOL> result_ = result . PluggableTestResult ( ssn ) <EOL> ssn . testResult = result_ <EOL> runner_ = runner . PluggableTestRunner ( ssn ) <EOL> ssn . testRunner = runner_ <EOL> ssn . plugins = [ <EOL> plugin ( session = ssn ) for plugin in session_export [ '<STR_LIT>' ] ] <EOL> rlog . debug ( "<STR_LIT>" , ssn . plugins ) <EOL> for plugin in ssn . plugins : <EOL> plugin . register ( ) <EOL> rlog . debug ( "<STR_LIT>" , plugin ) <EOL> if isinstance ( conn , collections . Sequence ) : <EOL> conn = connection . Client ( conn [ : <NUM_LIT:2> ] , authkey = conn [ <NUM_LIT:2> ] ) <EOL> event = SubprocessEvent ( loader_ , result_ , runner_ , ssn . plugins , conn ) <EOL> res = ssn . hooks . startSubprocess ( event ) <EOL> if event . handled and not res : <EOL> conn . send ( None ) <EOL> conn . close ( ) <EOL> ssn . hooks . stopSubprocess ( event ) <EOL> return <EOL> executor = event . executeTests <EOL> for testid in gentests ( conn ) : <EOL> if testid is None : <EOL> break <EOL> test = event . loader . loadTestsFromName ( testid ) <EOL> rlog . debug ( "<STR_LIT>" , testid , test ) <EOL> executor ( test , event . result ) <EOL> events = [ e for e in ssn . hooks . flush ( ) ] <EOL> conn . send ( ( testid , events ) ) <EOL> rlog . debug ( "<STR_LIT>" , testid ) <EOL> conn . send ( None ) <EOL> conn . close ( ) <EOL> ssn . hooks . stopSubprocess ( event ) <EOL> def gentests ( conn ) : <EOL> while True : <EOL> try : <EOL> testid = conn . recv ( ) <EOL> if testid is None : <EOL> return <EOL> yield testid <EOL> except EOFError : <EOL> return <EOL> class SubprocessEvent ( events . Event ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , loader , result , runner , plugins , connection , ** metadata ) : <EOL> self . loader = loader <EOL> self . result = result <EOL> self . runner = runner <EOL> self . plugins = plugins <EOL> self . connection = connection <EOL> self . executeTests = lambda test , result : test ( result ) <EOL> super ( SubprocessEvent , self ) . __init__ ( ** metadata ) <EOL> class RegisterInSubprocessEvent ( events . Event ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , ** metadata ) : <EOL> self . pluginClasses = [ ] <EOL> super ( RegisterInSubprocessEvent , self ) . __init__ ( ** metadata ) <EOL> class RecordingHook ( events . Hook ) : <EOL> def __init__ ( self , method , interface ) : <EOL> super ( RecordingHook , self ) . __init__ ( method ) <EOL> self . interface = interface <EOL> def __call__ ( self , event ) : <EOL> res = super ( RecordingHook , self ) . __call__ ( event ) <EOL> self . interface . log ( self . method , event ) <EOL> return res <EOL> class RecordingPluginInterface ( events . PluginInterface ) : <EOL> hookClass = RecordingHook <EOL> noLogMethods = set ( <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def __init__ ( self ) : <EOL> super ( RecordingPluginInterface , self ) . __init__ ( ) <EOL> self . events = [ ] <EOL> def log ( self , method , event ) : <EOL> self . events . append ( ( method , event ) ) <EOL> def flush ( self ) : <EOL> events = self . events [ : ] <EOL> self . events = [ ] <EOL> return events <EOL> def register ( self , method , plugin ) : <EOL> """<STR_LIT>""" <EOL> self . _hookForMethod ( method ) . append ( plugin ) <EOL> def __getattr__ ( self , attr ) : <EOL> if attr . startswith ( '<STR_LIT>' ) : <EOL> raise AttributeError ( '<STR_LIT>' % ( attr , self ) ) <EOL> return self . _hookForMethod ( attr ) <EOL> def _hookForMethod ( self , method ) : <EOL> try : <EOL> return self . hooks [ method ] <EOL> except KeyError : <EOL> if method in self . noLogMethods or method . startswith ( '<STR_LIT>' ) : <EOL> hook = events . Hook ( method ) <EOL> else : <EOL> hook = self . hookClass ( method , self ) <EOL> self . hooks [ method ] = hook <EOL> return hook </s>
