#!/usr/bin/env python
-#-*- coding:utf-8 -*-
+# -*- coding:utf-8 -*-
# Copyright (C) 2010-2018 CEA/DEN
#
verbose = False
+
class TestCase(unittest.TestCase):
- "Test a paramiko connection"""
-
- def setLoggerParamiko(self):
- """to get logs of paramiko, useful if problems"""
- import logging as LOGI
- loggerPrmk = LOGI.getLogger("paramiko")
- if len(loggerPrmk.handlers) != 0:
- print("logging.__file__ %s" % LOGI.__file__)
- print("logger paramiko have handler set yet, is a surprise")
- return
- if not verbose:
- # stay as it, null
- return
-
- #set a paramiko logger verbose
- handler = LOGI.StreamHandler()
- msg = "create paramiko logger, with handler on stdout"
-
- # handler = LOGI.MemoryHandler()
- # etc... https://docs.python.org/2/library/logging.handlers.html
- # msg = "create paramiko logger, with handler in memory"
-
- # original frm from paramiko
- # frm = '%(levelname)-.3s [%(asctime)s.%(msecs)03d] thr=%(thread)-3d %(name)s: %(message)s' # noqa
- frm = '%(levelname)-5s :: %(asctime)s :: %(name)s :: %(message)s'
- handler.setFormatter(LOGI.Formatter(frm, '%y%m%d_%H%M%S'))
- loggerPrmk.addHandler(handler)
-
- # logger is not notset but low, handlers needs setlevel greater
- loggerPrmk.setLevel(LOGI.DEBUG)
- handler.setLevel(LOGI.INFO) # LOGI.DEBUG) # may be other one
-
- loggerPrmk.info(msg)
-
-
- '''example from internet
+ "Test a paramiko connection" ""
+
+ def setLoggerParamiko(self):
+ """to get logs of paramiko, useful if problems"""
+ import logging as LOGI
+
+ loggerPrmk = LOGI.getLogger("paramiko")
+ if len(loggerPrmk.handlers) != 0:
+ print("logging.__file__ %s" % LOGI.__file__)
+ print("logger paramiko have handler set yet, is a surprise")
+ return
+ if not verbose:
+ # stay as it, null
+ return
+
+ # set a paramiko logger verbose
+ handler = LOGI.StreamHandler()
+ msg = "create paramiko logger, with handler on stdout"
+
+ # handler = LOGI.MemoryHandler()
+ # etc... https://docs.python.org/2/library/logging.handlers.html
+ # msg = "create paramiko logger, with handler in memory"
+
+ # original frm from paramiko
+ # frm = '%(levelname)-.3s [%(asctime)s.%(msecs)03d] thr=%(thread)-3d %(name)s: %(message)s' # noqa
+ frm = "%(levelname)-5s :: %(asctime)s :: %(name)s :: %(message)s"
+ handler.setFormatter(LOGI.Formatter(frm, "%y%m%d_%H%M%S"))
+ loggerPrmk.addHandler(handler)
+
+ # logger is not notset but low, handlers needs setlevel greater
+ loggerPrmk.setLevel(LOGI.DEBUG)
+ handler.setLevel(LOGI.INFO) # LOGI.DEBUG) # may be other one
+
+ loggerPrmk.info(msg)
+
+ """example from internet
def fetch_netmask(self, hostname, port=22):
private_key = os.path.expanduser('~/.ssh/id_rsa')
connection = open_ssh_connection('wambeke', hostname, port=port, key=private_key)
client.set_missing_host_key_policy(PK.AutoAddPolicy())
client.connect(hostname, port=port, timeout=5, username=username, key_filename=key)
return client
- '''
-
- def test_000(self):
- self.setLoggerParamiko()
-
-
- def test_010(self):
- # http://docs.paramiko.org/en/2.4/api/agent.html
-
- try:
- import paramiko as PK
- except:
- print("\nproblem 'import paramiko', no tests")
- return
-
- # port=22 # useless
- username = getpass.getuser()
- hostname = os.uname()[1]
- aFile = "/tmp/%s_test_paramiko.tmp" % username
- cmd = ("pwd; ls -alt {0}; cat {0}").format(aFile)
-
- # connect
- client = PK.SSHClient()
- client.set_missing_host_key_policy(PK.AutoAddPolicy())
- # client.connect(hostname, username=username, password="xxxxx")
- # client.connect(hostname, username=username, passphrase="yyyy", key_filename="/home/wambeke/.ssh/id_rsa_satjobs_passphrase")
- # client.connect(hostname, username=username)
-
- # timeout in seconds
- client.connect(hostname, username=username, timeout=1.)
-
- # obtain session
- session = client.get_transport().open_session()
- # Forward local agent
- PK.agent.AgentRequestHandler(session)
- # commands executed after this point will see the forwarded agent on the remote end.
-
- # one api
- session.exec_command("date > %s" % aFile)
- cmd = ("pwd; ls -alt {0}; cat {0} && echo OK").format(aFile)
- # another api
- stdin, stdout, stderr = client.exec_command(cmd)
- output = stdout.read()
- if verbose:
- print('stdout:\n%s' % output)
- self.assertTrue(aFile in output)
- self.assertTrue("OK" in output)
- client.close()
-
-if __name__ == '__main__':
+ """
+
+ def test_000(self):
+ self.setLoggerParamiko()
+
+ def test_010(self):
+ # http://docs.paramiko.org/en/2.4/api/agent.html
+
+ try:
+ import paramiko as PK
+ except:
+ print("\nproblem 'import paramiko', no tests")
+ return
+
+ # port=22 # useless
+ username = getpass.getuser()
+ hostname = os.uname()[1]
+ aFile = "/tmp/%s_test_paramiko.tmp" % username
+ cmd = ("pwd; ls -alt {0}; cat {0}").format(aFile)
+
+ # connect
+ client = PK.SSHClient()
+ client.set_missing_host_key_policy(PK.AutoAddPolicy())
+ # client.connect(hostname, username=username, password="xxxxx")
+ # client.connect(hostname, username=username, passphrase="yyyy", key_filename="/home/wambeke/.ssh/id_rsa_satjobs_passphrase")
+ # client.connect(hostname, username=username)
+
+ # timeout in seconds
+ client.connect(hostname, username=username, timeout=1.0)
+
+ # obtain session
+ session = client.get_transport().open_session()
+ # Forward local agent
+ PK.agent.AgentRequestHandler(session)
+ # commands executed after this point will see the forwarded agent on the remote end.
+
+ # one api
+ session.exec_command("date > %s" % aFile)
+ cmd = ("pwd; ls -alt {0}; cat {0} && echo OK").format(aFile)
+ # another api
+ stdin, stdout, stderr = client.exec_command(cmd)
+ output = stdout.read()
+ if verbose:
+ print("stdout:\n%s" % output)
+ self.assertTrue(aFile in output)
+ self.assertTrue("OK" in output)
+ client.close()
+
+
+if __name__ == "__main__":
# verbose = True # human eyes
unittest.main(exit=False)
pass
