diff options
| -rw-r--r-- | functest/ci/check_deployment.py | 163 | ||||
| -rw-r--r-- | functest/tests/unit/ci/test_check_deployment.py | 176 | ||||
| -rw-r--r-- | setup.cfg | 1 |
3 files changed, 340 insertions, 0 deletions
diff --git a/functest/ci/check_deployment.py b/functest/ci/check_deployment.py new file mode 100644 index 000000000..fe20dc8f1 --- /dev/null +++ b/functest/ci/check_deployment.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python + +# Copyright (c) 2017 Ericsson and others. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 + +""" +OpenStack deployment checker + +Verifies that: + - Credentials file is given and contains the right information + - OpenStack endpoints are reachable +""" + +import logging +import logging.config +import os +import pkg_resources +import socket +import time +from urlparse import urlparse + +from snaps.openstack.utils import glance_utils +from snaps.openstack.utils import keystone_utils +from snaps.openstack.utils import neutron_utils +from snaps.openstack.utils import nova_utils +from snaps.openstack.tests import openstack_tests + +__author__ = "Jose Lausuch <jose.lausuch@ericsson.com>" + +LOGGER = logging.getLogger(__name__) + + +def verify_connectivity(adress, port, timeout=10): + """ Returns true if an ip/port is reachable""" + connection = socket.socket() + count = 0 + while count < timeout: + try: + connection.connect((adress, port)) + LOGGER.debug('%s:%s is reachable!', adress, port) + return True + except socket.error: + count += 1 + time.sleep(1) + continue + LOGGER.error('%s:%s is not reachable.', adress, port) + return False + + +class CheckDeployment(object): + """ Check deployment class.""" + + def __init__(self, rc_file='/home/opnfv/functest/conf/openstack.creds'): + self.rc_file = rc_file + self.services = ('compute', 'network', 'image') + self.os_creds = None + + def check_rc(self): + """ Check if RC file exists and contains OS_AUTH_URL """ + if not os.path.isfile(self.rc_file): + raise IOError('RC file {} does not exist!'.format(self.rc_file)) + if 'OS_AUTH_URL' not in open(self.rc_file).read(): + raise SyntaxError('OS_AUTH_URL not defined in {}.'. + format(self.rc_file)) + + def check_auth_endpoint(self): + """ Verifies connectivity to the OS_AUTH_URL given in the RC file """ + rc_endpoint = self.os_creds.auth_url + if not (verify_connectivity(urlparse(rc_endpoint).hostname, + urlparse(rc_endpoint).port)): + raise Exception("OS_AUTH_URL {} is not reachable.". + format(rc_endpoint)) + LOGGER.info("Connectivity to OS_AUTH_URL %s ...OK", rc_endpoint) + + def check_public_endpoint(self): + """ Gets the public endpoint and verifies connectivity to it """ + public_endpoint = keystone_utils.get_endpoint(self.os_creds, + 'identity', + interface='public') + if not (verify_connectivity(urlparse(public_endpoint).hostname, + urlparse(public_endpoint).port)): + raise Exception("Public endpoint {} is not reachable.". + format(public_endpoint)) + LOGGER.info("Connectivity to the public endpoint %s ...OK", + public_endpoint) + + def check_service_endpoint(self, service): + """ Verifies connectivity to a given openstack service """ + endpoint = keystone_utils.get_endpoint(self.os_creds, + service, + interface='public') + if not (verify_connectivity(urlparse(endpoint).hostname, + urlparse(endpoint).port)): + raise Exception("{} endpoint {} is not reachable.". + format(service, endpoint)) + LOGGER.info("Connectivity to endpoint '%s' %s ...OK", + service, endpoint) + + def check_nova(self): + """ checks that a simple nova operation works """ + try: + client = nova_utils.nova_client(self.os_creds) + client.servers.list() + LOGGER.info("Nova service ...OK") + except Exception as error: + LOGGER.error("Nova service ...FAILED") + raise error + + def check_neutron(self): + """ checks that a simple neutron operation works """ + try: + client = neutron_utils.neutron_client(self.os_creds) + client.list_networks() + LOGGER.info("Neutron service ...OK") + except Exception as error: + LOGGER.error("Neutron service ...FAILED") + raise error + + def check_glance(self): + """ checks that a simple glance operation works """ + try: + client = glance_utils.glance_client(self.os_creds) + client.images.list() + LOGGER.info("Glance service ...OK") + except Exception as error: + LOGGER.error("Glance service ...FAILED") + raise error + + def check_all(self): + """ + Calls all the class functions and returns 0 if all of them succeed. + This is the method called by prepare_env or CLI + """ + self.check_rc() + try: + self.os_creds = openstack_tests.get_credentials( + os_env_file=self.rc_file, + proxy_settings_str=None, + ssh_proxy_cmd=None) + except: + raise Exception("Problem while getting credentials object.") + if self.os_creds is None: + raise Exception("Credentials is None.") + self.check_auth_endpoint() + self.check_public_endpoint() + for service in self.services: + self.check_service_endpoint(service) + self.check_nova() + self.check_neutron() + self.check_glance() + return 0 + + +def main(): + """Entry point""" + logging.config.fileConfig(pkg_resources.resource_filename( + 'functest', 'ci/logging.ini')) + deployment = CheckDeployment() + return deployment.check_all() diff --git a/functest/tests/unit/ci/test_check_deployment.py b/functest/tests/unit/ci/test_check_deployment.py new file mode 100644 index 000000000..1f44d0787 --- /dev/null +++ b/functest/tests/unit/ci/test_check_deployment.py @@ -0,0 +1,176 @@ +#!/usr/bin/env python + +# Copyright (c) 2017 Ericsson and others. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 + +import logging +import mock +import unittest + +from functest.ci import check_deployment + +__author__ = "Jose Lausuch <jose.lausuch@ericsson.com>" + + +class CheckDeploymentTesting(unittest.TestCase): + """The super class which testing classes could inherit.""" + # pylint: disable=missing-docstring + + logging.disable(logging.CRITICAL) + + def setUp(self): + self.client_test = mock.Mock() + self.deployment = check_deployment.CheckDeployment() + self.service_test = 'compute' + self.rc_file = self.deployment.rc_file + self.endpoint_test = 'http://192.168.0.6:5000/v3' + creds_attr = {'auth_url': self.endpoint_test, + 'proxy_settings': ''} + proxy_attr = {'host': '192.168.0.1', 'port': '5000'} + proxy_settings = mock.Mock() + proxy_settings.configure_mock(**proxy_attr) + self.os_creds = mock.Mock() + self.os_creds.configure_mock(**creds_attr) + self.os_creds.proxy_settings = proxy_settings + self.deployment.os_creds = self.os_creds + + def test_check_rc(self): + with mock.patch('functest.ci.check_deployment.os.path.isfile', + returns=True) as m, \ + mock.patch('__builtin__.open', + mock.mock_open(read_data='OS_AUTH_URL')): + self.deployment.check_rc() + self.assertTrue(m.called) + + def test_check_rc_missing_file(self): + with mock.patch('functest.ci.check_deployment.os.path.isfile', + return_value=False), \ + self.assertRaises(Exception) as context: + msg = 'RC file {} does not exist!'.format(self.rc_file) + self.deployment.check_rc(self.rc_file) + self.assertTrue(msg in context) + + def test_check_rc_missing_os_auth(self): + with mock.patch('__builtin__.open', + mock.mock_open(read_data='test')), \ + self.assertRaises(Exception) as context: + msg = 'OS_AUTH_URL not defined in {}.'.format(self.rc_file) + self.assertTrue(msg in context) + + def test_check_auth_endpoint(self): + with mock.patch('functest.ci.check_deployment.verify_connectivity', + return_value=True) as m: + self.deployment.check_auth_endpoint() + self.assertTrue(m.called) + + def test_check_auth_endpoint_not_reachable(self): + with mock.patch('functest.ci.check_deployment.verify_connectivity', + return_value=False) as m, \ + self.assertRaises(Exception) as context: + endpoint = self.os_creds.auth_url + self.deployment.check_auth_endpoint() + msg = "OS_AUTH_URL {} is not reachable.".format(endpoint) + self.assertTrue(m.called) + self.assertTrue(msg in context) + + def test_check_public_endpoint(self): + with mock.patch('functest.ci.check_deployment.verify_connectivity', + return_value=True) as m, \ + mock.patch('functest.ci.check_deployment.keystone_utils.' + 'get_endpoint') as n: + self.deployment.check_public_endpoint() + self.assertTrue(m.called) + self.assertTrue(n.called) + + def test_check_public_endpoint_not_reachable(self): + with mock.patch('functest.ci.check_deployment.verify_connectivity', + return_value=False) as m, \ + mock.patch('functest.ci.check_deployment.keystone_utils.' + 'get_endpoint', + return_value=self.endpoint_test) as n, \ + self.assertRaises(Exception) as context: + self.deployment.check_public_endpoint() + msg = ("Public endpoint {} is not reachable." + .format(self.mock_endpoint)) + self.assertTrue(m.called) + self.assertTrue(n.called) + self.assertTrue(msg in context) + + def test_check_service_endpoint(self): + with mock.patch('functest.ci.check_deployment.verify_connectivity', + return_value=True) as m, \ + mock.patch('functest.ci.check_deployment.keystone_utils.' + 'get_endpoint') as n: + self.deployment.check_service_endpoint(self.service_test) + self.assertTrue(m.called) + self.assertTrue(n.called) + + def test_check_service_endpoint_not_reachable(self): + with mock.patch('functest.ci.check_deployment.verify_connectivity', + return_value=False) as m, \ + mock.patch('functest.ci.check_deployment.keystone_utils.' + 'get_endpoint', + return_value=self.endpoint_test) as n, \ + self.assertRaises(Exception) as context: + self.deployment.check_service_endpoint(self.service_test) + msg = "{} endpoint {} is not reachable.".format(self.service_test, + self.endpoint_test) + self.assertTrue(m.called) + self.assertTrue(n.called) + self.assertTrue(msg in context) + + def test_check_nova(self): + with mock.patch('functest.ci.check_deployment.nova_utils.nova_client', + return_value=self.client_test) as m: + self.deployment.check_nova() + self.assertTrue(m.called) + + def test_check_nova_fail(self): + with mock.patch('functest.ci.check_deployment.nova_utils.nova_client', + return_value=self.client_test) as m, \ + mock.patch.object(self.client_test, 'servers.list', + side_effect=Exception): + self.deployment.check_nova() + self.assertTrue(m.called) + self.assertRaises(Exception) + + def test_check_neutron(self): + with mock.patch('functest.ci.check_deployment.neutron_utils.' + 'neutron_client', return_value=self.client_test) as m: + self.deployment.check_neutron() + self.assertTrue(m.called) + + def test_check_neutron_fail(self): + with mock.patch('functest.ci.check_deployment.neutron_utils.' + 'neutron_client', + return_value=self.client_test) as m, \ + mock.patch.object(self.client_test, 'list_networks', + side_effect=Exception), \ + self.assertRaises(Exception): + self.deployment.check_neutron() + self.assertRaises(Exception) + self.assertTrue(m.called) + + def test_check_glance(self): + with mock.patch('functest.ci.check_deployment.glance_utils.' + 'glance_client', return_value=self.client_test) as m: + self.deployment.check_glance() + self.assertTrue(m.called) + + def test_check_glance_fail(self): + with mock.patch('functest.ci.check_deployment.glance_utils.' + 'glance_client', return_value=self.client_test) as m, \ + mock.patch.object(self.client_test, 'images.list', + side_effect=Exception): + self.deployment.check_glance() + self.assertRaises(Exception) + self.assertTrue(m.called) + + +if __name__ == "__main__": + logging.disable(logging.CRITICAL) + unittest.main(verbosity=2) @@ -23,3 +23,4 @@ console_scripts = openstack_clean = functest.utils.openstack_clean:main prepare_env = functest.ci.prepare_env:main run_tests = functest.ci.run_tests:main + check_deployment = functest.ci.check_deployment:main |