functest: Add test code of DMA localagent

Change-Id: If1195b7b3d9000e3ec75bc3c4c108b0e9a1bc9e3
Signed-off-by: Toshiaki Takahashi <takahashi.tsc@ncos.nec.co.jp>

5 files changed, 500 insertions, 2 deletions

diff --git a/baro_tests/collectd.py b/baro_tests/collectd.py
index 922403f1..51350495 100644
--- a/baro_tests/collectd.py
+++ b/baro_tests/collectd.py
@@ -23,6 +23,7 @@ import time
 import logging
 import config_server
 import tests
+import local_agent
 from distutils import version
 from opnfv.deployment import factory
 
@@ -866,14 +867,19 @@ def main(bt_logger=None):
     print_overall_summary(
         compute_ids, plugin_labels, aodh_plugin_labels, results, out_plugins)
 
+    res_overall = 0
     for res in results:
         if not res[3]:
             logger.error('Some tests have failed or have not been executed')
             logger.error('Overall Result is Fail')
-            return 1
+            res_overall = 1
         else:
             pass
-    return 0
+
+    _print_label('Testing LocalAgent on compute nodes')
+    res_agent = local_agent.local_agent_main(logger, conf, computes)
+
+    return 0 if res_overall == 0 and res_agent == 0 else 1
 
 
 if __name__ == '__main__':
diff --git a/baro_tests/config_server.py b/baro_tests/config_server.py
index fd31c523..71d4f152 100644
--- a/baro_tests/config_server.py
+++ b/baro_tests/config_server.py
@@ -18,6 +18,7 @@ import time
 import os.path
 import os
 import re
+import yaml
 
 from opnfv.deployment import factory
 import paramiko
@@ -300,6 +301,97 @@ class ConfigServer(object):
                     return False
         return aodh_present
 
+    def is_redis_running(self, compute):
+        """Check whether redis service is running on compute"""
+        compute_name = compute.get_name()
+        nodes = get_apex_nodes()
+        for node in nodes:
+            if compute_name == node.get_dict()['name']:
+                stdout = node.run_cmd('sudo systemctl status docker'
+                                      '&& sudo docker ps'
+                                      '| grep barometer-redis')
+                if stdout and 'barometer-redis' in stdout:
+                    self.__logger.info(
+                        'Redis is running in node {}'.format(
+                         compute_name))
+                    return True
+        self.__logger.info(
+            'Redis is *not* running in node {}'.format(
+             compute_name))
+        return False
+
+    def is_localagent_server_running(self, compute):
+        """Check whether LocalAgent server is running on compute"""
+        compute_name = compute.get_name()
+        nodes = get_apex_nodes()
+        for node in nodes:
+            if compute_name == node.get_dict()['name']:
+                stdout = node.run_cmd('sudo systemctl status docker'
+                                      '&& sudo docker ps'
+                                      '| grep opnfv/barometer-localagent')
+                if stdout and '/server' in stdout:
+                    self.__logger.info(
+                        'LocalAgent Server is running in node {}'.format(
+                         compute_name))
+                    return True
+        self.__logger.info(
+            'LocalAgent Server is *not* running in node {}'.format(
+             compute_name))
+        return False
+
+    def is_localagent_infofetch_running(self, compute):
+        """Check whether LocalAgent infofetch is running on compute"""
+        compute_name = compute.get_name()
+        nodes = get_apex_nodes()
+        for node in nodes:
+            if compute_name == node.get_dict()['name']:
+                stdout = node.run_cmd('sudo systemctl status docker'
+                                      '&& sudo docker ps'
+                                      '| grep opnfv/barometer-localagent')
+                if stdout and '/infofetch' in stdout:
+                    self.__logger.info(
+                        'LocalAgent InfoFetch is running in node {}'.format(
+                         compute_name))
+                    return True
+        self.__logger.info(
+            'LocalAgent InfoFetch is *not* running in node {}'.format(
+             compute_name))
+        return False
+
+    def get_localagent_config(self, compute):
+        """Get config values of LocalAgent"""
+        compute_name = compute.get_name()
+        nodes = get_apex_nodes()
+        for node in nodes:
+            if compute_name == node.get_dict()['name']:
+                # We use following after functest accept python-toml
+                #     stdout = node.run_cmd(
+                #         'cat /etc/barometer-localagent/config.toml')
+                #     try:
+                #         agent_conf = toml.loads(stdout)
+                #     except (TypeError, TomlDecodeError) as e:
+                #         self.__logger.error(
+                #             'LocalAgent config error: {}'.format(e))
+                #         agent_conf = None
+                #     finally:
+                #         return agent_conf
+                readcmd = (
+                    'egrep "listen_port|amqp_"'
+                    ' /etc/barometer-localagent/config.toml'
+                    '| sed -e "s/#.*$//" | sed -e "s/=/:/"'
+                    )
+                stdout = node.run_cmd(readcmd)
+                agent_conf = {"server": yaml.load(stdout)}
+
+                pingcmd = (
+                    'ping -n -c1 ' + agent_conf["server"]["amqp_host"] +
+                    '| sed -ne "s/^.*bytes from //p" | sed -e "s/:.*//"'
+                    )
+                agent_conf["server"]["amqp_host"] = node.run_cmd(pingcmd)
+
+                return agent_conf
+        return None
+
     def is_mcelog_installed(self, compute, package):
         """Check whether package exists on compute node.
 
@@ -660,3 +752,129 @@ class ConfigServer(object):
                         return True
         else:
             return False
+
+    def check_localagent_dummy_included(self, compute, name):
+        """Check if dummy collectd config by LocalAgent
+           is included in collectd.conf file.
+
+        Keyword arguments:
+        compute -- compute node instance
+        name -- config file name
+        """
+        compute_name = compute.get_name()
+        nodes = get_apex_nodes()
+        for node in nodes:
+            if compute_name == node.get_dict()['name']:
+                dummy_conf = node.run_cmd('ls /etc/collectd/collectd.conf.d')
+                if name + '.conf' not in dummy_conf:
+                    self.__logger.error('check conf FAIL')
+                    return False
+                else:
+                    self.__logger.info('check conf PASS')
+                    fullpath = '/etc/collectd/collectd.conf.d/{}'.format(
+                               name + '.conf')
+                    self.__logger.info('Delete file {}'.format(fullpath))
+                    node.run_cmd('sudo rm -f ' + fullpath)
+                    return True
+        self.__logger.error('Some panic, compute not found')
+        return False
+
+    def create_testvm(self, compute_node, test_name):
+        nodes = get_apex_nodes()
+        compute_name = compute_node.get_name()
+
+        controller_node = None
+        for node in nodes:
+            if node.is_controller():
+                controller_node = node
+                break
+
+        self.__logger.debug('Creating Test VM on {}' .format(compute_name))
+        self.__logger.debug('Create command is executed in {}' .format(
+            (controller_node.get_dict()['name'])))
+
+        image_filename = 'cirros-0.4.0-x86_64-disk.img'
+        controller_node.run_cmd(
+            'curl -sO '
+            'http://download.cirros-cloud.net/0.4.0/'
+            + image_filename)
+
+        node.put_file(constants.ENV_FILE, 'overcloudrc.v3')
+        image = controller_node.run_cmd(
+            'source overcloudrc.v3;'
+            'openstack image create -f value -c id'
+            ' --disk-format qcow2 --file {0} {1}'
+            .format(image_filename, test_name))
+        flavor = controller_node.run_cmd(
+            'source overcloudrc.v3;'
+            'openstack flavor create -f value -c id {}'
+            .format(test_name))
+        host = controller_node.run_cmd(
+            'source overcloudrc.v3;'
+            'openstack hypervisor list -f value -c "Hypervisor Hostname"'
+            ' | grep "^{}\\."'
+            .format(compute_name))
+        server = controller_node.run_cmd(
+            'source overcloudrc.v3;'
+            'openstack server create -f value -c id'
+            ' --image {0} --flavor {1} --availability-zone {2} {3}'
+            .format(image, flavor, 'nova:' + host, test_name))
+
+        resources = {"image": image, "flavor": flavor, "server": server}
+
+        if server:
+            self.__logger.debug('VM created')
+        self.__logger.debug('VM info: {}'.format(resources))
+
+        return resources
+
+    def delete_testvm(self, resources):
+        nodes = get_apex_nodes()
+
+        controller_node = None
+        for node in nodes:
+            if node.is_controller():
+                controller_node = node
+                break
+
+        self.__logger.debug('Deleteing Test VM')
+        self.__logger.debug('VM to be deleted info: {}'.format(resources))
+        self.__logger.debug('Delete command is executed in {}' .format(
+            (controller_node.get_dict()['name'])))
+
+        server = resources.get('server', None)
+        flavor = resources.get('flavor', None)
+        image = resources.get('image', None)
+        if server:
+            controller_node.run_cmd(
+                'source overcloudrc.v3;'
+                'openstack server delete {}'.format(server))
+        if flavor:
+            controller_node.run_cmd(
+                'source overcloudrc.v3;'
+                'openstack flavor delete {}'.format(flavor))
+        if image:
+            controller_node.run_cmd(
+                'source overcloudrc.v3;'
+                'openstack image delete {}'.format(image))
+
+        self.__logger.debug('VM and other OpenStack resources deleted')
+
+    def test_localagent_infofetch_get_data(self, compute, test_name):
+        compute_name = compute.get_name()
+        nodes = get_apex_nodes()
+        for node in nodes:
+            if compute_name == node.get_dict()['name']:
+                stdout = node.run_cmd(
+                    'redis-cli keys "barometer-localagent/vm/*/vminfo"'
+                    ' | while read k; do redis-cli get $k; done'
+                    ' | grep {}'.format(test_name))
+                self.__logger.debug('InfoFetch data: {}'.format(stdout))
+                if stdout and test_name in stdout:
+                    self.__logger.info('PASS')
+                    return True
+                else:
+                    self.__logger.info('No test vm info')
+
+        self.__logger.info('FAIL')
+        return False
diff --git a/baro_tests/local_agent.py b/baro_tests/local_agent.py
new file mode 100644
index 00000000..8201dee0
--- /dev/null
+++ b/baro_tests/local_agent.py
@@ -0,0 +1,261 @@
+# -*- coding: utf-8 -*-
+
+# 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.
+# Patch on October 10 2017
+
+"""Executing test of LocalAgent"""
+
+import os
+import pika
+import requests
+import time
+
+import tests
+
+logger = None
+
+
+class LocalAgentClient(object):
+    """Client to request LocalAgent"""
+    def __init__(self, host, port, user, passwd):
+        """
+        Keyword arguments:
+        host -- Host URL
+        port -- Host Port
+        user -- Username
+        passwd -- Password
+        """
+        self._host = host
+        self._port = port
+        self._user = user
+        self._passwd = passwd
+
+    def set(self, file):
+        logger.error('Do nothing to LocalAgent')
+
+    def __str__(self):
+        return ('host: {0}, port: {1}, user: {2}, pass: {3}'
+                .format(self._host, self._port,
+                        self._user, (self._passwd and '<Filterd>')))
+
+
+class RestLocalAgentClient(LocalAgentClient):
+    """Client to request LocalAgent using REST"""
+    def __init__(self, host, port, user, passwd):
+        super(self.__class__, self).__init__(host, port, user, passwd)
+
+    def set(self, file):
+        logger.debug('Send to localagent using REST -- {}'.format(str(self)))
+
+        if not os.path.isfile(file):
+            print '{} is not found'.format(file)
+            return False
+        filename = os.path.basename(file)
+
+        url = 'http://{0}:{1}/collectd/conf'.format(self._host, self._port)
+        config = {'file': (filename, open(file, 'r'))}
+        requests.post(url, files=config)
+
+        return True
+
+
+class PubLocalAgentClient(LocalAgentClient):
+    """Client to request LocalAgent using AMQP Publish"""
+    def __init__(self, host, port, user, passwd):
+        super(self.__class__, self).__init__(host, port, user, passwd)
+
+    def set(self, file):
+        logger.debug('Send to localagent using AMQP Publish -- {}'
+                     .format(str(self)))
+
+        if not os.path.isfile(file):
+            print '{} is not found'.format(file)
+            return False
+        filename = os.path.basename(file)
+        filebody = open(file, 'r').read()
+        message = filename + '/' + filebody
+
+        credentials = pika.PlainCredentials(self._user, self._passwd)
+        connection = pika.BlockingConnection(pika.ConnectionParameters(
+                host=self._host, port=int(self._port),
+                credentials=credentials))
+        channel = connection.channel()
+        channel.exchange_declare(exchange='collectd-conf',
+                                 exchange_type='fanout')
+        channel.basic_publish(exchange='collectd-conf',
+                              routing_key='',
+                              body=message)
+
+        connection.close()
+        return True
+
+
+def _process_localagent_result(compute_node, testfunc,
+                               result, results_list, node):
+    """Print LocalAgent test result and append it to results list.
+
+    Keyword arguments:
+    testfunc -- localagent function name
+    result -- boolean test result
+    results_list -- results list
+    """
+    if result:
+        logger.info(
+            'Test case for {0} with LocalAgent PASSED on {1}.'.format(
+                node, testfunc))
+    else:
+        logger.error(
+            'Test case for {0} with LocalAgent FAILED on {1}.'.format(
+                node, testfunc))
+    results_list.append((compute_node, "LocalAgent", testfunc, result))
+
+
+def _print_result_of_localagent(compute_ids, results):
+    """Print results of LocalAgent.
+
+    Keyword arguments:
+    compute_ids -- list of compute node IDs
+    results -- results list
+    """
+    compute_node_names = ['Node-{}'.format(i) for i in range(
+        len((compute_ids)))]
+    all_computes_in_line = ''
+    for compute in compute_node_names:
+        all_computes_in_line += '| ' + compute + (' ' * (7 - len(compute)))
+    line_of_nodes = '| Test           ' + all_computes_in_line + '|'
+    logger.info('=' * 70)
+    logger.info('+' + ('-' * ((9 * len(compute_node_names))+16)) + '+')
+    logger.info(
+        '|' + ' ' * ((9*len(compute_node_names))/2)
+        + ' LOCALAGENT TEST'
+        + ' ' * (
+            9*len(compute_node_names) - (9*len(compute_node_names))/2)
+        + '|')
+    logger.info(
+        '+' + ('-' * 16) + '+' + (('-' * 8) + '+') * len(compute_node_names))
+    logger.info(line_of_nodes)
+    logger.info(
+        '+' + ('-' * 16) + '+' + (('-' * 8) + '+') * len(compute_node_names))
+
+    testname = "LocalAgent"
+    print_line = ''
+    for id in compute_ids:
+        all_result = \
+            'FAIL' if [
+                testfunc for comp_id, testname, testfunc, res in results
+                if comp_id == id and not res] else 'PASS'
+        print_line += '| ' + all_result + '   '
+    logger.info(
+        '| {}'.format(testname) + (' ' * (15 - len(testname)))
+        + print_line + '|')
+
+    for testfunc in ['Server', 'InfoFetch']:
+        print_line = ''
+        for id in compute_ids:
+            if (id, testname, testfunc, True) in results:
+                print_line += ' PASS   |'
+            elif (id, testname, testfunc, False) in results:
+                print_line += ' FAIL   |'
+            else:
+                print_line += ' SKIP   |'
+        logger.info(
+            '|  {}'.format(testfunc) + (' ' * (14-len(testfunc)))
+            + '|' + print_line)
+
+    logger.info(
+        '+' + ('-' * 16) + '+'
+        + (('-' * 8) + '+') * len(compute_node_names))
+    logger.info('=' * 70)
+
+
+def local_agent_main(bt_logger, conf, computes):
+    """Check LocalAgent of each compute node.
+
+    Keyword arguments:
+    bt_logger -- logger instance
+    computes -- compute node list
+    """
+    global logger
+    logger = bt_logger
+
+    compute_ids = []
+    agent_results = []
+    for compute_node in computes:
+        node_id = compute_node.get_id()
+        compute_ids.append(node_id)
+
+        agent_server_running = conf.is_localagent_server_running(compute_node)
+        agent_infofetch_running = (
+            conf.is_localagent_infofetch_running(compute_node) and
+            conf.is_redis_running(compute_node))
+
+        if agent_server_running:
+            test_name = 'barotest'
+            tmpfile = '/tmp/' + test_name + '.conf'
+
+            agent_config = conf.get_localagent_config(compute_node)
+            listen_ip = compute_node.get_ip()
+            listen_port = agent_config.get('server').get('listen_port')
+            amqp_host = agent_config.get('server').get('amqp_host')
+            amqp_port = agent_config.get('server').get('amqp_port')
+            amqp_user = agent_config.get('server').get('amqp_user')
+            amqp_passwd = agent_config.get('server').get('amqp_password')
+            rest_client = RestLocalAgentClient(
+                              listen_ip, listen_port, '', '')
+            pub_client = PubLocalAgentClient(
+                             amqp_host, amqp_port, amqp_user,
+                             amqp_passwd)
+
+            all_res = True
+            for client in [rest_client, pub_client]:
+                tests.test_localagent_server_set_collectd(
+                    compute_node, tmpfile, logger, client)
+                sleep_time = 1
+                logger.info(
+                    'Sleeping for {} seconds'.format(sleep_time)
+                    + ' before localagent server test...')
+                time.sleep(sleep_time)
+                res = conf.check_localagent_dummy_included(
+                          compute_node, test_name)
+                all_res = all_res and res
+
+            _process_localagent_result(
+                compute_node.get_id(), 'Server',
+                all_res, agent_results, compute_node.get_name())
+
+        if agent_infofetch_running:
+            test_name = 'barotest'
+            resources = conf.create_testvm(compute_node, test_name)
+            sleep_time = 5
+            logger.info(
+                'Sleeping for {} seconds'.format(sleep_time)
+                + ' before localagent infofetch test...')
+            time.sleep(sleep_time)
+            res = conf.test_localagent_infofetch_get_data(
+                      compute_node, test_name)
+            conf.delete_testvm(resources)
+
+            _process_localagent_result(
+                compute_node.get_id(), 'InfoFetch',
+                res, agent_results, compute_node.get_name())
+
+    _print_result_of_localagent(compute_ids, agent_results)
+
+    for res in agent_results:
+        if not res[3]:
+            logger.error('Some tests have failed or have not been executed')
+            logger.error('LocalAgent test is Fail')
+            return 1
+        else:
+            pass
+    return 0
diff --git a/baro_tests/tests.py b/baro_tests/tests.py
index 02eca90a..bd49583d 100644
--- a/baro_tests/tests.py
+++ b/baro_tests/tests.py
@@ -270,3 +270,15 @@ def test_csv_handles_plugin_data(
     logger.info('OK')
 
     return True
+
+
+def test_localagent_server_set_collectd(compute, file, logger, client):
+    with open(file, mode='w') as f:
+        f.write('# dummy conf\n')
+    res = client.set(file)
+    if res:
+        logger.info('set collectd PASS')
+    else:
+        logger.error('set collectd FAIL')
+
+    return res
diff --git a/requirements.txt b/requirements.txt
index 3fd72068..55e2f3c7 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -4,6 +4,7 @@
 paramiko>=2.0.0 # LGPLv2.1+
 requests>=2.14.2 # Apache-2.0
 python-keystoneclient>=3.8.0 # Apache-2.0
+toml # MIT
 opnfv # Apache-2.0
 functest # Apache-2.0
 xtesting # Apache-2.0