<s> from ooni import nettest <EOL> from ooni . utils import log <EOL> from twisted . internet import defer , protocol , reactor <EOL> from twisted . python import usage <EOL> import os <EOL> def which ( program ) : <EOL> def is_exe ( fpath ) : <EOL> return os . path . isfile ( fpath ) and os . access ( fpath , os . X_OK ) <EOL> fpath , fname = os . path . split ( program ) <EOL> if fpath : <EOL> if is_exe ( program ) : <EOL> return program <EOL> else : <EOL> for path in os . environ [ "<STR_LIT>" ] . split ( os . pathsep ) : <EOL> path = path . strip ( '<STR_LIT:">' ) <EOL> exe_file = os . path . join ( path , program ) <EOL> if is_exe ( exe_file ) : <EOL> return exe_file <EOL> return None <EOL> class UsageOptions ( usage . Options ) : <EOL> optParameters = [ <EOL> [ '<STR_LIT>' , '<STR_LIT:i>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT:s>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ] <EOL> class ScriptProcessProtocol ( protocol . ProcessProtocol ) : <EOL> def __init__ ( self , test_case ) : <EOL> self . test_case = test_case <EOL> self . deferred = defer . Deferred ( ) <EOL> def connectionMade ( self ) : <EOL> log . debug ( "<STR_LIT>" ) <EOL> self . transport . closeStdin ( ) <EOL> self . test_case . report [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> def outReceived ( self , data ) : <EOL> log . debug ( '<STR_LIT>' % data ) <EOL> self . test_case . report [ '<STR_LIT>' ] += data <EOL> def errReceived ( self , data ) : <EOL> log . err ( '<STR_LIT>' % data ) <EOL> self . transport . signalProcess ( '<STR_LIT>' ) <EOL> def processEnded ( self , status ) : <EOL> rc = status . value . exitCode <EOL> log . debug ( '<STR_LIT>' % ( rc , self . test_case . report [ '<STR_LIT>' ] ) ) <EOL> if rc == <NUM_LIT:0> : <EOL> self . deferred . callback ( self ) <EOL> else : <EOL> self . deferred . errback ( rc ) <EOL> class Script ( nettest . NetTestCase ) : <EOL> name = "<STR_LIT>" <EOL> version = "<STR_LIT>" <EOL> authors = "<STR_LIT>" <EOL> usageOptions = UsageOptions <EOL> requiredOptions = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> requiresRoot = False <EOL> requiresTor = False <EOL> def test_run_script ( self ) : <EOL> """<STR_LIT>""" <EOL> processProtocol = ScriptProcessProtocol ( self ) <EOL> interpreter = self . localOptions [ '<STR_LIT>' ] <EOL> if not which ( interpreter ) : <EOL> log . err ( '<STR_LIT>' % interpreter ) <EOL> return <EOL> reactor . spawnProcess ( processProtocol , <EOL> interpreter , <EOL> args = [ interpreter , self . localOptions [ '<STR_LIT>' ] ] , <EOL> env = { '<STR_LIT>' : os . environ [ '<STR_LIT>' ] } , <EOL> usePTY = True ) <EOL> if not reactor . running : <EOL> reactor . run ( ) <EOL> return processProtocol . deferred </s>
