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"""
18 from functest.opnfv_tests.vnf.router.utilvnf import Utilvnf
22 DEFAULT_CONNECT_TIMEOUT = 10
23 DEFAULT_CONNECT_RETRY_COUNT = 10
24 DEFAULT_SEND_TIMEOUT = 10
27 class SshClient(): # pylint: disable=too-many-instance-attributes
28 """ssh client class for vrouter testing"""
30 logger = logging.getLogger(__name__)
32 def __init__(self, ip_address, user, password=None, key_filename=None):
33 self.ip_address = ip_address
35 self.password = password
36 self.key_filename = key_filename
37 self.connected = False
40 self.logger.setLevel(logging.INFO)
42 self.ssh = paramiko.SSHClient()
43 self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
46 with open(self.util.test_env_config_yaml, encoding='utf-8') as file_fd:
47 test_env_config_yaml = yaml.safe_load(file_fd)
50 self.ssh_revieve_buff = test_env_config_yaml.get("general").get(
53 def connect(self, time_out=DEFAULT_CONNECT_TIMEOUT,
54 retrycount=DEFAULT_CONNECT_RETRY_COUNT):
55 # pylint: disable=missing-docstring
58 self.logger.info("SSH connect to %s.", self.ip_address)
59 self.ssh.connect(self.ip_address,
61 password=self.password,
62 key_filename=self.key_filename,
67 self.logger.info("SSH connection established to %s.",
70 self.shell = self.ssh.invoke_shell()
72 while not self.shell.recv_ready():
73 time.sleep(RECEIVE_ROOP_WAIT)
75 self.shell.recv(self.ssh_revieve_buff)
77 except Exception: # pylint: disable=broad-except
78 self.logger.info("SSH timeout for %s...", self.ip_address)
84 "Cannot establish connection to IP '%s'", self.ip_address)
85 self.connected = False
91 def send(self, cmd, prompt, timeout=DEFAULT_SEND_TIMEOUT):
92 # pylint: disable=missing-docstring
93 if self.connected is True:
94 self.shell.settimeout(timeout)
95 self.logger.debug("Commandset : '%s'", cmd)
98 self.shell.send(cmd + '\n')
99 except Exception: # pylint: disable=broad-except
100 self.logger.error("ssh send timeout : Command : '%s'", cmd)
104 while not res_buff.endswith(prompt):
105 time.sleep(RECEIVE_ROOP_WAIT)
107 res = self.shell.recv(self.ssh_revieve_buff)
108 except Exception: # pylint: disable=broad-except
109 self.logger.error("ssh receive timeout : Command : '%s'",
113 res_buff += res.decode("utf-8")
115 self.logger.debug("Response : '%s'", res_buff)
117 self.logger.error("Cannot connected to IP '%s'.", self.ip_address)
121 # pylint: disable=missing-docstring
122 if self.connected is True:
126 def error_check(response, err_strs=None):
127 # pylint: disable=missing-docstring
129 err_strs = ["error", "warn", "unknown command", "already exist"]