src/test/utils/SSHTestAgent.py

# Copyright 2017-present Open Networking Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


import os, sys, time
from paramiko import SSHClient, WarningPolicy, AutoAddPolicy
import logging
logging.getLogger('scapy.runtime').setLevel(logging.ERROR)
from CordTestUtils import log_test

class SSHTestAgent(object):
    key_file = os.getenv('SSH_KEY_FILE', None)
    host = os.getenv('CORD_TEST_HOST', '172.17.0.1')
    hosts_file = os.path.join(os.getenv('HOME'), '.ssh', 'known_hosts')
    user = 'ubuntu'
    password = None

    def __init__(self, host = host, user = user, password = password, port = 22):
        self.host = host
        self.user = user
        self.password = password
        self.port = port
        self.client = SSHClient()
        self.client.set_missing_host_key_policy(AutoAddPolicy())

    def run_cmd(self, cmd, timeout = 5):
        """Run the command on the test host"""
        host_remove = 'ssh-keygen -f "%s" -R [%s]:8101 2>/dev/null' %(self.hosts_file, self.host)
        try:
            os.system(host_remove)
        except: pass

        try:
            self.client.connect(self.host, username = self.user, password = self.password,
                                key_filename = self.key_file, timeout=timeout, port = self.port)
        except:
            log_test.error('Unable to connect to test host %s' %self.host)
            return False, None

        channel = self.client.get_transport().open_session()
        channel.exec_command(cmd)
        status_ready = False
        if channel.exit_status_ready():
            status = channel.recv_exit_status()
            status_ready = True
        else:
            status = 0
        output = None
        st = status == 0
        if st:
            output = ''
            while True:
                data = channel.recv(4096)
                if data:
                    output += data
                else:
                    break
        if status_ready is False:
            status = channel.recv_exit_status()
            st = status == 0
        time.sleep(0.1)
        channel.close()
        self.client.close()
        return st, output

if __name__ == '__main__':
    agent = SSHTestAgent(user = 'ubuntu', password = 'ubuntu')
    cmds = ('docker images', 'docker ps')
    for cmd in cmds:
        st, output = agent.run_cmd(cmd)
        print('Command \"%s\" returned with status: %s' %(cmd, st))
        if st:
            print('%s\n' %output)