test_ssh.py

# -*- coding: utf-8 -*-
importmock
fromStringIOimportStringIO
fromcontextlibimportcontextmanager
fromparamikoimportSSHClient
importapp.utils.SshUtilasssh
from . importRSA_PRIVATE_KEY


@contextmanager
defpatch_ssh(stdout='', stderr=''):
defmock_exec_command(*args, **kwargs):
stdin=StringIO()
stdin.close()
returnstdin, StringIO(stdout), StringIO(stderr)
withmock.patch.object(SSHClient, 'exec_command', new=mock_exec_command):
withmock.patch.object(SSHClient, 'connect'):
yield


deftest_ssh():
kwargs=dict(
ip='127.0.0.1',
port='22',
account='root',
pkey=RSA_PRIVATE_KEY,
shell='echo hello\n echo hello'
 )
withpatch_ssh('', ''):
success, log=ssh.do_ssh_cmd(**kwargs)
assertsuccess
withpatch_ssh('', 'message'):
success, log=ssh.do_ssh_cmd(**kwargs)
assertnotsuccess
assert'message'inlog
msg= ('fatal: Not a git repository '
'(or any parent up to mount point /tmp)\n'
'Stopping at filesystem boundary '
'(GIT_DISCOVERY_ACROSS_FILESYSTEM not set).')
withpatch_ssh(msg, ''):
success, log=ssh.do_ssh_cmd(**kwargs)
assertnotsuccess
assertmsginlog