3 # Copyright (c) 2017 Ericsson 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
11 OpenStack deployment checker
14 - Credentials file is given and contains the right information
15 - OpenStack endpoints are reachable
24 from urlparse import urlparse
26 from snaps.openstack.utils import glance_utils
27 from snaps.openstack.utils import keystone_utils
28 from snaps.openstack.utils import neutron_utils
29 from snaps.openstack.utils import nova_utils
30 from snaps.openstack.tests import openstack_tests
32 __author__ = "Jose Lausuch <jose.lausuch@ericsson.com>"
34 LOGGER = logging.getLogger(__name__)
37 def verify_connectivity(adress, port, timeout=10):
38 """ Returns true if an ip/port is reachable"""
39 connection = socket.socket()
41 while count < timeout:
43 connection.connect((adress, port))
44 LOGGER.debug('%s:%s is reachable!', adress, port)
50 LOGGER.error('%s:%s is not reachable.', adress, port)
54 class CheckDeployment(object):
55 """ Check deployment class."""
57 def __init__(self, rc_file='/home/opnfv/functest/conf/openstack.creds'):
58 self.rc_file = rc_file
59 self.services = ('compute', 'network', 'image')
63 """ Check if RC file exists and contains OS_AUTH_URL """
64 if not os.path.isfile(self.rc_file):
65 raise IOError('RC file {} does not exist!'.format(self.rc_file))
66 if 'OS_AUTH_URL' not in open(self.rc_file).read():
67 raise SyntaxError('OS_AUTH_URL not defined in {}.'.
70 def check_auth_endpoint(self):
71 """ Verifies connectivity to the OS_AUTH_URL given in the RC file """
72 rc_endpoint = self.os_creds.auth_url
73 if not (verify_connectivity(urlparse(rc_endpoint).hostname,
74 urlparse(rc_endpoint).port)):
75 raise Exception("OS_AUTH_URL {} is not reachable.".
77 LOGGER.info("Connectivity to OS_AUTH_URL %s ...OK", rc_endpoint)
79 def check_public_endpoint(self):
80 """ Gets the public endpoint and verifies connectivity to it """
81 public_endpoint = keystone_utils.get_endpoint(self.os_creds,
84 if not (verify_connectivity(urlparse(public_endpoint).hostname,
85 urlparse(public_endpoint).port)):
86 raise Exception("Public endpoint {} is not reachable.".
87 format(public_endpoint))
88 LOGGER.info("Connectivity to the public endpoint %s ...OK",
91 def check_service_endpoint(self, service):
92 """ Verifies connectivity to a given openstack service """
93 endpoint = keystone_utils.get_endpoint(self.os_creds,
96 if not (verify_connectivity(urlparse(endpoint).hostname,
97 urlparse(endpoint).port)):
98 raise Exception("{} endpoint {} is not reachable.".
99 format(service, endpoint))
100 LOGGER.info("Connectivity to endpoint '%s' %s ...OK",
103 def check_nova(self):
104 """ checks that a simple nova operation works """
106 client = nova_utils.nova_client(self.os_creds)
107 client.servers.list()
108 LOGGER.info("Nova service ...OK")
109 except Exception as error:
110 LOGGER.error("Nova service ...FAILED")
113 def check_neutron(self):
114 """ checks that a simple neutron operation works """
116 client = neutron_utils.neutron_client(self.os_creds)
117 client.list_networks()
118 LOGGER.info("Neutron service ...OK")
119 except Exception as error:
120 LOGGER.error("Neutron service ...FAILED")
123 def check_glance(self):
124 """ checks that a simple glance operation works """
126 client = glance_utils.glance_client(self.os_creds)
128 LOGGER.info("Glance service ...OK")
129 except Exception as error:
130 LOGGER.error("Glance service ...FAILED")
135 Calls all the class functions and returns 0 if all of them succeed.
136 This is the method called by prepare_env or CLI
140 self.os_creds = openstack_tests.get_credentials(
141 os_env_file=self.rc_file,
142 proxy_settings_str=None,
145 raise Exception("Problem while getting credentials object.")
146 if self.os_creds is None:
147 raise Exception("Credentials is None.")
148 self.check_auth_endpoint()
149 self.check_public_endpoint()
150 for service in self.services:
151 self.check_service_endpoint(service)
160 logging.config.fileConfig(pkg_resources.resource_filename(
161 'functest', 'ci/logging.ini'))
162 logging.captureWarnings(True)
163 deployment = CheckDeployment()
164 return deployment.check_all()