test_host_contact.py 文件源码

def _try_ssh_cmd(self, agent_ssh, auth_args, cmd):
        from chroma_core.services.job_scheduler.agent_rpc import AgentException

        try:
            return agent_ssh.ssh(cmd, auth_args = auth_args)
        except (AuthenticationException, SSHException):
            raise
        except Exception, e:
            # Re-raise wrapped in an AgentException
            raise AgentException(agent_ssh.address,
                                 "Unhandled exception: %s" % e,
                                 ", ".join(auth_args),
                                 '\n'.join(traceback.format_exception(*(sys.exc_info()))))