3 # Copyright (c) 2017 Okinawa Open Laboratory and others.
5 # All rights reserved. This program and the accompanying materials
6 # are made available under the terms of the Apache License, Version 2.0
7 # which accompanies this distribution, and is available at
8 # http://www.apache.org/licenses/LICENSE-2.0
10 """ssh client module for vrouter testing"""
17 from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf
21 DEFAULT_CONNECT_TIMEOUT = 10
22 DEFAULT_CONNECT_RETRY_COUNT = 10
23 DEFAULT_SEND_TIMEOUT = 10
26 class SshClient(object):
27 """ssh client class for vrouter testing"""
29 logger = logging.getLogger(__name__)
31 def __init__(self, ip_address, user, password=None, key_filename=None):
32 self.ip_address = ip_address
34 self.password = password
35 self.key_filename = key_filename
36 self.connected = False
39 self.logger.setLevel(logging.INFO)
41 self.ssh = paramiko.SSHClient()
42 self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
45 with open(self.util.test_env_config_yaml) as file_fd:
46 test_env_config_yaml = yaml.safe_load(file_fd)
49 self.ssh_revieve_buff = test_env_config_yaml.get("general").get(
52 def connect(self, time_out=DEFAULT_CONNECT_TIMEOUT,
53 retrycount=DEFAULT_CONNECT_RETRY_COUNT):
56 self.logger.info("SSH connect to %s.", self.ip_address)
57 self.ssh.connect(self.ip_address,
59 password=self.password,
60 key_filename=self.key_filename,
65 self.logger.info("SSH connection established to %s.",
68 self.shell = self.ssh.invoke_shell()
70 while not self.shell.recv_ready():
71 time.sleep(RECEIVE_ROOP_WAIT)
73 self.shell.recv(self.ssh_revieve_buff)
75 except: # pylint: disable=broad-except
76 self.logger.info("SSH timeout for %s...", self.ip_address)
81 self.logger.error("Cannot establish connection to IP '%s'. " +
84 self.connected = False
90 def send(self, cmd, prompt, timeout=DEFAULT_SEND_TIMEOUT):
91 if self.connected is True:
92 self.shell.settimeout(timeout)
93 self.logger.debug("Commandset : '%s'", cmd)
96 self.shell.send(cmd + '\n')
97 except: # pylint: disable=broad-except
98 self.logger.error("ssh send timeout : Command : '%s'", cmd)
102 while not res_buff.endswith(prompt):
103 time.sleep(RECEIVE_ROOP_WAIT)
105 res = self.shell.recv(self.ssh_revieve_buff)
106 except: # pylint: disable=broad-except
107 self.logger.error("ssh receive timeout : Command : '%s'",
113 self.logger.debug("Response : '%s'", res_buff)
116 self.logger.error("Cannot connected to IP '%s'.", self.ip_address)
120 if self.connected is True:
123 def error_check(response, err_strs=["error",