diff --git a/requirements.txt b/requirements.txt index 7682038fce17861ca1ee658e5c93583fc383f464..40fa0564e4dee9cc5183b79ceca2abc1833c62c5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -12,6 +12,7 @@ pandas tables future paramiko +sshtunnel psutil pbr PyScaffold diff --git a/wetb/utils/tests/test_ssh_client.py b/wetb/utils/tests/test_ssh_client.py index b031d7a5db37deeb64ba5f546b6f3a4fc73f0be5..567e763689e752a769ea0efbf0cd051429ac5aee 100644 --- a/wetb/utils/tests/test_ssh_client.py +++ b/wetb/utils/tests/test_ssh_client.py @@ -106,89 +106,37 @@ class TestSSHClient(unittest.TestCase): self.assertEqual(len(os.listdir(tfp+"tmp/input/data/")),2) shutil.rmtree(tfp + "tmp/") -# def test_ssh_gorm(self): -# if x: -# ssh = SSHClient('gorm.risoe.dk', 'mmpe', x.mmpe) -# _,out,_ = ssh.execute("hostname") -# self.assertEqual(out.strip(), "g-000.risoe.dk") - -# def test_ssh_g047(self): -# if x: -# gateway = SSHClient('gorm.risoe.dk', 'mmpe', x.mmpe) -# ssh = SSHClient('g-047', "mmpe", x.mmpe, gateway=gateway) -# self.assertEqual(ssh.execute('hostname')[1].strip(), "g-047") + def test_ssh_gorm(self): + if 0 or all: + if x: + ssh = SSHClient('gorm.risoe.dk', 'mmpe', x.mmpe) + _,out,_ = ssh.execute("hostname") + self.assertEqual(out.strip(), "g-000.risoe.dk") + + def test_ssh_g047(self): + if 1 or all: + if x: + gateway = SSHClient('gorm.risoe.dk', 'mmpe', x.mmpe) + ssh = SSHClient('g-047', "mmpe", x.mmpe, gateway=gateway) + self.assertEqual(ssh.execute('hostname')[1].strip(), "g-047") -# def test_ssh_risoe(self): -# if x: -# -# ssh = SSHClient('ssh.risoe.dk', 'mmpe', interactive_auth_handler = sshrisoe_interactive_auth_handler(x.mmpe)) -# _,out,_ = ssh.execute("hostname") -# self.assertEqual(out.strip(), "ssh-03.risoe.dk") + def test_ssh_risoe(self): + if 0 or all: + if x: + ssh = SSHClient('ssh.risoe.dk', 'mmpe', interactive_auth_handler = sshrisoe_interactive_auth_handler(x.mmpe)) + _,out,_ = ssh.execute("hostname") + self.assertEqual(out.strip(), "ssh-03.risoe.dk") def test_ssh_risoe_gorm(self): - if x: - - gateway = SSHClient('ssh.risoe.dk', 'mmpe', interactive_auth_handler = sshrisoe_interactive_auth_handler(x.mmpe)) - ssh = SSHClient('10.40.23.49', 'mmpe', x.mmpe, gateway = gateway) - _,out,_ = ssh.execute("hostname") - self.assertEqual(out.strip(), "g-000.risoe.dk") + if 0 or all: + if x: + + gateway = SSHClient('ssh.risoe.dk', 'mmpe', interactive_auth_handler = sshrisoe_interactive_auth_handler(x.mmpe)) + ssh = SSHClient('10.40.23.49', 'mmpe', x.mmpe, gateway = gateway) + _,out,_ = ssh.execute("hostname") + self.assertEqual(out.strip(), "g-000.risoe.dk") - -# def test_ssh_risoe(self): -# #logger = logging.getLogger("paramiko") -# #logger.setLevel(logging.DEBUG) # for example -# #ch = logging.StreamHandler(sys.stdout) -# #ch.setLevel(logging.DEBUG) -# #formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') -# #ch.setFormatter(formatter) -# #logger.addHandler(ch) -# ssh = SSHClient('ssh.risoe.dk', 'mmpe') -# print (ssh.connect()) -# return -# username = "mmpe" -# -# client = paramiko.client.SSHClient() -# -# # Any means of getting the PKey will do. This code assumes you've only got one key loaded in your active ssh-agent. -# # See also: -# # - http://docs.paramiko.org/en/1.17/api/keys.html#paramiko.pkey.PKey -# # - http://docs.paramiko.org/en/1.17/api/client.html#paramiko.client.SSHClient.connect -# my_pkey = None #paramiko.agent.Agent().get_keys()[0] -# -# try: -# client.connect( -# hostname="ssh.risoe.dk", -# port=22, -# username=username, -# look_for_keys=False, -# pkey=my_pkey -# ) -# except paramiko.ssh_exception.SSHException as e: -# pass -# -# transport = client.get_transport() -# -# # Sometimes sshd is configured to use 'keyboard-interactive' instead of 'password' to implement the YubiKey challenge. -# # In that case, you can use something like this. -# # The code below assumes the server will only ask one question and expect the YubiKey OTP as an answer. -# # If there's more questions to answer, you should handle those per the docs at: -# # http://docs.paramiko.org/en/1.17/api/transport.html#paramiko.transport.Transport.auth_interactive -# # -# def interactive_handler(title, instructions, prompt_list): -# if prompt_list: -# if prompt_list[0][0]=="AD Password: ": -# return [x.mmpe] -# return [getpass.getpass(prompt_list[0][0])] -# print ("here") -# return [] -# transport.auth_interactive(username, interactive_handler) -# -# #transport.auth_password(username, x.mmpe) -# -# # You should now be able to use client as the authenticated user. -# client.exec_command("echo hej") -# if __name__ == "__main__": #import sys;sys.argv = ['', 'Test.testName'] unittest.main() \ No newline at end of file