<s> """<STR_LIT>""" <EOL> from __future__ import print_function <EOL> import os <EOL> import tempfile <EOL> import time <EOL> from subprocess import Popen , PIPE , STDOUT <EOL> import nose <EOL> from IPython . paths import get_ipython_dir <EOL> from ipyparallel import Client , error <EOL> from ipyparallel . apps . launcher import ( LocalProcessLauncher , <EOL> ipengine_cmd_argv , <EOL> ipcontroller_cmd_argv , <EOL> SIGKILL , <EOL> ProcessStateError , <EOL> ) <EOL> launchers = [ ] <EOL> blackhole = open ( os . devnull , '<STR_LIT:w>' ) <EOL> class TestProcessLauncher ( LocalProcessLauncher ) : <EOL> """<STR_LIT>""" <EOL> def start ( self ) : <EOL> if self . state == '<STR_LIT>' : <EOL> self . process = Popen ( self . args , <EOL> stdout = nose . iptest_stdstreams_fileno ( ) , stderr = STDOUT , <EOL> env = os . environ , <EOL> cwd = self . work_dir <EOL> ) <EOL> self . notify_start ( self . process . pid ) <EOL> self . poll = self . process . poll <EOL> else : <EOL> s = '<STR_LIT>' % self . state <EOL> raise ProcessStateError ( s ) <EOL> def setup ( ) : <EOL> class RemoteErrorWithTB ( error . RemoteError ) : <EOL> def __str__ ( self ) : <EOL> s = super ( RemoteErrorWithTB , self ) . __str__ ( ) <EOL> return '<STR_LIT:\n>' . join ( [ s , self . traceback or '<STR_LIT>' ] ) <EOL> error . RemoteError = RemoteErrorWithTB <EOL> cluster_dir = os . path . join ( get_ipython_dir ( ) , '<STR_LIT>' ) <EOL> engine_json = os . path . join ( cluster_dir , '<STR_LIT>' , '<STR_LIT>' ) <EOL> client_json = os . path . join ( cluster_dir , '<STR_LIT>' , '<STR_LIT>' ) <EOL> for json in ( engine_json , client_json ) : <EOL> if os . path . exists ( json ) : <EOL> os . remove ( json ) <EOL> cp = TestProcessLauncher ( ) <EOL> cp . cmd_and_args = ipcontroller_cmd_argv + [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> cp . start ( ) <EOL> launchers . append ( cp ) <EOL> tic = time . time ( ) <EOL> while not os . path . exists ( engine_json ) or not os . path . exists ( client_json ) : <EOL> if cp . poll ( ) is not None : <EOL> raise RuntimeError ( "<STR_LIT>" % cp . poll ( ) ) <EOL> elif time . time ( ) - tic > <NUM_LIT:15> : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> add_engines ( <NUM_LIT:1> ) <EOL> def add_engines ( n = <NUM_LIT:1> , profile = '<STR_LIT>' , total = False ) : <EOL> """<STR_LIT>""" <EOL> rc = Client ( profile = profile ) <EOL> base = len ( rc ) <EOL> if total : <EOL> n = max ( n - base , <NUM_LIT:0> ) <EOL> eps = [ ] <EOL> for i in range ( n ) : <EOL> ep = TestProcessLauncher ( ) <EOL> ep . cmd_and_args = ipengine_cmd_argv + [ <EOL> '<STR_LIT>' % profile , <EOL> '<STR_LIT>' <EOL> ] <EOL> ep . start ( ) <EOL> launchers . append ( ep ) <EOL> eps . append ( ep ) <EOL> tic = time . time ( ) <EOL> while len ( rc ) < base + n : <EOL> if any ( [ ep . poll ( ) is not None for ep in eps ] ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> elif time . time ( ) - tic > <NUM_LIT:15> : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> rc . close ( ) <EOL> return eps <EOL> def teardown ( ) : <EOL> try : <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> except KeyboardInterrupt : <EOL> return <EOL> while launchers : <EOL> p = launchers . pop ( ) <EOL> if p . poll ( ) is None : <EOL> try : <EOL> p . stop ( ) <EOL> except Exception as e : <EOL> print ( e ) <EOL> pass <EOL> if p . poll ( ) is None : <EOL> try : <EOL> time . sleep ( <NUM_LIT> ) <EOL> except KeyboardInterrupt : <EOL> return <EOL> if p . poll ( ) is None : <EOL> try : <EOL> print ( '<STR_LIT>' ) <EOL> p . signal ( SIGKILL ) <EOL> except : <EOL> print ( "<STR_LIT>" , p ) <EOL> blackhole . close ( ) </s>
