import logging
import paramiko
import traceback
from functools import wraps
from time import perf_counter
import glaciation.testing.generic as tgf
from glaciation.cdu.api import CduAPIWrapper
[docs]def mainTest(func):
"""Decorator for main test function
It will wrap the main test script like this:
| Try:
| tgf.initTest() # log in a file the start of the test
| mainScript # execute main script
| tgf.finishTestOk() # log in a file execution SUCCESS
| except Exception as e:
| print(traceback.format_exc()) # print traceback
| tgf.finishTestNotOK(str(e)) # log in a file execution FAILED
| exit(-1)
"""
@wraps(func)
def wrapper(*args, **kwargs):
try:
g = func.__globals__ # use f.func_globals for py < 2.6
obj = object()
TESTER = g.get('TESTER', obj)
PROGNAME = g.get('PROGNAME', obj)
DESCRIPTION = g.get('DESCRIPTION', obj)
LOG_FILE_PATH = g.get('LOG_FILE_PATH', obj)
PROGNAME,TESTER,LOG_FILE_PATH = tgf.initTest(PROGNAME, TESTER, LOG_FILE_PATH, DESCRIPTION) # log in a file the start of the test
func(*args, **kwargs) # this is the user test
tgf.finishTestOk(PROGNAME)
CduAPIWrapper.disconnectAll()
CduAPIWrapper.asciiGlaciation()
except Exception as e:
print(traceback.format_exc())
logging.error(str(e))
print("An exception occurred: "+ str(e))
tgf.finishTestNotOK(str(e), PROGNAME)
try:
CduAPIWrapper.disconnectAll()
CduAPIWrapper.asciiGlaciation(CduAPIWrapper)
except Exception as e:
pass
exit(-1)
return wrapper
[docs]def measureTime(func):
"""Decorator function to measure and print out
time spend in execution in the given function
"""
@wraps(func)
def wrapper(*args, **kwargs):
t_start = perf_counter()
result = func(*args, **kwargs)
t_end = perf_counter()
print(f"Time spent in ({func.__name__}) is: {t_end-t_start}")
return result
return wrapper
[docs]class SSHHelper():
"""SSH Helper to run shh commands easily in remote devices
Parameters
----------
host : string
host address
user : string
username
key : string
user private key path (default is None)
pwd : string
password (default is None)
port : int
port number for the ssh connection (default is 22)
disAlgorithms : string
If given, must be a dictionary mapping algorithm type to an iterable of
algorithm identifiers, which will be disabled for the lifetime of the transport. \b
Example for TSW: dict(pubkeys=['rsa-sha2-256', 'rsa-sha2-512']) (default is None)
Raises
------
Exception
Error in connection
"""
def __init__(self, host, user, key=None, pwd="", port=22, disAlgorithms=None, timeout=3):
print("SSH: Connecting to %s as user: %s" % (host, user))
self.ssh = paramiko.SSHClient()
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
if key is not None:
privateKey = paramiko.RSAKey.from_private_key_file(key, pwd)
print("SSH: Private Key Loaded OK")
logging.getLogger("paramiko").setLevel(logging.ERROR)
self.ssh.connect(host,
port = port,
username = user,
password = pwd,
pkey = privateKey,
timeout = timeout,
auth_timeout = timeout,
disabled_algorithms = disAlgorithms)
print("SSH connection stablished")
[docs] def runCommand(self, command):
"""Runs a command in the previously connected device
Parameters
----------
command : string
command to be executed
Returns
------
outputs
Command outputs
"""
stdin, stdout, stderr = self.ssh.exec_command(command)
outputs = stdout.readlines()
errors = stderr.readlines()
for line in errors:
print(line)
for line in outputs:
print(line)
return outputs
[docs] def close(self):
"""Close SSH connection
"""
self.ssh.close()