# 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.
+from neutronclient.common.utils import str2bool
+
+from snaps import file_utils
+from snaps.openstack.utils import glance_utils, keystone_utils
+
__author__ = 'spisarski'
Neutron clients
:param compute_api_version: The OpenStack's API version to use for Nova
clients
+ :param heat_api_version: The OpenStack's API version to use for Heat
+ clients
:param user_domain_id: Used for v3 APIs
:param project_domain_id: Used for v3 APIs
:param interface: Used to specify the endpoint type for keystone as
self.password = kwargs.get('password')
self.auth_url = kwargs.get('auth_url')
self.project_name = kwargs.get('project_name')
- self.identity_api_version = kwargs.get('identity_api_version', 2)
- self.image_api_version = kwargs.get('image_api_version', 2)
- self.network_api_version = kwargs.get('network_api_version', 2)
- self.compute_api_version = kwargs.get('compute_api_version', 2)
- self.user_domain_id = kwargs.get('user_domain_id', 'default')
- self.project_domain_id = kwargs.get('project_domain_id', 'default')
- self.interface = kwargs.get('interface', 'admin')
+
+ if kwargs.get('identity_api_version') is None:
+ self.identity_api_version = keystone_utils.V2_VERSION_NUM
+ else:
+ self.identity_api_version = float(kwargs['identity_api_version'])
+
+ if kwargs.get('image_api_version') is None:
+ self.image_api_version = glance_utils.VERSION_2
+ else:
+ self.image_api_version = float(kwargs['image_api_version'])
+
+ if kwargs.get('network_api_version') is None:
+ self.network_api_version = 2
+ else:
+ self.network_api_version = float(kwargs['network_api_version'])
+
+ if kwargs.get('compute_api_version') is None:
+ self.compute_api_version = 2
+ else:
+ self.compute_api_version = float(kwargs['compute_api_version'])
+
+ if kwargs.get('heat_api_version') is None:
+ self.heat_api_version = 1
+ else:
+ self.heat_api_version = float(kwargs['heat_api_version'])
+
+ if kwargs.get('user_domain_id') is None:
+ self.user_domain_id = 'default'
+ else:
+ self.user_domain_id = kwargs['user_domain_id']
+
+ if kwargs.get('project_domain_id') is None:
+ self.project_domain_id = 'default'
+ else:
+ self.project_domain_id = kwargs['project_domain_id']
+
+ if kwargs.get('interface') is None:
+ self.interface = 'admin'
+ else:
+ self.interface = kwargs['interface']
+
self.cacert = kwargs.get('cacert', True)
+ if isinstance(kwargs.get('cacert'), str):
+ if file_utils.file_exists(kwargs['cacert']):
+ self.cacert = kwargs['cacert']
+ else:
+ self.cacert = str2bool(self.cacert)
if isinstance(kwargs.get('proxy_settings'), ProxySettings):
self.proxy_settings = kwargs.get('proxy_settings')
Constructor
:param host: the HTTP proxy host
:param port: the HTTP proxy port
+ :param https_host: the HTTPS proxy host (defaults to host)
+ :param https_port: the HTTPS proxy port (defaults to port)
+ :param port: the HTTP proxy port
:param ssh_proxy_cmd: the SSH proxy command string (optional)
"""
self.host = kwargs.get('host')
self.port = kwargs.get('port')
+
+ self.https_host = kwargs.get('https_host', self.host)
+ self.https_port = kwargs.get('https_port', self.port)
+
self.ssh_proxy_cmd = kwargs.get('ssh_proxy_cmd')
if not self.host or not self.port:
proxy_settings = ProxySettings(host='foo', port=1234)
self.assertEqual('foo', proxy_settings.host)
self.assertEqual(1234, proxy_settings.port)
+ self.assertEqual('foo', proxy_settings.https_host)
+ self.assertEqual(1234, proxy_settings.https_port)
self.assertIsNone(proxy_settings.ssh_proxy_cmd)
def test_minimum_kwargs(self):
proxy_settings = ProxySettings(**{'host': 'foo', 'port': 1234})
self.assertEqual('foo', proxy_settings.host)
self.assertEqual(1234, proxy_settings.port)
+ self.assertEqual('foo', proxy_settings.https_host)
+ self.assertEqual(1234, proxy_settings.https_port)
self.assertIsNone(proxy_settings.ssh_proxy_cmd)
def test_all(self):
- proxy_settings = ProxySettings(host='foo', port=1234,
- ssh_proxy_cmd='proxy command')
+ proxy_settings = ProxySettings(
+ host='foo', port=1234, https_host='bar', https_port=2345,
+ ssh_proxy_cmd='proxy command')
self.assertEqual('foo', proxy_settings.host)
self.assertEqual(1234, proxy_settings.port)
+ self.assertEqual('bar', proxy_settings.https_host)
+ self.assertEqual(2345, proxy_settings.https_port)
self.assertEqual('proxy command', proxy_settings.ssh_proxy_cmd)
def test_all_kwargs(self):
proxy_settings = ProxySettings(
- **{'host': 'foo', 'port': 1234, 'ssh_proxy_cmd': 'proxy command'})
+ **{'host': 'foo', 'port': 1234, 'https_host': 'bar',
+ 'https_port': 2345, 'ssh_proxy_cmd': 'proxy command'})
self.assertEqual('foo', proxy_settings.host)
self.assertEqual(1234, proxy_settings.port)
+ self.assertEqual('bar', proxy_settings.https_host)
+ self.assertEqual(2345, proxy_settings.https_port)
self.assertEqual('proxy command', proxy_settings.ssh_proxy_cmd)
self.assertEqual('bar', os_creds.password)
self.assertEqual('http://foo.bar:5000/v2', os_creds.auth_url)
self.assertEqual('hello', os_creds.project_name)
+ self.assertEqual(2, os_creds.identity_api_version)
+ self.assertEqual(2, os_creds.image_api_version)
+ self.assertEqual(2, os_creds.compute_api_version)
+ self.assertEqual(1, os_creds.heat_api_version)
+ self.assertEqual('default', os_creds.user_domain_id)
+ self.assertEqual('default', os_creds.project_domain_id)
+ self.assertEqual('admin', os_creds.interface)
+ self.assertTrue(os_creds.cacert)
self.assertIsNone(os_creds.proxy_settings)
def test_minimal_kwargs(self):
self.assertEqual('bar', os_creds.password)
self.assertEqual('http://foo.bar:5000/v2', os_creds.auth_url)
self.assertEqual('hello', os_creds.project_name)
+ self.assertEqual(2, os_creds.identity_api_version)
+ self.assertEqual(2, os_creds.image_api_version)
+ self.assertEqual(2, os_creds.compute_api_version)
+ self.assertEqual(1, os_creds.heat_api_version)
+ self.assertEqual('default', os_creds.user_domain_id)
+ self.assertEqual('default', os_creds.project_domain_id)
+ self.assertEqual('admin', os_creds.interface)
+ self.assertTrue(os_creds.cacert)
+ self.assertIsNone(os_creds.proxy_settings)
+
+ def test_all_kwargs_versions_str(self):
+ os_creds = OSCreds(
+ **{'username': 'foo', 'password': 'bar',
+ 'auth_url': 'http://foo.bar:5000/v2', 'project_name': 'hello',
+ 'identity_api_version': '5', 'image_api_version': '6',
+ 'compute_api_version': '7', 'heat_api_version': '8.0'})
+ self.assertEqual('foo', os_creds.username)
+ self.assertEqual('bar', os_creds.password)
+ self.assertEqual('http://foo.bar:5000/v2', os_creds.auth_url)
+ self.assertEqual('hello', os_creds.project_name)
+ self.assertEqual(5, os_creds.identity_api_version)
+ self.assertEqual(6, os_creds.image_api_version)
+ self.assertEqual(7, os_creds.compute_api_version)
+ self.assertEqual(8.0, os_creds.heat_api_version)
+ self.assertEqual('default', os_creds.user_domain_id)
+ self.assertEqual('default', os_creds.project_domain_id)
+ self.assertEqual('admin', os_creds.interface)
+ self.assertTrue(os_creds.cacert)
+ self.assertIsNone(os_creds.proxy_settings)
+
+ def test_all_kwargs_versions_num(self):
+ os_creds = OSCreds(
+ **{'username': 'foo', 'password': 'bar',
+ 'auth_url': 'http://foo.bar:5000/v2', 'project_name': 'hello',
+ 'identity_api_version': 5, 'image_api_version': 6,
+ 'compute_api_version': 7, 'heat_api_version': 8.0})
+ self.assertEqual('foo', os_creds.username)
+ self.assertEqual('bar', os_creds.password)
+ self.assertEqual('http://foo.bar:5000/v2', os_creds.auth_url)
+ self.assertEqual('hello', os_creds.project_name)
+ self.assertEqual(5, os_creds.identity_api_version)
+ self.assertEqual(6, os_creds.image_api_version)
+ self.assertEqual(7, os_creds.compute_api_version)
+ self.assertEqual(8.0, os_creds.heat_api_version)
+ self.assertEqual('default', os_creds.user_domain_id)
+ self.assertEqual('default', os_creds.project_domain_id)
+ self.assertEqual('admin', os_creds.interface)
+ self.assertTrue(os_creds.cacert)
self.assertIsNone(os_creds.proxy_settings)
def test_proxy_settings_obj(self):
self.assertEqual('bar', os_creds.password)
self.assertEqual('http://foo.bar:5000/v2', os_creds.auth_url)
self.assertEqual('hello', os_creds.project_name)
+ self.assertEqual(2, os_creds.identity_api_version)
+ self.assertEqual(2, os_creds.image_api_version)
+ self.assertEqual(2, os_creds.compute_api_version)
+ self.assertEqual(1, os_creds.heat_api_version)
+ self.assertEqual('default', os_creds.user_domain_id)
+ self.assertEqual('default', os_creds.project_domain_id)
+ self.assertEqual('admin', os_creds.interface)
+ self.assertTrue(os_creds.cacert)
self.assertEqual('foo', os_creds.proxy_settings.host)
self.assertEqual(1234, os_creds.proxy_settings.port)
self.assertIsNone(os_creds.proxy_settings.ssh_proxy_cmd)
self.assertEqual('bar', os_creds.password)
self.assertEqual('http://foo.bar:5000/v2', os_creds.auth_url)
self.assertEqual('hello', os_creds.project_name)
+ self.assertEqual(2, os_creds.identity_api_version)
+ self.assertEqual(2, os_creds.image_api_version)
+ self.assertEqual(2, os_creds.compute_api_version)
+ self.assertEqual(1, os_creds.heat_api_version)
+ self.assertEqual('default', os_creds.user_domain_id)
+ self.assertEqual('default', os_creds.project_domain_id)
+ self.assertEqual('admin', os_creds.interface)
+ self.assertTrue(os_creds.cacert)
self.assertEqual('foo', os_creds.proxy_settings.host)
self.assertEqual(1234, os_creds.proxy_settings.port)
self.assertIsNone(os_creds.proxy_settings.ssh_proxy_cmd)
self.assertEqual('bar', os_creds.password)
self.assertEqual('http://foo.bar:5000/v2', os_creds.auth_url)
self.assertEqual('hello', os_creds.project_name)
+ self.assertEqual(2, os_creds.identity_api_version)
+ self.assertEqual(2, os_creds.image_api_version)
+ self.assertEqual(2, os_creds.compute_api_version)
+ self.assertEqual(1, os_creds.heat_api_version)
+ self.assertEqual('default', os_creds.user_domain_id)
+ self.assertEqual('default', os_creds.project_domain_id)
+ self.assertEqual('admin', os_creds.interface)
+ self.assertTrue(os_creds.cacert)
self.assertEqual('foo', os_creds.proxy_settings.host)
self.assertEqual(1234, os_creds.proxy_settings.port)
self.assertIsNone(os_creds.proxy_settings.ssh_proxy_cmd)
self.assertEqual('bar', os_creds.password)
self.assertEqual('http://foo.bar:5000/v2', os_creds.auth_url)
self.assertEqual('hello', os_creds.project_name)
+ self.assertEqual(2, os_creds.identity_api_version)
+ self.assertEqual(2, os_creds.image_api_version)
+ self.assertEqual(2, os_creds.compute_api_version)
+ self.assertEqual(1, os_creds.heat_api_version)
+ self.assertEqual('default', os_creds.user_domain_id)
+ self.assertEqual('default', os_creds.project_domain_id)
+ self.assertEqual('admin', os_creds.interface)
+ self.assertTrue(os_creds.cacert)
self.assertEqual('foo', os_creds.proxy_settings.host)
self.assertEqual(1234, os_creds.proxy_settings.port)
self.assertIsNone(os_creds.proxy_settings.ssh_proxy_cmd)
"""
from snaps.openstack.utils import nova_utils
nova = nova_utils.nova_client(self.admin_os_creds)
- zones = nova_utils.get_nova_availability_zones(nova)
+ zone_hosts = nova_utils.get_availability_zone_hosts(nova)
# Create Instance on each server/zone
ctr = 0
- for zone in zones:
+ for zone in zone_hosts:
inst_name = self.vm_inst_name + '-' + zone
ctr += 1
port_settings = PortSettings(
# Validate instances to ensure they've been deployed to the correct
# server
index = 0
- for zone in zones:
+ for zone in zone_hosts:
creator = self.inst_creators[index]
self.assertTrue(creator.vm_active(block=True))
info = creator.get_vm_info()
from snaps.openstack.create_network import NetworkSettings, SubnetSettings
from snaps.openstack.create_router import RouterSettings
from snaps.openstack.os_credentials import OSCreds, ProxySettings
-from snaps.openstack.utils import glance_utils
__author__ = 'spisarski'
if not proj_name:
proj_name = config.get('OS_TENANT_NAME')
- proj_domain_id = 'default'
- user_domain_id = 'default'
-
- if config.get('OS_PROJECT_DOMAIN_ID'):
- proj_domain_id = config['OS_PROJECT_DOMAIN_ID']
- if config.get('OS_USER_DOMAIN_ID'):
- user_domain_id = config['OS_USER_DOMAIN_ID']
- if config.get('OS_IDENTITY_API_VERSION'):
- version = int(config['OS_IDENTITY_API_VERSION'])
- else:
- version = 2
-
proxy_settings = None
if proxy_settings_str:
tokens = re.split(':', proxy_settings_str)
if config.get('OS_INTERFACE'):
interface = config.get('OS_INTERFACE')
- os_creds = OSCreds(username=config['OS_USERNAME'],
- password=config['OS_PASSWORD'],
- auth_url=config['OS_AUTH_URL'],
- project_name=proj_name,
- identity_api_version=version,
- user_domain_id=user_domain_id,
- project_domain_id=proj_domain_id,
- interface=interface,
- proxy_settings=proxy_settings,
- cacert=https_cacert)
+ creds_dict = {
+ 'username': config['OS_USERNAME'],
+ 'password': config['OS_PASSWORD'],
+ 'auth_url': config['OS_AUTH_URL'],
+ 'project_name': proj_name,
+ 'identity_api_version': config.get('OS_IDENTITY_API_VERSION'),
+ 'image_api_version': config.get('OS_IMAGE_API_VERSION'),
+ 'network_api_version': config.get('OS_NETWORK_API_VERSION'),
+ 'compute_api_version': config.get('OS_COMPUTE_API_VERSION'),
+ 'heat_api_version': config.get('OS_HEAT_API_VERSION'),
+ 'user_domain_id': config.get('OS_USER_DOMAIN_ID'),
+ 'project_domain_id': config.get('OS_PROJECT_DOMAIN_ID'),
+ 'interface': interface,
+ 'proxy_settings': proxy_settings,
+ 'cacert': https_cacert}
else:
logger.info('Reading development os_env file - ' + dev_os_env_file)
config = file_utils.read_yaml(dev_os_env_file)
- identity_api_version = config.get('identity_api_version')
- if not identity_api_version:
- identity_api_version = 2
-
- image_api_version = config.get('image_api_version')
- if not image_api_version:
- image_api_version = glance_utils.VERSION_2
proxy_settings = None
proxy_str = config.get('http_proxy')
host=tokens[0], port=tokens[1],
ssh_proxy_cmd=config.get('ssh_proxy_cmd'))
- os_creds = OSCreds(username=config['username'],
- password=config['password'],
- auth_url=config['os_auth_url'],
- project_name=config['project_name'],
- identity_api_version=identity_api_version,
- image_api_version=image_api_version,
- proxy_settings=proxy_settings)
-
+ creds_dict = {
+ 'username': config['username'],
+ 'password': config['password'],
+ 'auth_url': config['os_auth_url'],
+ 'project_name': config['project_name'],
+ 'identity_api_version': config.get('identity_api_version'),
+ 'image_api_version': config.get('image_api_version'),
+ 'network_api_version': config.get('network_api_version'),
+ 'compute_api_version': config.get('compute_api_version'),
+ 'heat_api_version': config.get('heat_api_version'),
+ 'user_domain_id': config.get('user_domain_id'),
+ 'project_domain_id': config.get('project_domain_id'),
+ 'interface': config.get('interface'),
+ 'proxy_settings': proxy_settings, 'cacert': config.get('cacert')}
+
+ os_creds = OSCreds(**creds_dict)
logger.info('OS Credentials = %s', os_creds)
return os_creds
# limitations under the License.
import logging
import pkg_resources
+import requests
+from requests.packages.urllib3.exceptions import InsecureRequestWarning
import uuid
import unittest
dev_os_env_file = pkg_resources.resource_filename(
'snaps.openstack.tests.conf', 'os_env.yaml')
+requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
class OSComponentTestCase(unittest.TestCase):
:return: the client
"""
logger.debug('Retrieving Nova Client')
- return Client(1, session=keystone_utils.keystone_session(os_creds))
+ return Client(os_creds.heat_api_version,
+ session=keystone_utils.keystone_session(os_creds))
def get_stack_by_name(heat_cli, stack_name):
if stack_settings.template:
args['template'] = stack_settings.template
else:
- args['template'] = parse_heat_template_str(file_utils.read_file(stack_settings.template_path))
+ args['template'] = parse_heat_template_str(
+ file_utils.read_file(stack_settings.template_path))
args['stack_name'] = stack_settings.name
if stack_settings.env_values:
def parse_heat_template_str(tmpl_str):
- """Takes a heat template string, performs some simple validation and returns a dict containing the parsed structure.
- This function supports both JSON and YAML Heat template formats.
+ """
+ Takes a heat template string, performs some simple validation and returns a
+ dict containing the parsed structure. This function supports both JSON and
+ YAML Heat template formats.
"""
if tmpl_str.startswith('{'):
tpl = jsonutils.loads(tmpl_str)
logger = logging.getLogger('keystone_utils')
-V2_VERSION = 'v2.0'
+V2_VERSION_NUM = 2.0
+V2_VERSION_STR = 'v' + str(V2_VERSION_NUM)
def get_session_auth(os_creds):
req_session.proxies = {
'http':
os_creds.proxy_settings.host + ':' +
- os_creds.proxy_settings.port}
+ os_creds.proxy_settings.port,
+ 'https':
+ os_creds.proxy_settings.https_host + ':' +
+ os_creds.proxy_settings.https_port
+ }
return session.Session(auth=auth, session=req_session,
verify=os_creds.cacert)
raise KeystoneException(
'Cannot lookup project without the proper credentials')
- if keystone.version == V2_VERSION:
+ if keystone.version == V2_VERSION_STR:
projects = keystone.tenants.list()
else:
projects = keystone.projects.list(**{'name': project_name})
:param project_settings: the project configuration
:return: SNAPS-OO Project domain object
"""
- if keystone.version == V2_VERSION:
+ if keystone.version == V2_VERSION_STR:
os_project = keystone.tenants.create(
project_settings.name, project_settings.description,
project_settings.enabled)
:param keystone: the Keystone clien
:param project: the SNAPS-OO Project domain object
"""
- if keystone.version == V2_VERSION:
+ if keystone.version == V2_VERSION_STR:
keystone.tenants.delete(project.id)
else:
keystone.projects.delete(project.id)
project = get_project(keystone=keystone,
project_name=user_settings.project_name)
- if keystone.version == V2_VERSION:
+ if keystone.version == V2_VERSION_STR:
project_id = None
if project:
project_id = project.id
:param project: the OpenStack project object (only required for v2)
:return: a list of SNAPS-OO Role domain objects
"""
- if keystone.version == V2_VERSION:
+ if keystone.version == V2_VERSION_STR:
os_user = __get_os_user(keystone, user)
roles = keystone.roles.roles_for_user(os_user, project)
else:
"""
os_role = get_role_by_id(keystone, role.id)
- if keystone.version == V2_VERSION:
+ if keystone.version == V2_VERSION_STR:
keystone.roles.add_user_role(user, os_role, tenant=project)
else:
keystone.roles.grant(os_role, user=user, project=project)
'key_name': keypair_name,
'security_groups':
instance_settings.security_group_names,
- 'userdata': instance_settings.userdata,
- 'availability_zone':
- instance_settings.availability_zone}
+ 'userdata': instance_settings.userdata}
+
+ if instance_settings.availability_zone:
+ args['availability_zone'] = instance_settings.availability_zone
+
server = nova.servers.create(**args)
return VmInst(name=server.name, inst_id=server.id,
networks=server.networks)
nova.keypairs.delete(key.id)
-def get_nova_availability_zones(nova):
+def get_availability_zone_hosts(nova, zone_name='nova'):
"""
Returns the names of all nova active compute servers
:param nova: the Nova client
+ :param zone_name: the Nova client
:return: a list of compute server names
"""
out = list()
zones = nova.availability_zones.list()
for zone in zones:
- if zone.zoneName == 'nova':
+ if zone.zoneName == zone_name and zone.hosts:
for key, host in zone.hosts.items():
if host['nova-compute']['available']:
out.append(zone.zoneName + ':' + key)
if status == create_stack.STATUS_CREATE_COMPLETE:
is_active = True
break
+ elif status == create_stack.STATUS_CREATE_FAILED:
+ is_active = False
+ break
time.sleep(3)
logger = logging.getLogger('test_runner')
ARG_NOT_SET = "argument not set"
-LOG_LEVELS = {'FATAL': logging.FATAL, 'CRITICAL': logging.CRITICAL, 'ERROR': logging.ERROR, 'WARN': logging.WARN,
+LOG_LEVELS = {'FATAL': logging.FATAL, 'CRITICAL': logging.CRITICAL,
+ 'ERROR': logging.ERROR, 'WARN': logging.WARN,
'INFO': logging.INFO, 'DEBUG': logging.DEBUG}
-def __create_test_suite(source_filename, ext_net_name, proxy_settings, ssh_proxy_cmd, run_unit_tests,
- run_connection_tests, run_api_tests, run_integration_tests, run_staging_tests, flavor_metadata,
- image_metadata, use_keystone, use_floating_ips, log_level):
+def __create_test_suite(
+ source_filename, ext_net_name, proxy_settings, ssh_proxy_cmd,
+ run_unit_tests, run_connection_tests, run_api_tests,
+ run_integration_tests, run_staging_tests, flavor_metadata,
+ image_metadata, use_keystone, use_floating_ips, continuous_integration,
+ log_level):
"""
Compiles the tests that should run
:param source_filename: the OpenStack credentials file (required)
- :param ext_net_name: the name of the external network to use for floating IPs (required)
- :param run_unit_tests: when true, the tests not requiring OpenStack will be added to the test suite
- :param run_connection_tests: when true, the tests that perform simple connections to OpenStack are executed
- :param run_api_tests: when true, the tests that perform simple API calls to OpenStack are executed
+ :param ext_net_name: the name of the external network to use for floating
+ IPs (required)
+ :param run_unit_tests: when true, the tests not requiring OpenStack will be
+ added to the test suite
+ :param run_connection_tests: when true, the tests that perform simple
+ connections to OpenStack are executed
+ :param run_api_tests: when true, the tests that perform simple API calls to
+ OpenStack are executed
:param run_integration_tests: when true, the integration tests are executed
:param run_staging_tests: when true, the staging tests are executed
:param proxy_settings: <host>:<port> of the proxy server (optional)
- :param ssh_proxy_cmd: the command used to connect via SSH over some proxy server (optional)
- :param flavor_metadata: dict() object containing the metadata for flavors created for test VM instance
- :param image_metadata: dict() object containing the metadata for overriding default images within the tests
- :param use_keystone: when true, tests creating users and projects will be exercised and must be run on a host that
+ :param ssh_proxy_cmd: the command used to connect via SSH over some proxy
+ server (optional)
+ :param flavor_metadata: dict() object containing the metadata for flavors
+ created for test VM instance
+ :param image_metadata: dict() object containing the metadata for overriding
+ default images within the tests
+ :param use_keystone: when true, tests creating users and projects will be
+ exercised and must be run on a host that
has access to the cloud's administrative network
- :param use_floating_ips: when true, tests requiring floating IPs will be executed
+ :param use_floating_ips: when true, tests requiring floating IPs will be
+ executed
+ :param continuous_integration: when true, tests for CI will be run
:param log_level: the logging level
:return:
"""
suite = unittest.TestSuite()
- os_creds = openstack_tests.get_credentials(os_env_file=source_filename, proxy_settings_str=proxy_settings,
- ssh_proxy_cmd=ssh_proxy_cmd)
+ os_creds = openstack_tests.get_credentials(
+ os_env_file=source_filename, proxy_settings_str=proxy_settings,
+ ssh_proxy_cmd=ssh_proxy_cmd)
# Tests that do not require a remote connection to an OpenStack cloud
if run_unit_tests:
# Basic connection tests
if run_connection_tests:
test_suite_builder.add_openstack_client_tests(
- suite=suite, os_creds=os_creds, ext_net_name=ext_net_name, use_keystone=use_keystone, log_level=log_level)
+ suite=suite, os_creds=os_creds, ext_net_name=ext_net_name,
+ use_keystone=use_keystone, log_level=log_level)
# Tests the OpenStack API calls
if run_api_tests:
test_suite_builder.add_openstack_api_tests(
- suite=suite, os_creds=os_creds, ext_net_name=ext_net_name, use_keystone=use_keystone,
- image_metadata=image_metadata, log_level=log_level)
+ suite=suite, os_creds=os_creds, ext_net_name=ext_net_name,
+ use_keystone=use_keystone, image_metadata=image_metadata,
+ log_level=log_level)
# Long running integration type tests
if run_integration_tests:
test_suite_builder.add_openstack_integration_tests(
- suite=suite, os_creds=os_creds, ext_net_name=ext_net_name, use_keystone=use_keystone,
- flavor_metadata=flavor_metadata, image_metadata=image_metadata, use_floating_ips=use_floating_ips,
+ suite=suite, os_creds=os_creds, ext_net_name=ext_net_name,
+ use_keystone=use_keystone, flavor_metadata=flavor_metadata,
+ image_metadata=image_metadata, use_floating_ips=use_floating_ips,
log_level=log_level)
if run_staging_tests:
test_suite_builder.add_openstack_staging_tests(
- suite=suite, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level)
+ suite=suite, os_creds=os_creds, ext_net_name=ext_net_name,
+ log_level=log_level)
+
+ if continuous_integration:
+ test_suite_builder.add_openstack_ci_tests(
+ suite=suite, os_creds=os_creds, ext_net_name=ext_net_name,
+ use_keystone=use_keystone, flavor_metadata=flavor_metadata,
+ image_metadata=image_metadata, use_floating_ips=use_floating_ips,
+ log_level=log_level)
return suite
def main(arguments):
"""
Begins running unit tests.
- argv[1] if used must be the source filename else os_env.yaml will be leveraged instead
+ argv[1] if used must be the source filename else os_env.yaml will be
+ leveraged instead
argv[2] if used must be the proxy server <host>:<port>
"""
logger.info('Starting test suite')
connection = arguments.include_connection != ARG_NOT_SET
api = arguments.include_api != ARG_NOT_SET
integration = arguments.include_integration != ARG_NOT_SET
+ ci = arguments.continuous_integration != ARG_NOT_SET
staging = arguments.include_staging != ARG_NOT_SET
- if not unit and not connection and not api and not integration and not staging:
+ if (not unit and not connection and not api and not integration
+ and not staging and not ci):
unit = True
connection = True
api = True
integration = True
- suite = __create_test_suite(arguments.env, arguments.ext_net, arguments.proxy, arguments.ssh_proxy_cmd,
- unit, connection, api, integration, staging, flavor_metadata, image_metadata,
- arguments.use_keystone != ARG_NOT_SET,
- arguments.floating_ips != ARG_NOT_SET, log_level)
+ suite = __create_test_suite(
+ arguments.env, arguments.ext_net, arguments.proxy,
+ arguments.ssh_proxy_cmd, unit, connection, api,
+ integration, staging, flavor_metadata, image_metadata,
+ arguments.use_keystone != ARG_NOT_SET,
+ arguments.floating_ips != ARG_NOT_SET,
+ ci, log_level)
else:
logger.error('Environment file or external network not defined')
exit(1)
i += 1
if result.errors:
- logger.error('Number of errors in test suite - ' + str(len(result.errors)))
+ logger.error('Number of errors in test suite - %s',
+ len(result.errors))
for test, message in result.errors:
logger.error(str(test) + " ERROR with " + message)
if result.failures:
- logger.error('Number of failures in test suite - ' + str(len(result.failures)))
+ logger.error('Number of failures in test suite - %s',
+ len(result.failures))
for test, message in result.failures:
logger.error(str(test) + " FAILED with " + message)
- if (result.errors and len(result.errors) > 0) or (result.failures and len(result.failures) > 0):
+ if ((result.errors and len(result.errors) > 0)
+ or (result.failures and len(result.failures) > 0)):
logger.error('See above for test failures')
exit(1)
else:
- logger.info('All tests completed successfully in run #' + str(i))
+ logger.info('All tests completed successfully in run #%s', i)
- logger.info('Successful completion of ' + str(i) + ' test runs')
+ logger.info('Successful completion of %s test runs', i)
exit(0)
if __name__ == '__main__':
parser = argparse.ArgumentParser()
- parser.add_argument('-e', '--env', dest='env', required=True, help='OpenStack credentials file')
- parser.add_argument('-n', '--net', dest='ext_net', required=True, help='External network name')
- parser.add_argument('-p', '--proxy', dest='proxy', nargs='?', default=None,
- help='Optonal HTTP proxy socket (<host>:<port>)')
- parser.add_argument('-s', '--ssh-proxy-cmd', dest='ssh_proxy_cmd', nargs='?', default=None,
- help='Optonal SSH proxy command value')
- parser.add_argument('-l', '--log-level', dest='log_level', default='INFO',
- help='Logging Level (FATAL|CRITICAL|ERROR|WARN|INFO|DEBUG)')
- parser.add_argument('-u', '--unit-tests', dest='include_unit', default=ARG_NOT_SET, nargs='?',
- help='When argument is set, all tests not requiring OpenStack will be executed')
- parser.add_argument('-c', '--connection-tests', dest='include_connection', default=ARG_NOT_SET, nargs='?',
- help='When argument is set, simple OpenStack connection tests will be executed')
- parser.add_argument('-a', '--api-tests', dest='include_api', default=ARG_NOT_SET, nargs='?',
- help='When argument is set, OpenStack API tests will be executed')
- parser.add_argument('-i', '--integration-tests', dest='include_integration', default=ARG_NOT_SET, nargs='?',
- help='When argument is set, OpenStack integrations tests will be executed')
- parser.add_argument('-st', '--staging-tests', dest='include_staging', default=ARG_NOT_SET, nargs='?',
- help='When argument is set, OpenStack staging tests will be executed')
- parser.add_argument('-f', '--floating-ips', dest='floating_ips', default=ARG_NOT_SET, nargs='?',
- help='When argument is set, all integration tests requiring Floating IPs will be executed')
- parser.add_argument('-k', '--use-keystone', dest='use_keystone', default=ARG_NOT_SET, nargs='?',
- help='When argument is set, the tests will exercise the keystone APIs and must be run on a ' +
- 'machine that has access to the admin network' +
- ' and is able to create users and groups')
- parser.add_argument('-fm', '--flavor-meta', dest='flavor_metadata',
- default='{\"hw:mem_page_size\": \"any\"}',
- help='JSON string to be used as flavor metadata for all test instances created')
- parser.add_argument('-im', '--image-meta', dest='image_metadata_file',
- default=None,
- help='Location of YAML file containing the image metadata')
- parser.add_argument('-r', '--num-runs', dest='num_runs', default=1,
- help='Number of test runs to execute (default 1)')
+ parser.add_argument(
+ '-e', '--env', dest='env', required=True,
+ help='OpenStack credentials file')
+ parser.add_argument(
+ '-n', '--net', dest='ext_net', required=True,
+ help='External network name')
+ parser.add_argument(
+ '-p', '--proxy', dest='proxy', nargs='?', default=None,
+ help='Optonal HTTP proxy socket (<host>:<port>)')
+ parser.add_argument(
+ '-s', '--ssh-proxy-cmd', dest='ssh_proxy_cmd', nargs='?', default=None,
+ help='Optonal SSH proxy command value')
+ parser.add_argument(
+ '-l', '--log-level', dest='log_level', default='INFO',
+ help='Logging Level (FATAL|CRITICAL|ERROR|WARN|INFO|DEBUG)')
+ parser.add_argument(
+ '-u', '--unit-tests', dest='include_unit', default=ARG_NOT_SET,
+ nargs='?', help='When argument is set, all tests not requiring '
+ 'OpenStack will be executed')
+ parser.add_argument(
+ '-c', '--connection-tests', dest='include_connection',
+ default=ARG_NOT_SET, nargs='?',
+ help='When argument is set, simple OpenStack connection tests will be '
+ 'executed')
+ parser.add_argument(
+ '-a', '--api-tests', dest='include_api', default=ARG_NOT_SET,
+ nargs='?',
+ help='When argument is set, OpenStack API tests will be executed')
+ parser.add_argument(
+ '-i', '--integration-tests', dest='include_integration',
+ default=ARG_NOT_SET, nargs='?',
+ help='When argument is set, OpenStack integrations tests will be '
+ 'executed')
+ parser.add_argument(
+ '-st', '--staging-tests', dest='include_staging', default=ARG_NOT_SET,
+ nargs='?',
+ help='When argument is set, OpenStack staging tests will be executed')
+ parser.add_argument(
+ '-f', '--floating-ips', dest='floating_ips', default=ARG_NOT_SET,
+ nargs='?', help='When argument is set, all integration tests requiring'
+ ' Floating IPs will be executed')
+ parser.add_argument(
+ '-k', '--use-keystone', dest='use_keystone', default=ARG_NOT_SET,
+ nargs='?',
+ help='When argument is set, the tests will exercise the keystone APIs '
+ 'and must be run on a machine that has access to the admin '
+ 'network and is able to create users and groups')
+ parser.add_argument(
+ '-fm', '--flavor-meta', dest='flavor_metadata',
+ help='JSON string to be used as flavor metadata for all test instances'
+ ' created')
+ parser.add_argument(
+ '-im', '--image-meta', dest='image_metadata_file', default=None,
+ help='Location of YAML file containing the image metadata')
+ parser.add_argument(
+ '-ci', '--continuous-integration', dest='continuous_integration',
+ default=ARG_NOT_SET, nargs='?',
+ help='When argument is set, OpenStack integrations tests will be '
+ 'executed')
+ parser.add_argument(
+ '-r', '--num-runs', dest='num_runs', default=1,
+ help='Number of test runs to execute (default 1)')
args = parser.parse_args()
log_level=log_level))
+def add_openstack_ci_tests(
+ suite, os_creds, ext_net_name, use_keystone=True, flavor_metadata=None,
+ image_metadata=None, use_floating_ips=True, log_level=logging.INFO):
+ """
+ Adds tests written for a CI server to run the tests to validate code
+ changes
+ :param suite: the unittest.TestSuite object to which to add the tests
+ :param os_creds: and instance of OSCreds that holds the credentials
+ required by OpenStack
+ :param ext_net_name: the name of an external network on the cloud under
+ test
+ :param use_keystone: when True, tests requiring direct access to Keystone
+ are added as these need to be running on a host that
+ has access to the cloud's private network
+ :param image_metadata: dict() object containing metadata for creating an
+ image with custom config
+ (see YAML files in examples/image-metadata)
+ :param flavor_metadata: dict() object containing the metadata required by
+ your flavor based on your configuration:
+ (i.e. {'hw:mem_page_size': 'large'})
+ :param use_floating_ips: when true, all tests requiring Floating IPs will
+ be added to the suite
+ :param log_level: the logging level
+ :return: None as the tests will be adding to the 'suite' parameter object
+ """
+
+ add_unit_tests(suite)
+
+ add_openstack_client_tests(suite, os_creds, ext_net_name, use_keystone,
+ log_level)
+
+ add_openstack_api_tests(suite, os_creds, ext_net_name, use_keystone,
+ image_metadata, log_level)
+
+ suite.addTest(OSIntegrationTestCase.parameterize(
+ SimpleHealthCheck, os_creds=os_creds, ext_net_name=ext_net_name,
+ use_keystone=use_keystone,
+ flavor_metadata=flavor_metadata, image_metadata=image_metadata,
+ log_level=log_level))
+
+ if use_floating_ips:
+ suite.addTest(OSIntegrationTestCase.parameterize(
+ CreateInstanceSingleNetworkTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, use_keystone=use_keystone,
+ flavor_metadata=flavor_metadata, image_metadata=image_metadata,
+ log_level=log_level))
+
+ suite.addTest(OSIntegrationTestCase.parameterize(
+ CreateStackSuccessTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ use_keystone=use_keystone,
+ flavor_metadata=flavor_metadata, image_metadata=image_metadata,
+ log_level=log_level))
+ suite.addTest(OSIntegrationTestCase.parameterize(
+ CreateStackNegativeTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ use_keystone=use_keystone,
+ flavor_metadata=flavor_metadata, image_metadata=image_metadata,
+ log_level=log_level))
+
+
def add_openstack_staging_tests(suite, os_creds, ext_net_name,
log_level=logging.INFO):
"""
Code Review / snaps.git / commitdiff