<s> from twisted . internet import defer <EOL> from twisted . trial . unittest import TestCase <EOL> from carnifex . sshprocess import SSHProcessInductor <EOL> from twisted . internet . protocol import ProcessProtocol <EOL> from twisted . internet . error import ProcessTerminated , ProcessDone , ConnectionRefusedError <EOL> from carnifex . localprocess import LocalProcessInductor <EOL> from getpass import getpass <EOL> from carnifex . ssh . client import TooManyAuthFailures <EOL> DEBUG = False <EOL> if DEBUG : <EOL> import sys <EOL> from twisted . python import log <EOL> log . startLogging ( sys . stdout ) <EOL> defer . Deferred . debug = True <EOL> UID = None <EOL> PASSWORD = None or getpass ( ) <EOL> WRONG_USER , WRONG_PASSWORD = '<STR_LIT>' , '<STR_LIT>' <EOL> SUCCEEDING_COMMAND = '<STR_LIT:true>' <EOL> FAILING_COMMAND = '<STR_LIT:false>' <EOL> PYTHON_COMMAND = '<STR_LIT>' <EOL> class InductorTestMixin ( object ) : <EOL> def test_execute_exitcode_0 ( self ) : <EOL> """<STR_LIT>""" <EOL> processEndedDeferred = defer . Deferred ( ) <EOL> protocol = ProcessProtocol ( ) <EOL> protocol . processEnded = processEndedDeferred . callback <EOL> processDeferred = self . inductor . execute ( protocol , SUCCEEDING_COMMAND , <EOL> uid = UID ) <EOL> self . assertFailure ( processEndedDeferred , ProcessDone ) <EOL> return defer . DeferredList ( [ processDeferred , processEndedDeferred ] , <EOL> fireOnOneErrback = True ) <EOL> def test_execute_nonzero_exitcode ( self ) : <EOL> """<STR_LIT>""" <EOL> processEndedDeferred = defer . Deferred ( ) <EOL> protocol = ProcessProtocol ( ) <EOL> protocol . processEnded = processEndedDeferred . callback <EOL> processDeferred = self . inductor . execute ( protocol , FAILING_COMMAND , <EOL> uid = UID ) <EOL> self . assertFailure ( processEndedDeferred , ProcessTerminated ) <EOL> @ processEndedDeferred . addErrback <EOL> def checkExitCode ( failure ) : <EOL> exitCode = failure . value . exitCode <EOL> self . assertNotEqual ( exitCode , <NUM_LIT:0> ) <EOL> return defer . DeferredList ( [ processDeferred , processEndedDeferred ] , <EOL> fireOnOneErrback = True ) <EOL> def test_run_stdout_stderr_exit ( self ) : <EOL> """<STR_LIT>""" <EOL> stdoutText = "<STR_LIT>" <EOL> stderrText = "<STR_LIT>" <EOL> exitCode = <NUM_LIT> <EOL> pythonScript = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> % ( stdoutText , stderrText , exitCode ) ) <EOL> command = "<STR_LIT>" % ( PYTHON_COMMAND , pythonScript ) <EOL> disconnectedDeferred = defer . Deferred ( ) <EOL> protocol = ProcessProtocol ( ) <EOL> protocol . processEnded = disconnectedDeferred . callback <EOL> resultDeferred = self . inductor . run ( command , uid = UID ) <EOL> resultDeferred . addErrback ( self . fail ) <EOL> @ resultDeferred . addCallback <EOL> def checkResult ( ( r_stdoutText , r_stderrText , r_exitCode ) ) : <EOL> self . assertEqual ( r_stdoutText , stdoutText , "<STR_LIT>" ) <EOL> self . assertEqual ( r_stderrText , stderrText , "<STR_LIT>" ) <EOL> self . assertEqual ( r_exitCode , exitCode , "<STR_LIT>" ) <EOL> return resultDeferred <EOL> class LocalProcessInductorTest ( TestCase , InductorTestMixin ) : <EOL> def setUp ( self ) : <EOL> from twisted . internet import reactor <EOL> self . inductor = LocalProcessInductor ( reactor ) <EOL> class SSHProcessInductorTest ( TestCase , InductorTestMixin ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> from twisted . internet import reactor <EOL> host , port = '<STR_LIT:localhost>' , <NUM_LIT> <EOL> self . inductor = SSHProcessInductor ( reactor , host , port ) <EOL> self . inductor . setCredentials ( UID , PASSWORD ) <EOL> def test_authentication_failure ( self ) : <EOL> """<STR_LIT>""" <EOL> self . inductor . setCredentials ( WRONG_USER , WRONG_PASSWORD ) <EOL> protocol = ProcessProtocol ( ) <EOL> processDeferred = self . inductor . execute ( protocol , SUCCEEDING_COMMAND , <EOL> uid = WRONG_USER ) <EOL> return self . assertFailure ( processDeferred , TooManyAuthFailures ) <EOL> def test_connection_refused ( self ) : <EOL> """<STR_LIT>""" <EOL> from twisted . internet import reactor <EOL> host , port = '<STR_LIT:localhost>' , <NUM_LIT:1> <EOL> self . inductor = SSHProcessInductor ( reactor , host , port ) <EOL> deferred = self . inductor . getConnection ( ) <EOL> return self . assertFailure ( deferred , ConnectionRefusedError ) <EOL> def tearDown ( self ) : <EOL> inductor = self . inductor <EOL> self . inductor = None <EOL> inductor . disconnectAll ( ) </s>
