This patch moves the floating IP creation out of nova and into neutron.
Other changes include the use of domain objects for VM and Floating IP
instances, addition of new nova_utils tests to exercise the create
server functionality, and more PEP8 compliance.
JIRA: SNAPS-92
Change-Id: I16c12b26b56008901633e90ae307586ad2045f9b
Signed-off-by: spisarski <s.pisarski@cablelabs.com>
-python-novaclient!=7.0.0,<8.0.0,>=6.0.0 # Apache-2.0
+python-novaclient>=6.0.0 # Apache-2.0
python-neutronclient>=5.1.0 # Apache-2.0
python-keystoneclient>=3.8.0 # Apache-2.0
python-glanceclient>=2.5.0 # Apache-2.0
--- /dev/null
+# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
+# and others. All rights reserved.
+#
+# 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 unittest
+from snaps.domain.vm_inst import VmInst, FloatingIp
+
+
+class VmInstDomainObjectTests(unittest.TestCase):
+ """
+ Tests the construction of the snaps.domain.test.Image class
+ """
+
+ def test_construction_positional(self):
+ vm_inst = VmInst('name', 'id')
+ self.assertEqual('name', vm_inst.name)
+ self.assertEqual('id', vm_inst.id)
+
+ def test_construction_named(self):
+ vm_inst = VmInst(inst_id='id', name='name')
+ self.assertEqual('name', vm_inst.name)
+ self.assertEqual('id', vm_inst.id)
+
+
+class FloatingIpDomainObjectTests(unittest.TestCase):
+ """
+ Tests the construction of the snaps.domain.test.Image class
+ """
+
+ def test_construction_positional(self):
+ vm_inst = FloatingIp('id-123', '10.0.0.1')
+ self.assertEqual('id-123', vm_inst.id)
+ self.assertEqual('10.0.0.1', vm_inst.ip)
+
+ def test_construction_named(self):
+ vm_inst = FloatingIp(ip='10.0.0.1', inst_id='id-123')
+ self.assertEqual('id-123', vm_inst.id)
+ self.assertEqual('10.0.0.1', vm_inst.ip)
--- /dev/null
+# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
+# and others. All rights reserved.
+#
+# 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.
+
+
+class VmInst:
+ """
+ SNAPS domain object for Images. Should contain attributes that
+ are shared amongst cloud providers
+ """
+ def __init__(self, name, inst_id):
+ """
+ Constructor
+ :param name: the image's name
+ :param inst_id: the instance's id
+ """
+ self.name = name
+ self.id = inst_id
+
+
+class FloatingIp:
+ """
+ SNAPS domain object for Images. Should contain attributes that
+ are shared amongst cloud providers
+ """
+ def __init__(self, inst_id, ip):
+ """
+ Constructor
+ :param inst_id: the floating ip's id
+ :param ip: the IP address
+ """
+ self.id = inst_id
+ self.ip = ip
logger.info(
'Found existing machine with name - %s',
self.instance_settings.name)
- fips = self.__nova.floating_ips.list()
+ fips = neutron_utils.get_floating_ips(self.__nova)
for fip in fips:
if fip.instance_id == server.id:
self.__floating_ips.append(fip)
active, error, or timeout waiting. Floating IPs will be
assigned after active when block=True
"""
- nics = []
- for key, port in self.__ports:
- kv = dict()
- kv['port-id'] = port['port']['id']
- nics.append(kv)
-
- logger.info('Creating VM with name - ' + self.instance_settings.name)
- keypair_name = None
- if self.keypair_settings:
- keypair_name = self.keypair_settings.name
-
- flavor = nova_utils.get_flavor_by_name(self.__nova,
- self.instance_settings.flavor)
- if not flavor:
- raise Exception(
- 'Flavor not found with name - %s',
- self.instance_settings.flavor)
-
- image = glance_utils.get_image(
- glance_utils.glance_client(self.__os_creds),
- self.image_settings.name)
- if image:
- self.__vm = self.__nova.servers.create(
- name=self.instance_settings.name,
- flavor=flavor,
- image=image,
- nics=nics,
- key_name=keypair_name,
- security_groups=self.instance_settings.security_group_names,
- userdata=self.instance_settings.userdata,
- availability_zone=self.instance_settings.availability_zone)
-
- else:
- raise Exception(
- 'Cannot create instance, image cannot be located with name %s',
- self.image_settings.name)
-
- logger.info(
- 'Created instance with name - %s', self.instance_settings.name)
+ glance = glance_utils.glance_client(self.__os_creds)
+ self.__vm = nova_utils.create_server(
+ self.__nova, self.__neutron, glance, self.instance_settings,
+ self.image_settings, self.keypair_settings)
+ logger.info('Created instance with name - %s',
+ self.instance_settings.name)
if block:
if not self.vm_active(block=True):
'Fatal error, VM did not become ACTIVE within the alloted '
'time')
- # TODO - the call above should add security groups. The return object
- # shows they exist but the association had never been made by
- # OpenStack. This call is here to ensure they have been added
+ # Create server should do this but found it needed to occur here
for sec_grp_name in self.instance_settings.security_group_names:
if self.vm_active(block=True):
nova_utils.add_security_group(self.__nova, self.__vm,
if ext_gateway:
subnet = neutron_utils.get_subnet_by_name(
self.__neutron, floating_ip_setting.subnet_name)
- floating_ip = nova_utils.create_floating_ip(self.__nova,
- ext_gateway)
+ floating_ip = neutron_utils.create_floating_ip(
+ self.__neutron, ext_gateway)
self.__floating_ips.append(floating_ip)
self.__floating_ip_dict[floating_ip_setting.name] = floating_ip
for floating_ip in self.__floating_ips:
try:
logger.info('Deleting Floating IP - ' + floating_ip.ip)
- nova_utils.delete_floating_ip(self.__nova, floating_ip)
+ neutron_utils.delete_floating_ip(self.__neutron, floating_ip)
except Exception as e:
logger.error('Error deleting Floating IP - ' + str(e))
self.__floating_ips = list()
while count > 0:
logger.debug('Attempting to add floating IP to instance')
try:
- self.__vm.add_floating_ip(floating_ip, ip)
+ nova_utils.add_floating_ip_to_server(
+ self.__nova, self.__vm, floating_ip, ip)
logger.info(
'Added floating IP %s to port IP %s on instance %s',
floating_ip.ip, ip, self.instance_settings.name)
Returns the latest version of this server object from OpenStack
:return: Server object
"""
- return nova_utils.get_latest_server_object(self.__nova, self.__vm)
+ return self.__vm
+
+ def get_os_vm_server_obj(self):
+ """
+ Returns the OpenStack server object
+ :return: the server object
+ """
+ return nova_utils.get_latest_server_os_object(self.__nova, self.__vm)
def get_port_ip(self, port_name, subnet_name=None):
"""
elif len(self.__floating_ips) > 0:
return self.__floating_ips[0]
- def __config_nic(self, nic_name, port, floating_ip):
+ def __config_nic(self, nic_name, port, ip):
"""
Although ports/NICs can contain multiple IPs, this code currently only
supports the first.
:param nic_name: Name of the interface
:param port: The port information containing the expected IP values.
- :param floating_ip: The floating IP on which to apply the playbook.
+ :param ip: The IP on which to apply the playbook.
:return: the return value from ansible
"""
- ip = port['port']['fixed_ips'][0]['ip_address']
+ port_ip = port['port']['fixed_ips'][0]['ip_address']
variables = {
- 'floating_ip': floating_ip,
+ 'floating_ip': ip,
'nic_name': nic_name,
- 'nic_ip': ip
+ 'nic_ip': port_ip
}
if self.image_settings.nic_config_pb_loc and self.keypair_settings:
if not self.__vm:
return False
- instance = self.__nova.servers.get(self.__vm.id)
+ instance = nova_utils.get_latest_server_os_object(
+ self.__nova, self.__vm)
if not instance:
logger.warning('Cannot find instance with id - ' + self.__vm.id)
return False
if kwargs.get('security_group_names'):
if isinstance(kwargs['security_group_names'], list):
- self.security_group_names = kwargs['security_group_names']
+ self.security_group_names = kwargs['security_group_names']
elif isinstance(kwargs['security_group_names'], set):
self.security_group_names = kwargs['security_group_names']
elif isinstance(kwargs['security_group_names'], str):
"""
super(self.__class__, self).__start__()
+ self.nova = nova_utils.nova_client(self.os_creds)
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.vm_inst_name = guid + '-inst'
self.port_1_name = guid + 'port-1'
self.assertTrue(self.inst_creator.vm_active(block=True))
- self.assertTrue(check_dhcp_lease(vm, ip))
+ self.assertTrue(check_dhcp_lease(self.nova, vm, ip))
class CreateInstanceSimpleTests(OSIntegrationTestCase):
"""
super(self.__class__, self).__start__()
+ self.nova = nova_utils.nova_client(self.os_creds)
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.keypair_priv_filepath = 'tmp/' + guid
self.keypair_pub_filepath = self.keypair_priv_filepath + '.pub'
self.assertTrue(inst_creator.vm_active(block=True))
ip = inst_creator.get_port_ip(port_settings.name)
- self.assertTrue(check_dhcp_lease(vm_inst, ip))
+ self.assertTrue(check_dhcp_lease(self.nova, vm_inst, ip))
inst_creator.add_security_group(
self.sec_grp_creator.get_security_group())
self.assertTrue(inst_creator.vm_active(block=True))
ip = inst_creator.get_port_ip(port_settings.name)
- self.assertTrue(check_dhcp_lease(vm_inst, ip))
+ self.assertTrue(check_dhcp_lease(self.nova, vm_inst, ip))
inst_creator.add_security_group(
self.sec_grp_creator.get_security_group())
for zone in zones:
creator = self.inst_creators[index]
self.assertTrue(creator.vm_active(block=True))
- vm = creator.get_vm_inst()
+ vm = creator.get_os_vm_server_obj()
deployed_zone = vm._info['OS-EXT-AZ:availability_zone']
deployed_host = vm._info['OS-EXT-SRV-ATTR:host']
self.assertEqual(zone, deployed_zone + ':' + deployed_host)
"""
super(self.__class__, self).__start__()
+ self.nova = nova_utils.nova_client(self.os_creds)
+
# Initialize for tearDown()
self.image_creator = None
self.network_creators = list()
self.assertTrue(self.inst_creator.vm_active(block=True))
ip = self.inst_creator.get_port_ip(ports_settings[0].name)
- self.assertTrue(check_dhcp_lease(vm_inst, ip))
+ self.assertTrue(check_dhcp_lease(self.nova, vm_inst, ip))
# Add security group to VM
self.inst_creator.add_security_group(
self.sec_grp_creators.append(sec_grp_creator)
# Check that group has not been added
- self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
- sec_grp_settings.name))
+ self.assertFalse(inst_has_sec_grp(
+ self.nova, self.inst_creator.get_vm_inst(), sec_grp_settings.name))
# Add security group to instance after activated
self.inst_creator.add_security_group(sec_grp)
# Validate that security group has been added
- self.assertTrue(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
- sec_grp_settings.name))
+ self.assertTrue(inst_has_sec_grp(
+ self.nova, self.inst_creator.get_vm_inst(), sec_grp_settings.name))
def test_add_invalid_security_group(self):
"""
self.sec_grp_creators.append(sec_grp_creator)
# Check that group has not been added
- self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
- sec_grp_settings.name))
+ self.assertFalse(inst_has_sec_grp(
+ self.nova, self.inst_creator.get_vm_inst(), sec_grp_settings.name))
# Add security group to instance after activated
self.assertFalse(self.inst_creator.add_security_group(sec_grp))
# Validate that security group has been added
- self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
- sec_grp_settings.name))
+ self.assertFalse(inst_has_sec_grp(
+ self.nova, self.inst_creator.get_vm_inst(), sec_grp_settings.name))
def test_remove_security_group(self):
"""
self.assertIsNotNone(vm_inst)
# Check that group has been added
- self.assertTrue(inst_has_sec_grp(vm_inst, sec_grp_settings.name))
+ self.assertTrue(inst_has_sec_grp(
+ self.nova, vm_inst, sec_grp_settings.name))
# Add security group to instance after activated
self.assertTrue(self.inst_creator.remove_security_group(sec_grp))
# Validate that security group has been added
- self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
- sec_grp_settings.name))
+ self.assertFalse(inst_has_sec_grp(
+ self.nova, self.inst_creator.get_vm_inst(), sec_grp_settings.name))
def test_remove_security_group_never_added(self):
"""
self.assertIsNotNone(vm_inst)
# Check that group has been added
- self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
- sec_grp_settings.name))
+ self.assertFalse(inst_has_sec_grp(
+ self.nova, self.inst_creator.get_vm_inst(), sec_grp_settings.name))
# Add security group to instance after activated
self.assertFalse(self.inst_creator.remove_security_group(sec_grp))
# Validate that security group has been added
- self.assertFalse(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
- sec_grp_settings.name))
+ self.assertFalse(inst_has_sec_grp(
+ self.nova, self.inst_creator.get_vm_inst(), sec_grp_settings.name))
def test_add_same_security_group(self):
"""
self.assertIsNotNone(vm_inst)
# Check that group has been added
- self.assertTrue(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
- sec_grp_settings.name))
+ self.assertTrue(inst_has_sec_grp(
+ self.nova, self.inst_creator.get_vm_inst(), sec_grp_settings.name))
# Add security group to instance after activated
self.assertTrue(self.inst_creator.add_security_group(sec_grp))
# Validate that security group has been added
- self.assertTrue(inst_has_sec_grp(self.inst_creator.get_vm_inst(),
- sec_grp_settings.name))
+ self.assertTrue(inst_has_sec_grp(
+ self.nova, self.inst_creator.get_vm_inst(), sec_grp_settings.name))
-def inst_has_sec_grp(vm_inst, sec_grp_name):
+def inst_has_sec_grp(nova, vm_inst, sec_grp_name):
"""
Returns true if instance has a security group of a given name
+ :param nova: the nova client
+ :param vm_inst: the VmInst domain object
+ :param sec_grp_name: the name of the security group to validate
:return:
"""
- if not hasattr(vm_inst, 'security_groups'):
+ vm = nova_utils.get_latest_server_os_object(nova, vm_inst)
+ if not hasattr(vm, 'security_groups'):
return False
found = False
- for sec_grp_dict in vm_inst.security_groups:
+ for sec_grp_dict in vm.security_groups:
if sec_grp_name in sec_grp_dict['name']:
found = True
break
self.assertTrue(self.inst_creator.vm_active(block=True))
-def check_dhcp_lease(vm, ip, timeout=160):
+def check_dhcp_lease(nova_client, vm_domain, ip, timeout=160):
"""
Returns true if the expected DHCP lease has been acquired
- :param vm: the OpenStack VM instance object
+ :param nova_client: the nova client
+ :param vm_domain: the SNAPS VM instance domain object
:param ip: the IP address to look for
:param timeout: how long to query for IP address
:return:
logger.info("Looking for IP %s in the console log" % ip)
full_log = ''
while timeout > time.time() - start_time:
+ vm = nova_utils.get_latest_server_os_object(nova_client, vm_domain)
output = vm.get_console_output()
full_log = full_log + output
if re.search(ip, output):
logger.debug('Image metadata - ' + str(metadata))
if metadata and 'config' in metadata:
- return ImageSettings(config=metadata['config'])
+ return ImageSettings(**metadata['config'])
disk_file = None
if metadata:
from neutronclient.common.exceptions import NotFound
from neutronclient.neutron.client import Client
+from snaps.domain.vm_inst import FloatingIp
from snaps.openstack.utils import keystone_utils
__author__ = 'spisarski'
def neutron_client(os_creds):
"""
- Instantiates and returns a client for communications with OpenStack's Neutron server
+ Instantiates and returns a client for communications with OpenStack's
+ Neutron server
:param os_creds: the credentials for connecting to the OpenStack remote API
:return: the client object
"""
- return Client(api_version=os_creds.network_api_version, session=keystone_utils.keystone_session(os_creds))
+ return Client(api_version=os_creds.network_api_version,
+ session=keystone_utils.keystone_session(os_creds))
def create_network(neutron, os_creds, network_settings):
Creates a network for OpenStack
:param neutron: the client
:param os_creds: the OpenStack credentials
- :param network_settings: A dictionary containing the network configuration and is responsible for creating the
- network request JSON body
+ :param network_settings: A dictionary containing the network configuration
+ and is responsible for creating the network
+ request JSON body
:return: the network object
"""
if neutron and network_settings:
def get_network(neutron, network_name, project_id=None):
"""
- Returns an object (dictionary) of the first network found with a given name and project_id (if included)
+ Returns an object (dictionary) of the first network found with a given name
+ and project_id (if included)
:param neutron: the client
:param network_name: the name of the network to retrieve
:param project_id: the id of the network's project
Creates a network subnet for OpenStack
:param neutron: the client
:param network: the network object
- :param subnet_settings: A dictionary containing the subnet configuration and is responsible for creating the subnet
- request JSON body
+ :param subnet_settings: A dictionary containing the subnet configuration
+ and is responsible for creating the subnet request
+ JSON body
:param os_creds: the OpenStack credentials
:return: the subnet object
"""
if neutron and network and subnet_settings:
- json_body = {'subnets': [subnet_settings.dict_for_neutron(os_creds, network=network)]}
+ json_body = {'subnets': [subnet_settings.dict_for_neutron(
+ os_creds, network=network)]}
logger.info('Creating subnet with name ' + subnet_settings.name)
subnets = neutron.create_subnet(body=json_body)
return {'subnet': subnets['subnets'][0]}
Creates a router for OpenStack
:param neutron: the client
:param os_creds: the OpenStack credentials
- :param router_settings: A dictionary containing the router configuration and is responsible for creating the subnet
- request JSON body
+ :param router_settings: A dictionary containing the router configuration
+ and is responsible for creating the subnet request
+ JSON body
:return: the router object
"""
if neutron:
def add_interface_router(neutron, router, subnet=None, port=None):
"""
- Adds an interface router for OpenStack for either a subnet or port. Exception will be raised if requesting for both.
+ Adds an interface router for OpenStack for either a subnet or port.
+ Exception will be raised if requesting for both.
:param neutron: the client
:param router: the router object
:param subnet: the subnet object
:return: the interface router object
"""
if subnet and port:
- raise Exception('Cannot add interface to the router. Both subnet and port were sent in. Either or please.')
+ raise Exception('Cannot add interface to the router. Both subnet and '
+ 'port were sent in. Either or please.')
if neutron and router and (router or subnet):
- logger.info('Adding interface to router with name ' + router['router']['name'])
- return neutron.add_interface_router(router=router['router']['id'], body=__create_port_json_body(subnet, port))
+ logger.info('Adding interface to router with name ' +
+ router['router']['name'])
+ return neutron.add_interface_router(
+ router=router['router']['id'],
+ body=__create_port_json_body(subnet, port))
else:
- raise Exception("Unable to create interface router as neutron client, router or subnet were not created")
+ raise Exception('Unable to create interface router as neutron client,'
+ ' router or subnet were not created')
def remove_interface_router(neutron, router, subnet=None, port=None):
"""
if router:
try:
- logger.info('Removing router interface from router named ' + router['router']['name'])
- neutron.remove_interface_router(router=router['router']['id'], body=__create_port_json_body(subnet, port))
+ logger.info('Removing router interface from router named ' +
+ router['router']['name'])
+ neutron.remove_interface_router(
+ router=router['router']['id'],
+ body=__create_port_json_body(subnet, port))
except NotFound as e:
- logger.warning('Could not remove router interface. NotFound - ' + str(e))
+ logger.warning('Could not remove router interface. NotFound - %s',
+ e)
pass
else:
logger.warning('Could not remove router interface, No router object')
def __create_port_json_body(subnet=None, port=None):
"""
- Returns the dictionary required for creating and deleting router interfaces. Will only work on a subnet or port
- object. Will throw and exception if parameters contain both or neither
+ Returns the dictionary required for creating and deleting router
+ interfaces. Will only work on a subnet or port object. Will throw and
+ exception if parameters contain both or neither
:param subnet: the subnet object
:param port: the port object
:return: the dict
:return: the port object
"""
json_body = port_settings.dict_for_neutron(neutron, os_creds)
- logger.info('Creating port for network with name - ' + port_settings.network_name)
+ logger.info('Creating port for network with name - %s',
+ port_settings.network_name)
return neutron.create_port(body=json_body)
:param sec_grp_settings: the security group settings
:return: the security group object
"""
- logger.info('Creating security group with name - ' + sec_grp_settings.name)
- return neutron.create_security_group(sec_grp_settings.dict_for_neutron(keystone))
+ logger.info('Creating security group with name - %s',
+ sec_grp_settings.name)
+ return neutron.create_security_group(
+ sec_grp_settings.dict_for_neutron(keystone))
def delete_security_group(neutron, sec_grp):
:param neutron: the client
:param sec_grp: the security group object to delete
"""
- logger.info('Deleting security group with name - ' + sec_grp['security_group']['name'])
+ logger.info('Deleting security group with name - %s',
+ sec_grp['security_group']['name'])
return neutron.delete_security_group(sec_grp['security_group']['id'])
:param sec_grp_rule_settings: the security group rule settings
:return: the security group object
"""
- logger.info('Creating security group to security group - ' + sec_grp_rule_settings.sec_grp_name)
- return neutron.create_security_group_rule(sec_grp_rule_settings.dict_for_neutron(neutron))
+ logger.info('Creating security group to security group - %s',
+ sec_grp_rule_settings.sec_grp_name)
+ return neutron.create_security_group_rule(
+ sec_grp_rule_settings.dict_for_neutron(neutron))
def delete_security_group_rule(neutron, sec_grp_rule):
:param neutron: the client
:param sec_grp_rule: the security group rule object to delete
"""
- logger.info('Deleting security group rule with ID - ' + sec_grp_rule['security_group_rule']['id'])
- neutron.delete_security_group_rule(sec_grp_rule['security_group_rule']['id'])
+ logger.info('Deleting security group rule with ID - %s',
+ sec_grp_rule['security_group_rule']['id'])
+ neutron.delete_security_group_rule(
+ sec_grp_rule['security_group_rule']['id'])
def get_rules_by_security_group(neutron, sec_grp):
:param neutron: the client
:param sec_grp: the security group object
"""
- logger.info('Retrieving security group rules associate with the security group - ' +
- sec_grp['security_group']['name'])
+ logger.info('Retrieving security group rules associate with the '
+ 'security group - %s', sec_grp['security_group']['name'])
out = list()
- rules = neutron.list_security_group_rules(**{'security_group_id': sec_grp['security_group']['id']})
+ rules = neutron.list_security_group_rules(
+ **{'security_group_id': sec_grp['security_group']['id']})
for rule in rules['security_group_rules']:
if rule['security_group_id'] == sec_grp['security_group']['id']:
out.append({'security_group_rule': rule})
:param sec_grp: the security group object
:param rule_id: the rule's ID
"""
- rules = neutron.list_security_group_rules(**{'security_group_id': sec_grp['security_group']['id']})
+ rules = neutron.list_security_group_rules(
+ **{'security_group_id': sec_grp['security_group']['id']})
for rule in rules['security_group_rules']:
if rule['id'] == rule_id:
return {'security_group_rule': rule}
def get_external_networks(neutron):
"""
- Returns a list of external OpenStack network object/dict for all external networks
+ Returns a list of external OpenStack network object/dict for all external
+ networks
:param neutron: the client
:return: a list of external networks (empty list if none configured)
"""
out = list()
- for network in neutron.list_networks(**{'router:external': True})['networks']:
+ for network in neutron.list_networks(
+ **{'router:external': True})['networks']:
out.append({'network': network})
return out
+
+
+def get_floating_ips(neutron):
+ """
+ Returns all of the floating IPs
+ :param neutron: the Neutron client
+ :return: a list of SNAPS FloatingIp objects
+ """
+ out = list()
+ fips = neutron.list_floatingips()
+ for fip in fips['floatingips']:
+ out.append(FloatingIp(inst_id=fip['id'],
+ ip=fip['floating_ip_address']))
+
+ return out
+
+
+def create_floating_ip(neutron, ext_net_name):
+ """
+ Returns the floating IP object that was created with this call
+ :param neutron: the Neutron client
+ :param ext_net_name: the name of the external network on which to apply the
+ floating IP address
+ :return: the SNAPS FloatingIp object
+ """
+ logger.info('Creating floating ip to external network - ' + ext_net_name)
+ ext_net = get_network(neutron, ext_net_name)
+ if ext_net:
+ fip = neutron.create_floatingip(
+ body={'floatingip':
+ {'floating_network_id': ext_net['network']['id']}})
+
+ return FloatingIp(inst_id=fip['floatingip']['id'],
+ ip=fip['floatingip']['floating_ip_address'])
+ else:
+ raise Exception('Cannot create floating IP, '
+ 'external network not found')
+
+
+def get_floating_ip(neutron, floating_ip):
+ """
+ Returns a floating IP object that should be identical to the floating_ip
+ parameter
+ :param neutron: the Neutron client
+ :param floating_ip: the SNAPS FloatingIp object
+ :return: hopefully the same floating IP object input
+ """
+ logger.debug('Attempting to retrieve existing floating ip with IP - %s',
+ floating_ip.ip)
+ os_fip = get_os_floating_ip(neutron, floating_ip)
+ if os_fip:
+ return FloatingIp(
+ inst_id=os_fip['id'], ip=os_fip['floating_ip_address'])
+
+
+def get_os_floating_ip(neutron, floating_ip):
+ """
+ Returns an OpenStack floating IP object
+ parameter
+ :param neutron: the Neutron client
+ :param floating_ip: the SNAPS FloatingIp object
+ :return: hopefully the same floating IP object input
+ """
+ logger.debug('Attempting to retrieve existing floating ip with IP - %s',
+ floating_ip.ip)
+ fips = neutron.list_floatingips(ip=floating_ip.id)
+
+ for fip in fips['floatingips']:
+ if fip['id'] == floating_ip.id:
+ return fip
+
+
+def delete_floating_ip(neutron, floating_ip):
+ """
+ Responsible for deleting a floating IP
+ :param neutron: the Neutron client
+ :param floating_ip: the SNAPS FloatingIp object
+ :return:
+ """
+ logger.debug('Attempting to delete existing floating ip with IP - %s',
+ floating_ip.ip)
+ return neutron.delete_floatingip(floating_ip.id)
# See the License for the specific language governing permissions and
# limitations under the License.
-from cryptography.hazmat.primitives import serialization
-from cryptography.hazmat.primitives.asymmetric import rsa
-from cryptography.hazmat.backends import default_backend
-
-import os
import logging
-from snaps.openstack.utils import keystone_utils
+import os
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
from novaclient.client import Client
from novaclient.exceptions import NotFound
+from snaps.domain.vm_inst import VmInst
+from snaps.openstack.utils import keystone_utils, glance_utils, neutron_utils
__author__ = 'spisarski'
def nova_client(os_creds):
"""
- Instantiates and returns a client for communications with OpenStack's Nova server
+ Instantiates and returns a client for communications with OpenStack's Nova
+ server
:param os_creds: The connection credentials to the OpenStack API
:return: the client object
"""
logger.debug('Retrieving Nova Client')
- return Client(os_creds.compute_api_version, session=keystone_utils.keystone_session(os_creds))
+ return Client(os_creds.compute_api_version,
+ session=keystone_utils.keystone_session(os_creds))
+
+
+def create_server(nova, neutron, glance, instance_settings, image_settings,
+ keypair_settings=None):
+ """
+ Creates a VM instance
+ :param nova: the nova client (required)
+ :param neutron: the neutron client for retrieving ports (required)
+ :param glance: the glance client (required)
+ :param instance_settings: the VM instance settings object (required)
+ :param image_settings: the VM's image settings object (required)
+ :param keypair_settings: the VM's keypair settings object (optional)
+ :return: a snaps.domain.VmInst object
+ """
+
+ ports = list()
+
+ for port_setting in instance_settings.port_settings:
+ ports.append(neutron_utils.get_port_by_name(
+ neutron, port_setting.name))
+ nics = []
+ for port in ports:
+ kv = dict()
+ kv['port-id'] = port['port']['id']
+ nics.append(kv)
+
+ logger.info('Creating VM with name - ' + instance_settings.name)
+ keypair_name = None
+ if keypair_settings:
+ keypair_name = keypair_settings.name
+
+ flavor = get_flavor_by_name(nova, instance_settings.flavor)
+ if not flavor:
+ raise Exception(
+ 'Flavor not found with name - %s',
+ instance_settings.flavor)
+
+ image = glance_utils.get_image(glance, image_settings.name)
+ if image:
+ args = {'name': instance_settings.name,
+ 'flavor': flavor,
+ 'image': image,
+ 'nics': nics,
+ 'key_name': keypair_name,
+ 'security_groups':
+ instance_settings.security_group_names,
+ 'userdata': instance_settings.userdata,
+ 'availability_zone':
+ instance_settings.availability_zone}
+ server = nova.servers.create(**args)
+ return VmInst(name=server.name, inst_id=server.id)
+ else:
+ raise Exception(
+ 'Cannot create instance, image cannot be located with name %s',
+ image_settings.name)
def get_servers_by_name(nova, name):
:param name: the server name
:return: the list of servers
"""
- return nova.servers.list(search_opts={'name': name})
+ out = list()
+ servers = nova.servers.list(search_opts={'name': name})
+ for server in servers:
+ out.append(VmInst(name=server.name, inst_id=server.id))
+ return out
+
+
+def get_latest_server_os_object(nova, server):
+ """
+ Returns a server with a given id
+ :param nova: the Nova client
+ :param server: the domain VmInst object
+ :return: the list of servers or None if not found
+ """
+ return nova.servers.get(server.id)
def get_latest_server_object(nova, server):
:param server: the old server object
:return: the list of servers or None if not found
"""
- return nova.servers.get(server)
+ server = get_latest_server_os_object(nova, server)
+ return VmInst(name=server.name, inst_id=server.id)
def create_keys(key_size=2048):
:param key_size: the number of bytes for the key size
:return: the cryptography keys
"""
- return rsa.generate_private_key(backend=default_backend(), public_exponent=65537,
+ return rsa.generate_private_key(backend=default_backend(),
+ public_exponent=65537,
key_size=key_size)
:param keys: the keys generated by create_keys() from cryptography
:return: the OpenSSH public key
"""
- return keys.public_key().public_bytes(serialization.Encoding.OpenSSH, serialization.PublicFormat.OpenSSH)
+ return keys.public_key().public_bytes(serialization.Encoding.OpenSSH,
+ serialization.PublicFormat.OpenSSH)
def save_keys_to_files(keys=None, pub_file_path=None, priv_file_path=None):
os.mkdir(pub_dir)
public_handle = open(pub_file_path, 'wb')
public_bytes = keys.public_key().public_bytes(
- serialization.Encoding.OpenSSH, serialization.PublicFormat.OpenSSH)
+ serialization.Encoding.OpenSSH,
+ serialization.PublicFormat.OpenSSH)
public_handle.write(public_bytes)
public_handle.close()
os.chmod(pub_file_path, 0o400)
if not os.path.isdir(priv_dir):
os.mkdir(priv_dir)
private_handle = open(priv_file_path, 'wb')
- private_handle.write(keys.private_bytes(encoding=serialization.Encoding.PEM,
- format=serialization.PrivateFormat.TraditionalOpenSSL,
- encryption_algorithm=serialization.NoEncryption()))
+ private_handle.write(
+ keys.private_bytes(
+ encoding=serialization.Encoding.PEM,
+ format=serialization.PrivateFormat.TraditionalOpenSSL,
+ encryption_algorithm=serialization.NoEncryption()))
private_handle.close()
os.chmod(priv_file_path, 0o400)
logger.info("Saved private key to - " + priv_file_path)
nova.keypairs.delete(key)
-def get_floating_ip_pools(nova):
- """
- Returns all of the available floating IP pools
- :param nova: the Nova client
- :return: a list of pools
- """
- return nova.floating_ip_pools.list()
-
-
-def get_floating_ips(nova):
- """
- Returns all of the floating IPs
- :param nova: the Nova client
- :return: a list of floating IPs
- """
- return nova.floating_ips.list()
-
-
-def create_floating_ip(nova, ext_net_name):
- """
- Returns the floating IP object that was created with this call
- :param nova: the Nova client
- :param ext_net_name: the name of the external network on which to apply the floating IP address
- :return: the floating IP object
- """
- logger.info('Creating floating ip to external network - ' + ext_net_name)
- return nova.floating_ips.create(ext_net_name)
-
-
-def get_floating_ip(nova, floating_ip):
- """
- Returns a floating IP object that should be identical to the floating_ip parameter
- :param nova: the Nova client
- :param floating_ip: the floating IP object to lookup
- :return: hopefully the same floating IP object input
- """
- logger.debug('Attempting to retrieve existing floating ip with IP - ' + floating_ip.ip)
- return nova.floating_ips.get(floating_ip)
-
-
-def delete_floating_ip(nova, floating_ip):
- """
- Responsible for deleting a floating IP
- :param nova: the Nova client
- :param floating_ip: the floating IP object to delete
- :return:
- """
- logger.debug('Attempting to delete existing floating ip with IP - ' + floating_ip.ip)
- return nova.floating_ips.delete(floating_ip)
-
-
def get_nova_availability_zones(nova):
"""
Returns the names of all nova active compute servers
"""
Deletes a VM instance
:param nova: the nova client
- :param vm_inst: the OpenStack instance object to delete
+ :param vm_inst: the snaps.domain.VmInst object
"""
- nova.servers.delete(vm_inst)
+ nova.servers.delete(vm_inst.id)
def get_flavor_by_name(nova, name):
:param flavor_settings: the flavor settings
:return: the Flavor
"""
- return nova.flavors.create(name=flavor_settings.name, flavorid=flavor_settings.flavor_id, ram=flavor_settings.ram,
- vcpus=flavor_settings.vcpus, disk=flavor_settings.disk,
- ephemeral=flavor_settings.ephemeral, swap=flavor_settings.swap,
- rxtx_factor=flavor_settings.rxtx_factor, is_public=flavor_settings.is_public)
+ return nova.flavors.create(name=flavor_settings.name,
+ flavorid=flavor_settings.flavor_id,
+ ram=flavor_settings.ram,
+ vcpus=flavor_settings.vcpus,
+ disk=flavor_settings.disk,
+ ephemeral=flavor_settings.ephemeral,
+ swap=flavor_settings.swap,
+ rxtx_factor=flavor_settings.rxtx_factor,
+ is_public=flavor_settings.is_public)
def delete_flavor(nova, flavor):
:param vm: the OpenStack server object (VM) to alter
:param security_group: the OpenStack security group object to add
"""
- nova.servers.remove_security_group(str(vm.id), security_group['security_group']['name'])
+ nova.servers.remove_security_group(
+ str(vm.id), security_group['security_group']['name'])
+
+
+def add_floating_ip_to_server(nova, vm, floating_ip, ip_addr):
+ """
+ Adds a floating IP to a server instance
+ :param nova: the nova client
+ :param vm: VmInst domain object
+ :param floating_ip: FloatingIp domain object
+ :param ip_addr: the IP to which to bind the floating IP to
+ """
+ vm = get_latest_server_os_object(nova, vm)
+ vm.add_floating_ip(floating_ip.ip, ip_addr)
# limitations under the License.
import uuid
-from snaps.openstack.utils import keystone_utils
-from snaps.openstack.create_security_group import SecurityGroupSettings, SecurityGroupRuleSettings, Direction
-from snaps.openstack.tests import openstack_tests
-from snaps.openstack.utils import neutron_utils
-from snaps.openstack.create_network import NetworkSettings, SubnetSettings, PortSettings
from snaps.openstack import create_router
-from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
+from snaps.openstack.create_network import NetworkSettings, SubnetSettings, \
+ PortSettings
+from snaps.openstack.create_security_group import SecurityGroupSettings, \
+ SecurityGroupRuleSettings, Direction
+from snaps.openstack.tests import openstack_tests
from snaps.openstack.tests import validation_utils
+from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
+from snaps.openstack.utils import keystone_utils
+from snaps.openstack.utils import neutron_utils
__author__ = 'spisarski'
with self.assertRaises(Exception):
neutron = neutron_utils.neutron_client(
- OSCreds(username='user', password='pass', auth_url='url', project_name='project'))
+ OSCreds(username='user', password='pass', auth_url='url',
+ project_name='project'))
neutron.list_networks()
def test_retrieve_ext_network_name(self):
"""
- Tests the neutron_utils.get_external_network_names to ensure the configured self.ext_net_name is contained
- within the returned list
+ Tests the neutron_utils.get_external_network_names to ensure the
+ configured self.ext_net_name is contained within the returned list
:return:
"""
neutron = neutron_utils.neutron_client(self.os_creds)
self.port_name = str(guid) + '-port'
self.neutron = neutron_utils.neutron_client(self.os_creds)
self.network = None
- self.net_config = openstack_tests.get_pub_net_config(net_name=guid + '-pub-net')
+ self.net_config = openstack_tests.get_pub_net_config(
+ net_name=guid + '-pub-net')
def tearDown(self):
"""
"""
if self.network:
neutron_utils.delete_network(self.neutron, self.network)
- validate_network(self.neutron, self.network['network']['name'], False)
+ validate_network(self.neutron, self.network['network']['name'],
+ False)
def test_create_network(self):
"""
Tests the neutron_utils.create_neutron_net() function
"""
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds, self.net_config.network_settings)
+ self.assertEqual(self.net_config.network_settings.name,
+ self.network['network']['name'])
+ self.assertTrue(validate_network(self.neutron,
+ self.net_config.network_settings.name,
+ True))
def test_create_network_empty_name(self):
"""
- Tests the neutron_utils.create_neutron_net() function with an empty network name
+ Tests the neutron_utils.create_neutron_net() function with an empty
+ network name
"""
with self.assertRaises(Exception):
- self.network = neutron_utils.create_network(self.neutron, self.os_creds,
- network_settings=NetworkSettings(name=''))
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds,
+ network_settings=NetworkSettings(name=''))
def test_create_network_null_name(self):
"""
- Tests the neutron_utils.create_neutron_net() function when the network name is None
+ Tests the neutron_utils.create_neutron_net() function when the network
+ name is None
"""
with self.assertRaises(Exception):
- self.network = neutron_utils.create_network(self.neutron, self.os_creds,
- network_settings=NetworkSettings())
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds,
+ network_settings=NetworkSettings())
class NeutronUtilsSubnetTests(OSComponentTestCase):
self.network = None
self.subnet = None
self.net_config = openstack_tests.get_pub_net_config(
- net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet', external_net=self.ext_net_name)
+ net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
+ external_net=self.ext_net_name)
def tearDown(self):
"""
if self.subnet:
neutron_utils.delete_subnet(self.neutron, self.subnet)
validate_subnet(self.neutron, self.subnet.get('name'),
- self.net_config.network_settings.subnet_settings[0].cidr, False)
+ self.net_config.network_settings.subnet_settings[
+ 0].cidr, False)
if self.network:
neutron_utils.delete_network(self.neutron, self.network)
- validate_network(self.neutron, self.network['network']['name'], False)
+ validate_network(self.neutron, self.network['network']['name'],
+ False)
def test_create_subnet(self):
"""
Tests the neutron_utils.create_neutron_net() function
"""
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
-
- self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0],
- self.os_creds, network=self.network)
- validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name,
- self.net_config.network_settings.subnet_settings[0].cidr, True)
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds, self.net_config.network_settings)
+ self.assertEqual(self.net_config.network_settings.name,
+ self.network['network']['name'])
+ self.assertTrue(validate_network(self.neutron,
+ self.net_config.network_settings.name,
+ True))
+
+ subnet_setting = self.net_config.network_settings.subnet_settings[0]
+ self.subnet = neutron_utils.create_subnet(
+ self.neutron, subnet_setting,
+ self.os_creds, network=self.network)
+ validate_subnet(
+ self.neutron,
+ subnet_setting.name,
+ subnet_setting.cidr, True)
def test_create_subnet_null_name(self):
"""
- Tests the neutron_utils.create_neutron_subnet() function for an Exception when the subnet name is None
+ Tests the neutron_utils.create_neutron_subnet() function for an
+ Exception when the subnet name is None
"""
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds, self.net_config.network_settings)
+ self.assertEqual(self.net_config.network_settings.name,
+ self.network['network']['name'])
+ self.assertTrue(validate_network(self.neutron,
+ self.net_config.network_settings.name,
+ True))
with self.assertRaises(Exception):
SubnetSettings(cidr=self.net_config.subnet_cidr)
def test_create_subnet_empty_name(self):
"""
- Tests the neutron_utils.create_neutron_net() function with an empty name
+ Tests the neutron_utils.create_neutron_net() function with an empty
+ name
"""
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds, self.net_config.network_settings)
+ self.assertEqual(self.net_config.network_settings.name,
+ self.network['network']['name'])
+ self.assertTrue(validate_network(self.neutron,
+ self.net_config.network_settings.name,
+ True))
- neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0],
- self.os_creds, network=self.network)
- validate_subnet(self.neutron, '', self.net_config.network_settings.subnet_settings[0].cidr, True)
+ subnet_setting = self.net_config.network_settings.subnet_settings[0]
+ neutron_utils.create_subnet(
+ self.neutron, subnet_setting,
+ self.os_creds, network=self.network)
+ validate_subnet(self.neutron, '',
+ subnet_setting.cidr, True)
def test_create_subnet_null_cidr(self):
"""
- Tests the neutron_utils.create_neutron_subnet() function for an Exception when the subnet CIDR value is None
+ Tests the neutron_utils.create_neutron_subnet() function for an
+ Exception when the subnet CIDR value is None
"""
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds, self.net_config.network_settings)
+ self.assertEqual(self.net_config.network_settings.name,
+ self.network['network']['name'])
+ self.assertTrue(validate_network(self.neutron,
+ self.net_config.network_settings.name,
+ True))
with self.assertRaises(Exception):
- sub_sets = SubnetSettings(cidr=None, name=self.net_config.subnet_name)
- neutron_utils.create_subnet(self.neutron, sub_sets, self.os_creds, network=self.network)
+ sub_sets = SubnetSettings(cidr=None,
+ name=self.net_config.subnet_name)
+ neutron_utils.create_subnet(self.neutron, sub_sets, self.os_creds,
+ network=self.network)
def test_create_subnet_empty_cidr(self):
"""
- Tests the neutron_utils.create_neutron_subnet() function for an Exception when the subnet CIDR value is empty
+ Tests the neutron_utils.create_neutron_subnet() function for an
+ Exception when the subnet CIDR value is empty
"""
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds, self.net_config.network_settings)
+ self.assertEqual(self.net_config.network_settings.name,
+ self.network['network']['name'])
+ self.assertTrue(validate_network(self.neutron,
+ self.net_config.network_settings.name,
+ True))
with self.assertRaises(Exception):
- sub_sets = SubnetSettings(cidr='', name=self.net_config.subnet_name)
- neutron_utils.create_subnet(self.neutron, sub_sets, self.os_creds, network=self.network)
+ sub_sets = SubnetSettings(cidr='',
+ name=self.net_config.subnet_name)
+ neutron_utils.create_subnet(self.neutron, sub_sets, self.os_creds,
+ network=self.network)
class NeutronUtilsRouterTests(OSComponentTestCase):
Cleans the remote OpenStack objects
"""
if self.interface_router:
- neutron_utils.remove_interface_router(self.neutron, self.router, self.subnet)
+ neutron_utils.remove_interface_router(self.neutron, self.router,
+ self.subnet)
if self.router:
neutron_utils.delete_router(self.neutron, self.router)
if self.subnet:
neutron_utils.delete_subnet(self.neutron, self.subnet)
validate_subnet(self.neutron, self.subnet.get('name'),
- self.net_config.network_settings.subnet_settings[0].cidr, False)
+ self.net_config.network_settings.subnet_settings[
+ 0].cidr, False)
if self.network:
neutron_utils.delete_network(self.neutron, self.network)
- validate_network(self.neutron, self.network['network']['name'], False)
+ validate_network(self.neutron, self.network['network']['name'],
+ False)
def test_create_router_simple(self):
"""
- Tests the neutron_utils.create_neutron_net() function when an external gateway is requested
+ Tests the neutron_utils.create_neutron_net() function when an external
+ gateway is requested
"""
- self.router = neutron_utils.create_router(self.neutron, self.os_creds, self.net_config.router_settings)
- validate_router(self.neutron, self.net_config.router_settings.name, True)
+ self.router = neutron_utils.create_router(
+ self.neutron, self.os_creds, self.net_config.router_settings)
+ validate_router(self.neutron, self.net_config.router_settings.name,
+ True)
def test_create_router_with_public_interface(self):
"""
- Tests the neutron_utils.create_neutron_net() function when an external gateway is requested
+ Tests the neutron_utils.create_neutron_net() function when an external
+ gateway is requested
"""
+ subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.net_config = openstack_tests.OSNetworkConfig(
self.net_config.network_settings.name,
- self.net_config.network_settings.subnet_settings[0].name,
- self.net_config.network_settings.subnet_settings[0].cidr, self.net_config.router_settings.name,
+ subnet_setting.name,
+ subnet_setting.cidr,
+ self.net_config.router_settings.name,
self.ext_net_name)
- self.router = neutron_utils.create_router(self.neutron, self.os_creds, self.net_config.router_settings)
- validate_router(self.neutron, self.net_config.router_settings.name, True)
+ self.router = neutron_utils.create_router(
+ self.neutron, self.os_creds, self.net_config.router_settings)
+ validate_router(self.neutron, self.net_config.router_settings.name,
+ True)
# TODO - Add validation that the router gatway has been set
def test_create_router_empty_name(self):
"""
with self.assertRaises(Exception):
this_router_settings = create_router.RouterSettings(name='')
- self.router = neutron_utils.create_router(self.neutron, self.os_creds, this_router_settings)
+ self.router = neutron_utils.create_router(self.neutron,
+ self.os_creds,
+ this_router_settings)
def test_create_router_null_name(self):
"""
- Tests the neutron_utils.create_neutron_subnet() function when the subnet CIDR value is None
+ Tests the neutron_utils.create_neutron_subnet() function when the
+ subnet CIDR value is None
"""
with self.assertRaises(Exception):
this_router_settings = create_router.RouterSettings()
- self.router = neutron_utils.create_router(self.neutron, self.os_creds, this_router_settings)
+ self.router = neutron_utils.create_router(self.neutron,
+ self.os_creds,
+ this_router_settings)
validate_router(self.neutron, None, True)
def test_add_interface_router(self):
"""
Tests the neutron_utils.add_interface_router() function
"""
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
-
- self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0],
- self.os_creds, self.network)
- validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name,
- self.net_config.network_settings.subnet_settings[0].cidr, True)
-
- self.router = neutron_utils.create_router(self.neutron, self.os_creds, self.net_config.router_settings)
- validate_router(self.neutron, self.net_config.router_settings.name, True)
-
- self.interface_router = neutron_utils.add_interface_router(self.neutron, self.router, self.subnet)
- validate_interface_router(self.interface_router, self.router, self.subnet)
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds, self.net_config.network_settings)
+ self.assertEqual(self.net_config.network_settings.name,
+ self.network['network']['name'])
+ self.assertTrue(validate_network(self.neutron,
+ self.net_config.network_settings.name,
+ True))
+
+ subnet_setting = self.net_config.network_settings.subnet_settings[0]
+ self.subnet = neutron_utils.create_subnet(
+ self.neutron, subnet_setting,
+ self.os_creds, self.network)
+ validate_subnet(
+ self.neutron,
+ subnet_setting.name,
+ subnet_setting.cidr, True)
+
+ self.router = neutron_utils.create_router(
+ self.neutron, self.os_creds, self.net_config.router_settings)
+ validate_router(self.neutron, self.net_config.router_settings.name,
+ True)
+
+ self.interface_router = neutron_utils.add_interface_router(
+ self.neutron, self.router, self.subnet)
+ validate_interface_router(self.interface_router, self.router,
+ self.subnet)
def test_add_interface_router_null_router(self):
"""
- Tests the neutron_utils.add_interface_router() function for an Exception when the router value is None
- """
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
-
- self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0],
- self.os_creds, self.network)
- validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name,
- self.net_config.network_settings.subnet_settings[0].cidr, True)
+ Tests the neutron_utils.add_interface_router() function for an
+ Exception when the router value is None
+ """
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds, self.net_config.network_settings)
+ self.assertEqual(self.net_config.network_settings.name,
+ self.network['network']['name'])
+ self.assertTrue(validate_network(self.neutron,
+ self.net_config.network_settings.name,
+ True))
+
+ subnet_setting = self.net_config.network_settings.subnet_settings[0]
+ self.subnet = neutron_utils.create_subnet(
+ self.neutron, subnet_setting,
+ self.os_creds, self.network)
+ validate_subnet(
+ self.neutron,
+ subnet_setting.name,
+ subnet_setting.cidr, True)
with self.assertRaises(Exception):
- self.interface_router = neutron_utils.add_interface_router(self.neutron, self.router, self.subnet)
+ self.interface_router = neutron_utils.add_interface_router(
+ self.neutron, self.router, self.subnet)
def test_add_interface_router_null_subnet(self):
"""
- Tests the neutron_utils.add_interface_router() function for an Exception when the subnet value is None
+ Tests the neutron_utils.add_interface_router() function for an
+ Exception when the subnet value is None
"""
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds, self.net_config.network_settings)
+ self.assertEqual(self.net_config.network_settings.name,
+ self.network['network']['name'])
+ self.assertTrue(validate_network(self.neutron,
+ self.net_config.network_settings.name,
+ True))
- self.router = neutron_utils.create_router(self.neutron, self.os_creds, self.net_config.router_settings)
- validate_router(self.neutron, self.net_config.router_settings.name, True)
+ self.router = neutron_utils.create_router(
+ self.neutron, self.os_creds, self.net_config.router_settings)
+ validate_router(self.neutron, self.net_config.router_settings.name,
+ True)
with self.assertRaises(Exception):
- self.interface_router = neutron_utils.add_interface_router(self.neutron, self.router, self.subnet)
+ self.interface_router = neutron_utils.add_interface_router(
+ self.neutron, self.router, self.subnet)
def test_create_port(self):
"""
Tests the neutron_utils.create_port() function
"""
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds, self.net_config.network_settings)
+ self.assertEqual(self.net_config.network_settings.name,
+ self.network['network']['name'])
+ self.assertTrue(validate_network(self.neutron,
+ self.net_config.network_settings.name,
+ True))
- self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0],
- self.os_creds, self.network)
- validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name,
- self.net_config.network_settings.subnet_settings[0].cidr, True)
+ subnet_setting = self.net_config.network_settings.subnet_settings[0]
+ self.subnet = neutron_utils.create_subnet(
+ self.neutron, subnet_setting, self.os_creds, self.network)
+ validate_subnet(self.neutron, subnet_setting.name,
+ subnet_setting.cidr, True)
self.port = neutron_utils.create_port(
self.neutron, self.os_creds, PortSettings(
name=self.port_name,
- ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings[0].name, 'ip': ip_1}],
+ ip_addrs=[{
+ 'subnet_name': subnet_setting.name,
+ 'ip': ip_1}],
network_name=self.net_config.network_settings.name))
validate_port(self.neutron, self.port, self.port_name)
"""
Tests the neutron_utils.create_port() function
"""
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds, self.net_config.network_settings)
+ self.assertEqual(self.net_config.network_settings.name,
+ self.network['network']['name'])
+ self.assertTrue(validate_network(self.neutron,
+ self.net_config.network_settings.name,
+ True))
- self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0],
- self.os_creds, self.network)
- validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name,
- self.net_config.network_settings.subnet_settings[0].cidr, True)
+ subnet_setting = self.net_config.network_settings.subnet_settings[0]
+ self.subnet = neutron_utils.create_subnet(
+ self.neutron, subnet_setting, self.os_creds, self.network)
+ validate_subnet(self.neutron, subnet_setting.name, subnet_setting.cidr,
+ True)
self.port = neutron_utils.create_port(
self.neutron, self.os_creds, PortSettings(
- name=self.port_name, network_name=self.net_config.network_settings.name,
- ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings[0].name, 'ip': ip_1}]))
+ name=self.port_name,
+ network_name=self.net_config.network_settings.name,
+ ip_addrs=[{
+ 'subnet_name': subnet_setting.name,
+ 'ip': ip_1}]))
validate_port(self.neutron, self.port, self.port_name)
def test_create_port_null_name(self):
"""
- Tests the neutron_utils.create_port() function for an Exception when the port name value is None
- """
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
-
- self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0],
- self.os_creds, self.network)
- validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name,
- self.net_config.network_settings.subnet_settings[0].cidr, True)
+ Tests the neutron_utils.create_port() function for an Exception when
+ the port name value is None
+ """
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds, self.net_config.network_settings)
+ self.assertEqual(self.net_config.network_settings.name,
+ self.network['network']['name'])
+ self.assertTrue(validate_network(self.neutron,
+ self.net_config.network_settings.name,
+ True))
+
+ subnet_setting = self.net_config.network_settings.subnet_settings[0]
+ self.subnet = neutron_utils.create_subnet(
+ self.neutron, subnet_setting,
+ self.os_creds, self.network)
+ validate_subnet(
+ self.neutron,
+ subnet_setting.name,
+ subnet_setting.cidr, True)
with self.assertRaises(Exception):
- self.port = neutron_utils.create_port(self.neutron, self.os_creds, PortSettings(
- network_name=self.net_config.network_settings.name,
- ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings.name, 'ip': ip_1}]))
+ self.port = neutron_utils.create_port(
+ self.neutron, self.os_creds,
+ PortSettings(
+ network_name=self.net_config.network_settings.name,
+ ip_addrs=[{
+ 'subnet_name': subnet_setting.name,
+ 'ip': ip_1}]))
def test_create_port_null_network_object(self):
"""
- Tests the neutron_utils.create_port() function for an Exception when the network object is None
+ Tests the neutron_utils.create_port() function for an Exception when
+ the network object is None
"""
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
-
- self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0],
- self.os_creds, self.network)
- validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name,
- self.net_config.network_settings.subnet_settings[0].cidr, True)
-
with self.assertRaises(Exception):
- self.port = neutron_utils.create_port(self.neutron, self.os_creds, PortSettings(
- self.neutron, self.port_name, self.net_config.network_settings.name,
- ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings.name, 'ip': ip_1}]))
+ self.port = neutron_utils.create_port(
+ self.neutron, self.os_creds,
+ PortSettings(
+ name=self.port_name,
+ network_name=self.net_config.network_settings.name,
+ ip_addrs=[{
+ 'subnet_name':
+ self.net_config.network_settings.subnet_settings[
+ 0].name,
+ 'ip': ip_1}]))
def test_create_port_null_ip(self):
"""
- Tests the neutron_utils.create_port() function for an Exception when the IP value is None
- """
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
-
- self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0],
- self.os_creds, self.network)
- validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name,
- self.net_config.network_settings.subnet_settings[0].cidr, True)
+ Tests the neutron_utils.create_port() function for an Exception when
+ the IP value is None
+ """
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds, self.net_config.network_settings)
+ self.assertEqual(self.net_config.network_settings.name,
+ self.network['network']['name'])
+ self.assertTrue(validate_network(self.neutron,
+ self.net_config.network_settings.name,
+ True))
+
+ subnet_setting = self.net_config.network_settings.subnet_settings[0]
+ self.subnet = neutron_utils.create_subnet(
+ self.neutron, subnet_setting,
+ self.os_creds, self.network)
+ validate_subnet(
+ self.neutron,
+ subnet_setting.name,
+ subnet_setting.cidr, True)
with self.assertRaises(Exception):
- self.port = neutron_utils.create_port(self.neutron, self.os_creds, PortSettings(
- name=self.port_name, network_name=self.net_config.network_settings.name,
- ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings.name, 'ip': None}]))
+ self.port = neutron_utils.create_port(
+ self.neutron, self.os_creds,
+ PortSettings(
+ name=self.port_name,
+ network_name=self.net_config.network_settings.name,
+ ip_addrs=[{
+ 'subnet_name': subnet_setting.name,
+ 'ip': None}]))
def test_create_port_invalid_ip(self):
"""
- Tests the neutron_utils.create_port() function for an Exception when the IP value is None
- """
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
-
- self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0],
- self.os_creds, self.network)
- validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name,
- self.net_config.network_settings.subnet_settings[0].cidr, True)
+ Tests the neutron_utils.create_port() function for an Exception when
+ the IP value is None
+ """
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds, self.net_config.network_settings)
+ self.assertEqual(self.net_config.network_settings.name,
+ self.network['network']['name'])
+ self.assertTrue(validate_network(self.neutron,
+ self.net_config.network_settings.name,
+ True))
+
+ subnet_setting = self.net_config.network_settings.subnet_settings[0]
+ self.subnet = neutron_utils.create_subnet(
+ self.neutron,
+ subnet_setting,
+ self.os_creds, self.network)
+ validate_subnet(self.neutron,
+ subnet_setting.name,
+ subnet_setting.cidr, True)
with self.assertRaises(Exception):
- self.port = neutron_utils.create_port(self.neutron, self.os_creds, PortSettings(
- name=self.port_name, network_name=self.net_config.network_settings.name,
- ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings.name, 'ip': 'foo'}]))
+ self.port = neutron_utils.create_port(
+ self.neutron, self.os_creds,
+ PortSettings(
+ name=self.port_name,
+ network_name=self.net_config.network_settings.name,
+ ip_addrs=[{
+ 'subnet_name': subnet_setting.name,
+ 'ip': 'foo'}]))
def test_create_port_invalid_ip_to_subnet(self):
"""
- Tests the neutron_utils.create_port() function for an Exception when the IP value is None
- """
- self.network = neutron_utils.create_network(self.neutron, self.os_creds, self.net_config.network_settings)
- self.assertEqual(self.net_config.network_settings.name, self.network['network']['name'])
- self.assertTrue(validate_network(self.neutron, self.net_config.network_settings.name, True))
-
- self.subnet = neutron_utils.create_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0],
- self.os_creds, self.network)
- validate_subnet(self.neutron, self.net_config.network_settings.subnet_settings[0].name,
- self.net_config.network_settings.subnet_settings[0].cidr, True)
+ Tests the neutron_utils.create_port() function for an Exception when
+ the IP value is None
+ """
+ self.network = neutron_utils.create_network(
+ self.neutron, self.os_creds, self.net_config.network_settings)
+ self.assertEqual(self.net_config.network_settings.name,
+ self.network['network']['name'])
+ self.assertTrue(validate_network(self.neutron,
+ self.net_config.network_settings.name,
+ True))
+
+ subnet_setting = self.net_config.network_settings.subnet_settings[0]
+ self.subnet = neutron_utils.create_subnet(
+ self.neutron,
+ subnet_setting,
+ self.os_creds, self.network)
+ validate_subnet(self.neutron,
+ subnet_setting.name,
+ subnet_setting.cidr, True)
with self.assertRaises(Exception):
- self.port = neutron_utils.create_port(self.neutron, self.os_creds, PortSettings(
- name=self.port_name, network_name=self.net_config.network_settings.name,
- ip_addrs=[{'subnet_name': self.net_config.network_settings.subnet_settings.name,
- 'ip': '10.197.123.100'}]))
+ self.port = neutron_utils.create_port(
+ self.neutron, self.os_creds,
+ PortSettings(
+ name=self.port_name,
+ network_name=self.net_config.network_settings.name,
+ ip_addrs=[{
+ 'subnet_name': subnet_setting.name,
+ 'ip': '10.197.123.100'}]))
class NeutronUtilsSecurityGroupTests(OSComponentTestCase):
for security_group in self.security_groups:
try:
- neutron_utils.delete_security_group(self.neutron, security_group)
+ neutron_utils.delete_security_group(self.neutron,
+ security_group)
except:
pass
Tests the neutron_utils.create_security_group() function
"""
sec_grp_settings = SecurityGroupSettings(name=self.sec_grp_name)
- security_group = neutron_utils.create_security_group(self.neutron, self.keystone, sec_grp_settings)
+ security_group = neutron_utils.create_security_group(self.neutron,
+ self.keystone,
+ sec_grp_settings)
- self.assertTrue(sec_grp_settings.name, security_group['security_group']['name'])
+ self.assertTrue(sec_grp_settings.name,
+ security_group['security_group']['name'])
- sec_grp_get = neutron_utils.get_security_group(self.neutron, sec_grp_settings.name)
+ sec_grp_get = neutron_utils.get_security_group(self.neutron,
+ sec_grp_settings.name)
self.assertIsNotNone(sec_grp_get)
self.assertTrue(validation_utils.objects_equivalent(
security_group['security_group'], sec_grp_get['security_group']))
neutron_utils.delete_security_group(self.neutron, security_group)
- sec_grp_get = neutron_utils.get_security_group(self.neutron, sec_grp_settings.name)
+ sec_grp_get = neutron_utils.get_security_group(self.neutron,
+ sec_grp_settings.name)
self.assertIsNone(sec_grp_get)
def test_create_sec_grp_no_name(self):
"""
- Tests the SecurityGroupSettings constructor and neutron_utils.create_security_group() function to ensure
- that attempting to create a security group without a name will raise an exception
+ Tests the SecurityGroupSettings constructor and
+ neutron_utils.create_security_group() function to ensure that
+ attempting to create a security group without a name will raise an
+ exception
"""
with self.assertRaises(Exception):
sec_grp_settings = SecurityGroupSettings()
self.security_groups.append(
- neutron_utils.create_security_group(self.neutron, self.keystone, sec_grp_settings))
+ neutron_utils.create_security_group(self.neutron,
+ self.keystone,
+ sec_grp_settings))
def test_create_sec_grp_no_rules(self):
"""
Tests the neutron_utils.create_security_group() function
"""
- sec_grp_settings = SecurityGroupSettings(name=self.sec_grp_name, description='hello group')
- self.security_groups.append(neutron_utils.create_security_group(self.neutron, self.keystone, sec_grp_settings))
+ sec_grp_settings = SecurityGroupSettings(name=self.sec_grp_name,
+ description='hello group')
+ self.security_groups.append(
+ neutron_utils.create_security_group(self.neutron, self.keystone,
+ sec_grp_settings))
- self.assertTrue(sec_grp_settings.name, self.security_groups[0]['security_group']['name'])
- self.assertTrue(sec_grp_settings.description, self.security_groups[0]['security_group']['description'])
+ self.assertTrue(sec_grp_settings.name,
+ self.security_groups[0]['security_group']['name'])
+ self.assertTrue(sec_grp_settings.description,
+ self.security_groups[0]['security_group'][
+ 'description'])
- sec_grp_get = neutron_utils.get_security_group(self.neutron, sec_grp_settings.name)
+ sec_grp_get = neutron_utils.get_security_group(self.neutron,
+ sec_grp_settings.name)
self.assertIsNotNone(sec_grp_get)
self.assertTrue(validation_utils.objects_equivalent(
- self.security_groups[0]['security_group'], sec_grp_get['security_group']))
+ self.security_groups[0]['security_group'],
+ sec_grp_get['security_group']))
def test_create_sec_grp_one_rule(self):
"""
Tests the neutron_utils.create_security_group() function
"""
- sec_grp_rule_settings = SecurityGroupRuleSettings(sec_grp_name=self.sec_grp_name, direction=Direction.ingress)
- sec_grp_settings = SecurityGroupSettings(name=self.sec_grp_name, description='hello group',
- rule_settings=[sec_grp_rule_settings])
+ sec_grp_rule_settings = SecurityGroupRuleSettings(
+ sec_grp_name=self.sec_grp_name, direction=Direction.ingress)
+ sec_grp_settings = SecurityGroupSettings(
+ name=self.sec_grp_name, description='hello group',
+ rule_settings=[sec_grp_rule_settings])
- self.security_groups.append(neutron_utils.create_security_group(self.neutron, self.keystone, sec_grp_settings))
- free_rules = neutron_utils.get_rules_by_security_group(self.neutron, self.security_groups[0])
+ self.security_groups.append(
+ neutron_utils.create_security_group(self.neutron, self.keystone,
+ sec_grp_settings))
+ free_rules = neutron_utils.get_rules_by_security_group(
+ self.neutron, self.security_groups[0])
for free_rule in free_rules:
self.security_group_rules.append(free_rule)
self.security_group_rules.append(
- neutron_utils.create_security_group_rule(self.neutron, sec_grp_settings.rule_settings[0]))
+ neutron_utils.create_security_group_rule(
+ self.neutron, sec_grp_settings.rule_settings[0]))
# Refresh object so it is populated with the newly added rule
- security_group = neutron_utils.get_security_group(self.neutron, sec_grp_settings.name)
+ security_group = neutron_utils.get_security_group(
+ self.neutron, sec_grp_settings.name)
- rules = neutron_utils.get_rules_by_security_group(self.neutron, security_group)
+ rules = neutron_utils.get_rules_by_security_group(self.neutron,
+ security_group)
- self.assertTrue(validation_utils.objects_equivalent(self.security_group_rules, rules))
+ self.assertTrue(
+ validation_utils.objects_equivalent(self.security_group_rules,
+ rules))
- self.assertTrue(sec_grp_settings.name, security_group['security_group']['name'])
- self.assertTrue(sec_grp_settings.description, security_group['security_group']['description'])
+ self.assertTrue(sec_grp_settings.name,
+ security_group['security_group']['name'])
+ self.assertTrue(sec_grp_settings.description,
+ security_group['security_group']['description'])
- sec_grp_get = neutron_utils.get_security_group(self.neutron, sec_grp_settings.name)
+ sec_grp_get = neutron_utils.get_security_group(self.neutron,
+ sec_grp_settings.name)
self.assertIsNotNone(sec_grp_get)
self.assertTrue(validation_utils.objects_equivalent(
security_group['security_group'], sec_grp_get['security_group']))
self.security_groups.append(neutron_utils.create_security_group(
self.neutron, self.keystone,
- SecurityGroupSettings(name=self.sec_grp_name + '-1', description='hello group')))
+ SecurityGroupSettings(name=self.sec_grp_name + '-1',
+ description='hello group')))
self.security_groups.append(neutron_utils.create_security_group(
self.neutron, self.keystone,
- SecurityGroupSettings(name=self.sec_grp_name + '-2', description='hello group')))
+ SecurityGroupSettings(name=self.sec_grp_name + '-2',
+ description='hello group')))
sec_grp_1b = neutron_utils.get_security_group_by_id(
self.neutron, self.security_groups[0]['security_group']['id'])
sec_grp_2b = neutron_utils.get_security_group_by_id(
self.neutron, self.security_groups[1]['security_group']['id'])
- self.assertEqual(self.security_groups[0]['security_group']['id'], sec_grp_1b['security_group']['id'])
- self.assertEqual(self.security_groups[1]['security_group']['id'], sec_grp_2b['security_group']['id'])
+ self.assertEqual(self.security_groups[0]['security_group']['id'],
+ sec_grp_1b['security_group']['id'])
+ self.assertEqual(self.security_groups[1]['security_group']['id'],
+ sec_grp_2b['security_group']['id'])
+
+
+class NeutronUtilsFloatingIpTests(OSComponentTestCase):
+ """
+ Test basic nova keypair functionality
+ """
+
+ def setUp(self):
+ """
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file within OpenStack
+ """
+ self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.floating_ip = None
+
+ def tearDown(self):
+ """
+ Cleans the image and downloaded image file
+ """
+ if self.floating_ip:
+ neutron_utils.delete_floating_ip(self.neutron, self.floating_ip)
+
+ def test_floating_ips(self):
+ """
+ Tests the creation of a floating IP
+ :return:
+ """
+ initial_fips = neutron_utils.get_floating_ips(self.neutron)
+
+ self.floating_ip = neutron_utils.create_floating_ip(self.neutron,
+ self.ext_net_name)
+ all_fips = neutron_utils.get_floating_ips(self.neutron)
+ self.assertEqual(len(initial_fips) + 1, len(all_fips))
+ returned = neutron_utils.get_floating_ip(self.neutron,
+ self.floating_ip)
+ self.assertEqual(self.floating_ip.id, returned.id)
+ self.assertEqual(self.floating_ip.ip, returned.ip)
"""
def validate_network(neutron, name, exists):
"""
- Returns true if a network for a given name DOES NOT exist if the exists parameter is false conversely true.
- Returns false if a network for a given name DOES exist if the exists parameter is true conversely false.
+ Returns true if a network for a given name DOES NOT exist if the exists
+ parameter is false conversely true. Returns false if a network for a given
+ name DOES exist if the exists parameter is true conversely false.
:param neutron: The neutron client
:param name: The expected network name
:param exists: Whether or not the network name should exist or not
def validate_subnet(neutron, name, cidr, exists):
"""
- Returns true if a subnet for a given name DOES NOT exist if the exists parameter is false conversely true.
- Returns false if a subnet for a given name DOES exist if the exists parameter is true conversely false.
+ Returns true if a subnet for a given name DOES NOT exist if the exists
+ parameter is false conversely true. Returns false if a subnet for a given
+ name DOES exist if the exists parameter is true conversely false.
:param neutron: The neutron client
:param name: The expected subnet name
:param cidr: The expected CIDR value
def validate_router(neutron, name, exists):
"""
- Returns true if a router for a given name DOES NOT exist if the exists parameter is false conversely true.
- Returns false if a router for a given name DOES exist if the exists parameter is true conversely false.
+ Returns true if a router for a given name DOES NOT exist if the exists
+ parameter is false conversely true. Returns false if a router for a given
+ name DOES exist if the exists parameter is true conversely false.
:param neutron: The neutron client
:param name: The expected router name
:param exists: Whether or not the network name should exist or not
def validate_interface_router(interface_router, router, subnet):
"""
- Returns true if the router ID & subnet ID have been properly included into the interface router object
+ Returns true if the router ID & subnet ID have been properly included into
+ the interface router object
:param interface_router: the object to validate
:param router: to validate against the interface_router
:param subnet: to validate against the interface_router
def validate_port(neutron, port_obj, this_port_name):
"""
- Returns true if a port for a given name DOES NOT exist if the exists parameter is false conversely true.
- Returns false if a port for a given name DOES exist if the exists parameter is true conversely false.
+ Returns true if a port for a given name DOES NOT exist if the exists
+ parameter is false conversely true. Returns false if a port for a given
+ name DOES exist if the exists parameter is true conversely false.
:param neutron: The neutron client
:param port_obj: The port object to lookup
:param this_port_name: The expected router name
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
-import os
import uuid
-from snaps.openstack.utils import nova_utils
+import os
+from snaps.openstack.create_flavor import FlavorSettings, OpenStackFlavor
+from snaps.openstack.create_image import OpenStackImage
+from snaps.openstack.create_instance import VmInstanceSettings
+from snaps.openstack.create_network import OpenStackNetwork, PortSettings
+from snaps.openstack.tests import openstack_tests
from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
-from snaps.openstack.create_flavor import FlavorSettings
+from snaps.openstack.utils import nova_utils, neutron_utils, glance_utils
__author__ = 'spisarski'
from snaps.openstack.os_credentials import OSCreds
nova = nova_utils.nova_client(
- OSCreds(username='user', password='pass', auth_url=self.os_creds.auth_url,
- project_name=self.os_creds.project_name, proxy_settings=self.os_creds.proxy_settings))
+ OSCreds(username='user', password='pass',
+ auth_url=self.os_creds.auth_url,
+ project_name=self.os_creds.project_name,
+ proxy_settings=self.os_creds.proxy_settings))
# This should throw an exception
with self.assertRaises(Exception):
def setUp(self):
"""
- Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
- within OpenStack
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file within OpenStack
"""
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.priv_key_file_path = 'tmp/' + guid
self.public_key = nova_utils.public_key_openssh(self.keys)
self.keypair_name = guid
self.keypair = None
- self.floating_ip = None
def tearDown(self):
"""
except:
pass
- if self.floating_ip:
- nova_utils.delete_floating_ip(self.nova, self.floating_ip)
-
def test_create_keypair(self):
"""
Tests the creation of an OpenStack keypair that does not exist.
"""
- self.keypair = nova_utils.upload_keypair(self.nova, self.keypair_name, self.public_key)
+ self.keypair = nova_utils.upload_keypair(self.nova, self.keypair_name,
+ self.public_key)
result = nova_utils.keypair_exists(self.nova, self.keypair)
self.assertEqual(self.keypair, result)
keypair = nova_utils.get_keypair_by_name(self.nova, self.keypair_name)
"""
Tests the creation of an OpenStack keypair that does not exist.
"""
- self.keypair = nova_utils.upload_keypair(self.nova, self.keypair_name, self.public_key)
+ self.keypair = nova_utils.upload_keypair(self.nova, self.keypair_name,
+ self.public_key)
result = nova_utils.keypair_exists(self.nova, self.keypair)
self.assertEqual(self.keypair, result)
nova_utils.delete_keypair(self.nova, self.keypair)
Tests that the generated RSA keys are properly saved to files
:return:
"""
- nova_utils.save_keys_to_files(self.keys, self.pub_key_file_path, self.priv_key_file_path)
- self.keypair = nova_utils.upload_keypair_file(self.nova, self.keypair_name, self.pub_key_file_path)
+ nova_utils.save_keys_to_files(self.keys, self.pub_key_file_path,
+ self.priv_key_file_path)
+ self.keypair = nova_utils.upload_keypair_file(self.nova,
+ self.keypair_name,
+ self.pub_key_file_path)
pub_key_file = open(os.path.expanduser(self.pub_key_file_path))
pub_key = pub_key_file.read()
pub_key_file.close()
self.assertEqual(self.keypair.public_key, pub_key)
- def test_floating_ips(self):
- """
- Tests the creation of a floating IP
- :return:
- """
- ips = nova_utils.get_floating_ips(self.nova)
- self.assertIsNotNone(ips)
-
- self.floating_ip = nova_utils.create_floating_ip(self.nova, self.ext_net_name)
- returned = nova_utils.get_floating_ip(self.nova, self.floating_ip)
- self.assertEqual(self.floating_ip, returned)
-
class NovaUtilsFlavorTests(OSComponentTestCase):
"""
def setUp(self):
"""
- Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
- within OpenStack
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file within OpenStack
"""
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
- self.flavor_settings = FlavorSettings(name=guid + '-name', flavor_id=guid + '-id', ram=1, disk=1, vcpus=1,
- ephemeral=1, swap=2, rxtx_factor=3.0, is_public=False)
+ self.flavor_settings = FlavorSettings(name=guid + '-name',
+ flavor_id=guid + '-id', ram=1,
+ disk=1, vcpus=1,
+ ephemeral=1, swap=2,
+ rxtx_factor=3.0, is_public=False)
self.nova = nova_utils.nova_client(self.os_creds)
self.flavor = None
self.flavor = nova_utils.create_flavor(self.nova, self.flavor_settings)
self.validate_flavor()
nova_utils.delete_flavor(self.nova, self.flavor)
- flavor = nova_utils.get_flavor_by_name(self.nova, self.flavor_settings.name)
+ flavor = nova_utils.get_flavor_by_name(self.nova,
+ self.flavor_settings.name)
self.assertIsNone(flavor)
def validate_flavor(self):
else:
self.assertEqual(self.flavor_settings.swap, self.flavor.swap)
- self.assertEqual(self.flavor_settings.rxtx_factor, self.flavor.rxtx_factor)
+ self.assertEqual(self.flavor_settings.rxtx_factor,
+ self.flavor.rxtx_factor)
self.assertEqual(self.flavor_settings.is_public, self.flavor.is_public)
+
+
+class NovaUtilsInstanceTests(OSComponentTestCase):
+ """
+ Tests the creation of VM instances via nova_utils.py
+ """
+
+ def setUp(self):
+ """
+ Setup objects required by VM instances
+ :return:
+ """
+
+ guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+
+ self.nova = nova_utils.nova_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.glance = glance_utils.glance_client(self.os_creds)
+
+ self.image_creator = None
+ self.network_creator = None
+ self.flavor_creator = None
+ self.port = None
+ self.vm_inst = None
+
+ try:
+ image_settings = openstack_tests.cirros_image_settings(
+ name=guid + '-image', image_metadata=self.image_metadata)
+ self.image_creator = OpenStackImage(
+ self.os_creds, image_settings=image_settings)
+ self.image_creator.create()
+
+ network_settings = openstack_tests.get_priv_net_config(
+ guid + '-net', guid + '-subnet').network_settings
+ self.network_creator = OpenStackNetwork(
+ self.os_creds, network_settings)
+ self.network_creator.create()
+
+ self.flavor_creator = OpenStackFlavor(
+ self.os_creds,
+ FlavorSettings(
+ name=guid + '-flavor-name', ram=128, disk=10, vcpus=1))
+ self.flavor_creator.create()
+
+ port_settings = PortSettings(name=guid + '-port',
+ network_name=network_settings.name)
+ self.port = neutron_utils.create_port(
+ self.neutron, self.os_creds, port_settings)
+
+ self.instance_settings = VmInstanceSettings(
+ name=guid + '-vm_inst',
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=[port_settings])
+ except:
+ self.tearDown()
+ raise
+
+ def tearDown(self):
+ """
+ Cleanup deployed resources
+ :return:
+ """
+ if self.vm_inst:
+ try:
+ nova_utils.delete_vm_instance(self.nova, self.vm_inst)
+ except:
+ pass
+ if self.port:
+ try:
+ neutron_utils.delete_port(self.neutron, self.port)
+ except:
+ pass
+ if self.flavor_creator:
+ try:
+ self.flavor_creator.clean()
+ except:
+ pass
+ if self.network_creator:
+ try:
+ self.network_creator.clean()
+ except:
+ pass
+ if self.image_creator:
+ try:
+ self.image_creator.clean()
+ except:
+ pass
+
+ def test_create_instance(self):
+ """
+ Tests the nova_utils.create_server() method
+ :return:
+ """
+
+ self.vm_inst = nova_utils.create_server(
+ self.nova, self.neutron, self.glance, self.instance_settings,
+ self.image_creator.image_settings)
+
+ self.assertIsNotNone(self.vm_inst)
+
+ vm_inst = nova_utils.get_latest_server_object(self.nova, self.vm_inst)
+
+ self.assertEqual(self.vm_inst.name, vm_inst.name)
+ self.assertEqual(self.vm_inst.id, vm_inst.id)
# See the License for the specific language governing permissions and
# limitations under the License.
+import uuid
+
import os
import pkg_resources
-import uuid
from scp import SCPClient
-from snaps.openstack.create_security_group import SecurityGroupRuleSettings, Direction, Protocol, \
- OpenStackSecurityGroup, SecurityGroupSettings
-
from snaps.openstack import create_flavor
-from snaps.openstack import create_instance
from snaps.openstack import create_image
+from snaps.openstack import create_instance
from snaps.openstack import create_keypairs
from snaps.openstack import create_network
from snaps.openstack import create_router
+from snaps.openstack.create_security_group import (
+ SecurityGroupRuleSettings, Direction, Protocol, OpenStackSecurityGroup,
+ SecurityGroupSettings)
from snaps.openstack.tests import openstack_tests, create_instance_tests
from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
+from snaps.openstack.utils import nova_utils
from snaps.provisioning import ansible_utils
VM_BOOT_TIMEOUT = 600
class AnsibleProvisioningTests(OSIntegrationTestCase):
"""
- Test for the CreateInstance class with two NIC/Ports, eth0 with floating IP and eth1 w/o
+ Test for the CreateInstance class with two NIC/Ports, eth0 with floating IP
+ and eth1 w/o
"""
def setUp(self):
"""
- Instantiates the CreateImage object that is responsible for downloading and creating an OS image file
- within OpenStack
+ Instantiates the CreateImage object that is responsible for downloading
+ and creating an OS image file within OpenStack
"""
super(self.__class__, self).__start__()
+ self.nova = nova_utils.nova_client(self.os_creds)
+
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.keypair_priv_filepath = 'tmp/' + guid
self.keypair_pub_filepath = self.keypair_priv_filepath + '.pub'
try:
# Create Image
- os_image_settings = openstack_tests.ubuntu_image_settings(name=guid + '-' + '-image',
- image_metadata=self.image_metadata)
- self.image_creator = create_image.OpenStackImage(self.os_creds, os_image_settings)
+ os_image_settings = openstack_tests.ubuntu_image_settings(
+ name=guid + '-' + '-image',
+ image_metadata=self.image_metadata)
+ self.image_creator = create_image.OpenStackImage(self.os_creds,
+ os_image_settings)
self.image_creator.create()
# First network is public
self.pub_net_config = openstack_tests.get_pub_net_config(
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
- router_name=guid + '-pub-router', external_net=self.ext_net_name)
+ router_name=guid + '-pub-router',
+ external_net=self.ext_net_name)
- self.network_creator = create_network.OpenStackNetwork(self.os_creds, self.pub_net_config.network_settings)
+ self.network_creator = create_network.OpenStackNetwork(
+ self.os_creds, self.pub_net_config.network_settings)
self.network_creator.create()
# Create routers
- self.router_creator = create_router.OpenStackRouter(self.os_creds, self.pub_net_config.router_settings)
+ self.router_creator = create_router.OpenStackRouter(
+ self.os_creds, self.pub_net_config.router_settings)
self.router_creator.create()
# Create Flavor
self.flavor_creator = create_flavor.OpenStackFlavor(
self.admin_os_creds,
- create_flavor.FlavorSettings(name=guid + '-flavor-name', ram=2048, disk=10, vcpus=2,
+ create_flavor.FlavorSettings(name=guid + '-flavor-name',
+ ram=2048, disk=10, vcpus=2,
metadata=self.flavor_metadata))
self.flavor_creator.create()
# Create Key/Pair
self.keypair_creator = create_keypairs.OpenStackKeypair(
self.os_creds, create_keypairs.KeypairSettings(
- name=self.keypair_name, public_filepath=self.keypair_pub_filepath,
+ name=self.keypair_name,
+ public_filepath=self.keypair_pub_filepath,
private_filepath=self.keypair_priv_filepath))
self.keypair_creator.create()
# Create Security Group
sec_grp_name = guid + '-sec-grp'
- rule1 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name, direction=Direction.ingress,
+ rule1 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name,
+ direction=Direction.ingress,
protocol=Protocol.icmp)
- rule2 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name, direction=Direction.ingress,
- protocol=Protocol.tcp, port_range_min=22, port_range_max=22)
+ rule2 = SecurityGroupRuleSettings(sec_grp_name=sec_grp_name,
+ direction=Direction.ingress,
+ protocol=Protocol.tcp,
+ port_range_min=22,
+ port_range_max=22)
self.sec_grp_creator = OpenStackSecurityGroup(
self.os_creds,
- SecurityGroupSettings(name=sec_grp_name, rule_settings=[rule1, rule2]))
+ SecurityGroupSettings(name=sec_grp_name,
+ rule_settings=[rule1, rule2]))
self.sec_grp_creator.create()
# Create instance
ports_settings = list()
ports_settings.append(
- create_network.PortSettings(name=self.port_1_name,
- network_name=self.pub_net_config.network_settings.name))
+ create_network.PortSettings(
+ name=self.port_1_name,
+ network_name=self.pub_net_config.network_settings.name))
instance_settings = create_instance.VmInstanceSettings(
- name=self.vm_inst_name, flavor=self.flavor_creator.flavor_settings.name, port_settings=ports_settings,
+ name=self.vm_inst_name,
+ flavor=self.flavor_creator.flavor_settings.name,
+ port_settings=ports_settings,
floating_ip_settings=[create_instance.FloatingIpSettings(
name=self.floating_ip_name, port_name=self.port_1_name,
router_name=self.pub_net_config.router_settings.name)])
self.inst_creator = create_instance.OpenStackVmInstance(
- self.os_creds, instance_settings, self.image_creator.image_settings,
+ self.os_creds, instance_settings,
+ self.image_creator.image_settings,
keypair_settings=self.keypair_creator.keypair_settings)
except:
self.tearDown()
def test_apply_simple_playbook(self):
"""
- Tests application of an Ansible playbook that simply copies over a file:
- 1. Have a ~/.ansible.cfg (or alternate means) to set host_key_checking = False
- 2. Set the following environment variable in your executing shell: ANSIBLE_HOST_KEY_CHECKING=False
- Should this not be performed, the creation of the host ssh key will cause your ansible calls to fail.
+ Tests application of an Ansible playbook that simply copies over a file
+ 1. Have a ~/.ansible.cfg (or alternate means) to
+ set host_key_checking = False
+ 2. Set the following environment variable in your executing shell:
+ ANSIBLE_HOST_KEY_CHECKING=False
+ Should this not be performed, the creation of the host ssh key will
+ cause your ansible calls to fail.
"""
vm = self.inst_creator.create(block=True)
self.assertTrue(self.inst_creator.vm_ssh_active(block=True))
priv_ip = self.inst_creator.get_port_ip(self.port_1_name)
- self.assertTrue(create_instance_tests.check_dhcp_lease(vm, priv_ip))
+ self.assertTrue(create_instance_tests.check_dhcp_lease(
+ self.nova, vm, priv_ip))
# Apply Security Group
- self.inst_creator.add_security_group(self.sec_grp_creator.get_security_group())
+ self.inst_creator.add_security_group(
+ self.sec_grp_creator.get_security_group())
ssh_client = self.inst_creator.ssh_client()
self.assertIsNotNone(ssh_client)
self.assertIsNotNone(out)
self.assertGreater(len(out), 1)
- # Need to use the first floating IP as subsequent ones are currently broken with Apex CO
+ # Need to use the first floating IP as subsequent ones are currently
+ # broken with Apex CO
ip = self.inst_creator.get_floating_ip().ip
user = self.inst_creator.get_image_user()
priv_key = self.inst_creator.keypair_settings.private_filepath
- relative_pb_path = pkg_resources.resource_filename('snaps.provisioning.tests.playbooks', 'simple_playbook.yml')
+ relative_pb_path = pkg_resources.resource_filename(
+ 'snaps.provisioning.tests.playbooks', 'simple_playbook.yml')
retval = self.inst_creator.apply_ansible_playbook(relative_pb_path)
self.assertEqual(0, retval)
- ssh = ansible_utils.ssh_client(ip, user, priv_key, self.os_creds.proxy_settings)
+ ssh = ansible_utils.ssh_client(ip, user, priv_key,
+ self.os_creds.proxy_settings)
self.assertIsNotNone(ssh)
scp = SCPClient(ssh.get_transport())
scp.get('~/hello.txt', self.test_file_local_path)
def test_apply_template_playbook(self):
"""
- Tests application of an Ansible playbook that applies a template to a file:
- 1. Have a ~/.ansible.cfg (or alternate means) to set host_key_checking = False
- 2. Set the following environment variable in your executing shell: ANSIBLE_HOST_KEY_CHECKING=False
- Should this not be performed, the creation of the host ssh key will cause your ansible calls to fail.
+ Tests application of an Ansible playbook that applies a template to a
+ file:
+ 1. Have a ~/.ansible.cfg (or alternate means) to set
+ host_key_checking = False
+ 2. Set the following environment variable in your executing shell:
+ ANSIBLE_HOST_KEY_CHECKING=False
+ Should this not be performed, the creation of the host ssh key will
+ cause your ansible calls to fail.
"""
vm = self.inst_creator.create(block=True)
self.assertTrue(self.inst_creator.vm_ssh_active(block=True))
priv_ip = self.inst_creator.get_port_ip(self.port_1_name)
- self.assertTrue(create_instance_tests.check_dhcp_lease(vm, priv_ip))
+ self.assertTrue(create_instance_tests.check_dhcp_lease(
+ self.nova, vm, priv_ip))
# Apply Security Group
- self.inst_creator.add_security_group(self.sec_grp_creator.get_security_group())
+ self.inst_creator.add_security_group(
+ self.sec_grp_creator.get_security_group())
- # Need to use the first floating IP as subsequent ones are currently broken with Apex CO
+ # Need to use the first floating IP as subsequent ones are currently
+ # broken with Apex CO
ip = self.inst_creator.get_floating_ip().ip
user = self.inst_creator.get_image_user()
priv_key = self.inst_creator.keypair_settings.private_filepath
- relative_pb_path = pkg_resources.resource_filename('snaps.provisioning.tests.playbooks',
- 'template_playbook.yml')
- retval = self.inst_creator.apply_ansible_playbook(relative_pb_path, variables={'name': 'Foo'})
+ relative_pb_path = pkg_resources.resource_filename(
+ 'snaps.provisioning.tests.playbooks',
+ 'template_playbook.yml')
+ retval = self.inst_creator.apply_ansible_playbook(relative_pb_path,
+ variables={
+ 'name': 'Foo'})
self.assertEqual(0, retval)
- ssh = ansible_utils.ssh_client(ip, user, priv_key, self.os_creds.proxy_settings)
+ ssh = ansible_utils.ssh_client(ip, user, priv_key,
+ self.os_creds.proxy_settings)
self.assertIsNotNone(ssh)
scp = SCPClient(ssh.get_transport())
scp.get('/tmp/hello.txt', self.test_file_local_path)
from snaps.domain.test.image_tests import ImageDomainObjectTests
from snaps.domain.test.stack_tests import StackDomainObjectTests
-from snaps.openstack.tests.create_stack_tests import StackSettingsUnitTests, CreateStackSuccessTests, \
- CreateStackNegativeTests
-from snaps.openstack.utils.tests.glance_utils_tests import GlanceSmokeTests, GlanceUtilsTests
+from snaps.domain.test.vm_inst_tests import (VmInstDomainObjectTests,
+ FloatingIpDomainObjectTests)
from snaps.openstack.tests.create_flavor_tests import CreateFlavorTests
-from snaps.openstack.utils.tests.heat_utils_tests import HeatUtilsCreateStackTests, HeatSmokeTests
+from snaps.openstack.tests.create_image_tests import (
+ CreateImageSuccessTests, CreateImageNegativeTests, ImageSettingsUnitTests,
+ CreateMultiPartImageTests)
+from snaps.openstack.tests.create_instance_tests import (
+ CreateInstanceSingleNetworkTests, CreateInstancePubPrivNetTests,
+ CreateInstanceOnComputeHost, CreateInstanceSimpleTests,
+ FloatingIpSettingsUnitTests, InstanceSecurityGroupTests,
+ VmInstanceSettingsUnitTests, CreateInstancePortManipulationTests,
+ SimpleHealthCheck, CreateInstanceFromThreePartImage,
+ CreateInstanceMockOfflineTests)
+from snaps.openstack.tests.create_keypairs_tests import (
+ CreateKeypairsTests, KeypairSettingsUnitTests)
+from snaps.openstack.tests.create_network_tests import (
+ CreateNetworkSuccessTests, NetworkSettingsUnitTests, PortSettingsUnitTests,
+ SubnetSettingsUnitTests, CreateNetworkTypeTests)
+from snaps.openstack.tests.create_project_tests import (
+ CreateProjectSuccessTests, ProjectSettingsUnitTests,
+ CreateProjectUserTests)
+from snaps.openstack.tests.create_router_tests import (
+ CreateRouterSuccessTests, CreateRouterNegativeTests)
+from snaps.openstack.tests.create_security_group_tests import (
+ CreateSecurityGroupTests, SecurityGroupRuleSettingsUnitTests,
+ SecurityGroupSettingsUnitTests)
+from snaps.openstack.tests.create_stack_tests import (
+ StackSettingsUnitTests, CreateStackSuccessTests, CreateStackNegativeTests)
+from snaps.openstack.tests.create_user_tests import (
+ UserSettingsUnitTests, CreateUserSuccessTests)
+from snaps.openstack.tests.os_source_file_test import (
+ OSComponentTestCase, OSIntegrationTestCase)
+from snaps.openstack.utils.tests.glance_utils_tests import (
+ GlanceSmokeTests, GlanceUtilsTests)
+from snaps.openstack.utils.tests.heat_utils_tests import (
+ HeatUtilsCreateStackTests, HeatSmokeTests)
+from snaps.openstack.utils.tests.keystone_utils_tests import (
+ KeystoneSmokeTests, KeystoneUtilsTests)
+from snaps.openstack.utils.tests.neutron_utils_tests import (
+ NeutronSmokeTests, NeutronUtilsNetworkTests, NeutronUtilsSubnetTests,
+ NeutronUtilsRouterTests, NeutronUtilsSecurityGroupTests,
+ NeutronUtilsFloatingIpTests)
+from snaps.openstack.utils.tests.nova_utils_tests import (
+ NovaSmokeTests, NovaUtilsKeypairTests, NovaUtilsFlavorTests,
+ NovaUtilsInstanceTests)
+from snaps.provisioning.tests.ansible_utils_tests import (
+ AnsibleProvisioningTests)
from snaps.tests.file_utils_tests import FileUtilsTests
-from snaps.openstack.tests.create_security_group_tests import CreateSecurityGroupTests, \
- SecurityGroupRuleSettingsUnitTests, SecurityGroupSettingsUnitTests
-from snaps.openstack.tests.create_project_tests import CreateProjectSuccessTests, ProjectSettingsUnitTests, \
- CreateProjectUserTests
-from snaps.openstack.tests.create_user_tests import UserSettingsUnitTests, CreateUserSuccessTests
-from snaps.openstack.utils.tests.keystone_utils_tests import KeystoneSmokeTests, KeystoneUtilsTests
-from snaps.openstack.utils.tests.neutron_utils_tests import NeutronSmokeTests, NeutronUtilsNetworkTests, \
- NeutronUtilsSubnetTests, NeutronUtilsRouterTests, NeutronUtilsSecurityGroupTests
-from snaps.openstack.tests.create_image_tests import CreateImageSuccessTests, CreateImageNegativeTests, \
- ImageSettingsUnitTests, CreateMultiPartImageTests
-from snaps.openstack.tests.create_keypairs_tests import CreateKeypairsTests, KeypairSettingsUnitTests
-from snaps.openstack.tests.create_network_tests import CreateNetworkSuccessTests, NetworkSettingsUnitTests, \
- PortSettingsUnitTests, SubnetSettingsUnitTests, CreateNetworkTypeTests
-from snaps.openstack.tests.create_router_tests import CreateRouterSuccessTests, CreateRouterNegativeTests
-from snaps.openstack.tests.create_instance_tests import CreateInstanceSingleNetworkTests, \
- CreateInstancePubPrivNetTests, CreateInstanceOnComputeHost, CreateInstanceSimpleTests, \
- FloatingIpSettingsUnitTests, InstanceSecurityGroupTests, VmInstanceSettingsUnitTests, \
- CreateInstancePortManipulationTests, SimpleHealthCheck, CreateInstanceFromThreePartImage, \
- CreateInstanceMockOfflineTests
-from snaps.provisioning.tests.ansible_utils_tests import AnsibleProvisioningTests
-from snaps.openstack.tests.os_source_file_test import OSComponentTestCase, OSIntegrationTestCase
-from snaps.openstack.utils.tests.nova_utils_tests import NovaSmokeTests, NovaUtilsKeypairTests, NovaUtilsFlavorTests
__author__ = 'spisarski'
:return: None as the tests will be adding to the 'suite' parameter object
"""
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(FileUtilsTests))
- suite.addTest(unittest.TestLoader().loadTestsFromTestCase(SecurityGroupRuleSettingsUnitTests))
- suite.addTest(unittest.TestLoader().loadTestsFromTestCase(SecurityGroupSettingsUnitTests))
- suite.addTest(unittest.TestLoader().loadTestsFromTestCase(ImageSettingsUnitTests))
- suite.addTest(unittest.TestLoader().loadTestsFromTestCase(ImageDomainObjectTests))
- suite.addTest(unittest.TestLoader().loadTestsFromTestCase(KeypairSettingsUnitTests))
- suite.addTest(unittest.TestLoader().loadTestsFromTestCase(UserSettingsUnitTests))
- suite.addTest(unittest.TestLoader().loadTestsFromTestCase(ProjectSettingsUnitTests))
- suite.addTest(unittest.TestLoader().loadTestsFromTestCase(NetworkSettingsUnitTests))
- suite.addTest(unittest.TestLoader().loadTestsFromTestCase(SubnetSettingsUnitTests))
- suite.addTest(unittest.TestLoader().loadTestsFromTestCase(PortSettingsUnitTests))
- suite.addTest(unittest.TestLoader().loadTestsFromTestCase(FloatingIpSettingsUnitTests))
- suite.addTest(unittest.TestLoader().loadTestsFromTestCase(VmInstanceSettingsUnitTests))
- suite.addTest(unittest.TestLoader().loadTestsFromTestCase(StackDomainObjectTests))
- suite.addTest(unittest.TestLoader().loadTestsFromTestCase(StackSettingsUnitTests))
-
-
-def add_openstack_client_tests(suite, os_creds, ext_net_name, use_keystone=True, log_level=logging.INFO):
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ SecurityGroupRuleSettingsUnitTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ SecurityGroupSettingsUnitTests))
+ suite.addTest(
+ unittest.TestLoader().loadTestsFromTestCase(ImageSettingsUnitTests))
+ suite.addTest(
+ unittest.TestLoader().loadTestsFromTestCase(ImageDomainObjectTests))
+ suite.addTest(
+ unittest.TestLoader().loadTestsFromTestCase(KeypairSettingsUnitTests))
+ suite.addTest(
+ unittest.TestLoader().loadTestsFromTestCase(UserSettingsUnitTests))
+ suite.addTest(
+ unittest.TestLoader().loadTestsFromTestCase(ProjectSettingsUnitTests))
+ suite.addTest(
+ unittest.TestLoader().loadTestsFromTestCase(NetworkSettingsUnitTests))
+ suite.addTest(
+ unittest.TestLoader().loadTestsFromTestCase(SubnetSettingsUnitTests))
+ suite.addTest(
+ unittest.TestLoader().loadTestsFromTestCase(PortSettingsUnitTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ FloatingIpSettingsUnitTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ VmInstanceSettingsUnitTests))
+ suite.addTest(
+ unittest.TestLoader().loadTestsFromTestCase(StackDomainObjectTests))
+ suite.addTest(
+ unittest.TestLoader().loadTestsFromTestCase(StackSettingsUnitTests))
+ suite.addTest(
+ unittest.TestLoader().loadTestsFromTestCase(VmInstDomainObjectTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ FloatingIpDomainObjectTests))
+
+
+def add_openstack_client_tests(suite, os_creds, ext_net_name,
+ use_keystone=True, log_level=logging.INFO):
"""
Adds tests written to exercise OpenStack client retrieval
: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 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 log_level: the logging level
:return: None as the tests will be adding to the 'suite' parameter object
"""
# Basic connection tests
- suite.addTest(OSComponentTestCase.parameterize(GlanceSmokeTests, os_creds=os_creds, ext_net_name=ext_net_name,
- log_level=log_level))
+ suite.addTest(
+ OSComponentTestCase.parameterize(
+ GlanceSmokeTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ log_level=log_level))
if use_keystone:
- suite.addTest(OSComponentTestCase.parameterize(KeystoneSmokeTests, os_creds=os_creds, ext_net_name=ext_net_name,
- log_level=log_level))
+ suite.addTest(
+ OSComponentTestCase.parameterize(
+ KeystoneSmokeTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, log_level=log_level))
- suite.addTest(OSComponentTestCase.parameterize(NeutronSmokeTests, os_creds=os_creds, ext_net_name=ext_net_name,
- log_level=log_level))
- suite.addTest(OSComponentTestCase.parameterize(NovaSmokeTests, os_creds=os_creds, ext_net_name=ext_net_name,
- log_level=log_level))
- suite.addTest(OSComponentTestCase.parameterize(HeatSmokeTests, os_creds=os_creds, ext_net_name=ext_net_name,
- log_level=log_level))
+ suite.addTest(
+ OSComponentTestCase.parameterize(
+ NeutronSmokeTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ log_level=log_level))
+ suite.addTest(
+ OSComponentTestCase.parameterize(
+ NovaSmokeTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ log_level=log_level))
+ suite.addTest(
+ OSComponentTestCase.parameterize(
+ HeatSmokeTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ log_level=log_level))
-def add_openstack_api_tests(suite, os_creds, ext_net_name, use_keystone=True, image_metadata=None,
- log_level=logging.INFO):
+def add_openstack_api_tests(suite, os_creds, ext_net_name, use_keystone=True,
+ image_metadata=None, log_level=logging.INFO):
"""
Adds tests written to exercise all existing OpenStack APIs
: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
+ :param os_creds: 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 log_level: the logging level
:return: None as the tests will be adding to the 'suite' parameter object
# Tests the OpenStack API calls
if use_keystone:
suite.addTest(OSComponentTestCase.parameterize(
- KeystoneUtilsTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
+ KeystoneUtilsTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
- CreateUserSuccessTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
+ CreateUserSuccessTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
- CreateProjectSuccessTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
+ CreateProjectSuccessTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
- CreateProjectUserTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
+ CreateProjectUserTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
- GlanceUtilsTests, os_creds=os_creds, ext_net_name=ext_net_name, image_metadata=image_metadata,
+ GlanceUtilsTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ image_metadata=image_metadata,
+ log_level=log_level))
+ suite.addTest(OSComponentTestCase.parameterize(
+ NeutronUtilsNetworkTests, os_creds=os_creds, ext_net_name=ext_net_name,
log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
- NeutronUtilsNetworkTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
+ NeutronUtilsSubnetTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ log_level=log_level))
+ suite.addTest(OSComponentTestCase.parameterize(
+ NeutronUtilsRouterTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
- NeutronUtilsSubnetTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
+ NeutronUtilsSecurityGroupTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
- NeutronUtilsRouterTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
+ NeutronUtilsFloatingIpTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
- NeutronUtilsSecurityGroupTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
+ NovaUtilsKeypairTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
- NovaUtilsKeypairTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
+ NovaUtilsFlavorTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
- NovaUtilsFlavorTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
+ NovaUtilsInstanceTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
- CreateFlavorTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
+ CreateFlavorTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
- HeatUtilsCreateStackTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level,
+ HeatUtilsCreateStackTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, log_level=log_level,
image_metadata=image_metadata))
-def add_openstack_integration_tests(suite, os_creds, ext_net_name, use_keystone=True, flavor_metadata=None,
- image_metadata=None, use_floating_ips=True, log_level=logging.INFO):
+def add_openstack_integration_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 to exercise all long-running OpenStack integration tests meaning they will be creating VM
- instances and potentially performing some SSH functions through floating IPs
+ Adds tests written to exercise all long-running OpenStack integration tests
+ meaning they will be creating VM instances and potentially performing some
+ SSH functions through floatingIPs
: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
+ :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:
+ :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 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
"""
- # Tests the OpenStack API calls via a creator. If use_keystone, objects will be created with a custom user
- # and project
+ # Tests the OpenStack API calls via a creator. If use_keystone, objects
+ # will be created with a custom user and project
# Creator Object tests
suite.addTest(OSIntegrationTestCase.parameterize(
- CreateSecurityGroupTests, 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))
+ CreateSecurityGroupTests, 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(
- CreateImageSuccessTests, 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))
+ CreateImageSuccessTests, 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(
- CreateImageNegativeTests, 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))
+ CreateImageNegativeTests, 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(
- CreateMultiPartImageTests, 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))
+ CreateMultiPartImageTests, 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(
- CreateKeypairsTests, 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))
+ CreateKeypairsTests, 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(
- CreateNetworkSuccessTests, 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))
+ CreateNetworkSuccessTests, 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(
- CreateRouterSuccessTests, 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))
+ CreateRouterSuccessTests, 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(
- CreateRouterNegativeTests, 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))
+ CreateRouterNegativeTests, 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))
# VM Instances
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))
+ 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))
suite.addTest(OSIntegrationTestCase.parameterize(
- CreateInstanceSimpleTests, 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))
+ CreateInstanceSimpleTests, 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(
- CreateInstancePortManipulationTests, 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))
+ CreateInstancePortManipulationTests, 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(
- InstanceSecurityGroupTests, 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))
+ InstanceSecurityGroupTests, 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(
- CreateInstanceOnComputeHost, 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))
+ CreateInstanceOnComputeHost, 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(
- CreateInstanceFromThreePartImage, 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))
+ CreateInstanceFromThreePartImage, 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))
+ 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))
+ 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))
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))
+ 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(
- CreateInstancePubPrivNetTests, 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))
+ CreateInstancePubPrivNetTests, 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(
- AnsibleProvisioningTests, 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))
+ AnsibleProvisioningTests, 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):
+def add_openstack_staging_tests(suite, os_creds, ext_net_name,
+ log_level=logging.INFO):
"""
- Adds tests that are still in development have not been designed to run successfully against all OpenStack pods
+ Adds tests that are still in development have not been designed to run
+ successfully against all OpenStack pods
: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 os_creds: 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 log_level: the logging level
:return: None as the tests will be adding to the 'suite' parameter object
"""
suite.addTest(OSComponentTestCase.parameterize(
- CreateNetworkTypeTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
+ CreateNetworkTypeTests, os_creds=os_creds, ext_net_name=ext_net_name,
+ log_level=log_level))
suite.addTest(OSComponentTestCase.parameterize(
- CreateInstanceMockOfflineTests, os_creds=os_creds, ext_net_name=ext_net_name, log_level=log_level))
+ CreateInstanceMockOfflineTests, os_creds=os_creds,
+ ext_net_name=ext_net_name, log_level=log_level))
Code Review / snaps.git / commitdiff