+---------------------------------------+---------------+-----------------------------------------------------------+
| Test Name | Heat API | Description |
+=======================================+===============+===========================================================+
-| test_create_stack | 1 | Tests the heat_utils.create_stack() with a test template |
+| test_create_stack | 1-3 | Tests the heat_utils.create_stack() with a test template |
+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_create_stack_x2 | 1 | Tests the heat_utils.create_stack() with a test template |
+| test_create_stack_x2 | 1-3 | Tests the heat_utils.create_stack() with a test template |
| | | and attempts to deploy a second time w/o actually |
| | | deploying any objects |
+---------------------------------------+---------------+-----------------------------------------------------------+
+---------------------------------------+---------------+-----------------------------------------------------------+
| Test Name | Heat API | Description |
+=======================================+===============+===========================================================+
-| test_get_settings_from_stack | 1 | Tests the heat_utils functions that are responsible for |
+| test_get_settings_from_stack | 1-3 | Tests the heat_utils functions that are responsible for |
| | | reverse engineering settings objects of the types deployed|
| | | by Heat |
+---------------------------------------+---------------+-----------------------------------------------------------+
+heat_utils_tests.py - HeatUtilsRouterTests
+------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | Heat API | Description |
++=======================================+===============+===========================================================+
+| test_create_router_with_stack | 1-3 | Tests ability of the function |
+| | | heat_utils.get_stack_routers() to return the correct |
+| | | OpenStackRouter instance |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
heat_utils_tests.py - HeatUtilsVolumeTests
------------------------------------------
+---------------------------------------+---------------+-----------------------------------------------------------+
| Test Name | Heat API | Description |
+=======================================+===============+===========================================================+
-| test_create_vol_with_stack | 1 | Tests ability of the function |
+| test_create_vol_with_stack | 1-3 | Tests ability of the function |
| | | heat_utils.create_stack() to return the correct |
| | | Volume domain objects deployed with Heat |
+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_create_vol_types_with_stack | 1 | Tests ability of the function |
+| test_create_vol_types_with_stack | 1-3 | Tests ability of the function |
| | | heat_utils.get_stack_volumes_types() to return the correct|
| | | VolumeType domain objects deployed with Heat |
+---------------------------------------+---------------+-----------------------------------------------------------+
+---------------------------------------+---------------+-----------------------------------------------------------+
| Test Name | Heat API | Description |
+=======================================+===============+===========================================================+
-| test_create_keypair_with_stack | 1 | Tests ability of the function |
+| test_create_keypair_with_stack | 1-3 | Tests ability of the function |
| | | heat_utils.create_stack() to return the correct |
| | | Keypair domain objects deployed with Heat |
+---------------------------------------+---------------+-----------------------------------------------------------+
+---------------------------------------+---------------+-----------------------------------------------------------+
| Test Name | Heat API | Description |
+=======================================+===============+===========================================================+
-| test_create_flavor_with_stack | 1 | Tests ability of the function |
+| test_create_flavor_with_stack | 1-3 | Tests ability of the function |
| | | heat_utils.get_stack_flavors() to return the correct |
| | | Flavor domain objects deployed with Heat |
+---------------------------------------+---------------+-----------------------------------------------------------+
| | | deploying |
+---------------------------------------+---------------+-----------------------------------------------------------+
+create_stack_tests.py - CreateStackFloatingIpTests
+--------------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | Heat API | Description |
++=======================================+===============+===========================================================+
+| test_connect_via_ssh_heat_vm | 1 | Ensures that an OpenStackHeatStack instance can create a |
+| | | VM with a floating IP that can be accessed via |
+| | | OpenStackVmInstance |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
+create_stack_tests.py - CreateStackRouterTests
+----------------------------------------------
+
++---------------------------------------+---------------+-----------------------------------------------------------+
+| Test Name | Heat API | Description |
++=======================================+===============+===========================================================+
+| test_retrieve_router_creator | 1 | Ensures that an OpenStackHeatStack instance can return a |
+| | | OpenStackRouter instance that it was responsible for |
+| | | deploying |
++---------------------------------------+---------------+-----------------------------------------------------------+
+
create_stack_tests.py - CreateStackFlavorTests
----------------------------------------------
"""
self.name = kwargs.get('name')
self.id = kwargs.get('id')
+ self.network_id = kwargs.get('network_id')
self.cidr = kwargs.get('cidr')
self.ip_version = kwargs.get('ip_version')
self.gateway_ip = kwargs.get('gateway_ip')
def __eq__(self, other):
return (self.name == other.name and
self.id == other.id and
+ self.network_id == other.network_id and
self.cidr == other.cidr and
self.ip_version == other.ip_version and
self.gateway_ip == other.gateway_ip and
Constructor
:param name: the router's name
:param id: the router's id
+ :param status: the router's status
+ :param tenant_id: the router's project/tenant ID
+ :param admin_state_up: Router is up when True
+ :param external_gateway_info: dict() for populating external_network_id
+ and external_fixed_ips
+ external_network_id: ID of the external network to route
+ in dict under key 'external_fixed_ips'
+ external_fixed_ips: List IP addresses associated with the
+ external_network_id found in dict under
+ key 'network_id'
+ :param port_subnets: list of tuples where #1 is the Port domain object
+ and #2 is a list of associated Subnet domain
+ objects
"""
self.name = kwargs.get('name')
self.id = kwargs.get('id')
self.status = kwargs.get('status')
self.tenant_id = kwargs.get('tenant_id')
self.admin_state_up = kwargs.get('admin_state_up')
- self.external_gateway_info = kwargs.get('external_gateway_info')
+ self.port_subnets = kwargs.get('port_subnets')
+
+ if (kwargs.get('external_gateway_info') and
+ isinstance(kwargs.get('external_gateway_info'), dict) and
+ kwargs.get('external_gateway_info').get('external_fixed_ips')):
+ gateway_info = kwargs.get('external_gateway_info')
+
+ self.external_network_id = gateway_info.get('network_id')
+ self.external_fixed_ips = gateway_info.get('external_fixed_ips')
+ else:
+ self.external_fixed_ips = kwargs.get('external_fixed_ips', None)
+ self.external_network_id = kwargs.get('external_network_id', None)
def __eq__(self, other):
return (self.name == other.name and self.id == other.id and
self.status == other.status and
self.tenant_id == other.tenant_id and
self.admin_state_up == other.admin_state_up and
- self.external_gateway_info == other.external_gateway_info)
+ self.external_network_id == other.external_network_id and
+ self.external_fixed_ips == other.external_fixed_ips and
+ self.port_subnets == other.port_subnets)
class InterfaceRouter:
self.assertEqual('hello', router.status)
self.assertEqual('1234', router.tenant_id)
self.assertEqual('yes', router.admin_state_up)
- self.assertEqual('no', router.external_gateway_info)
+ self.assertIsNone(router.external_fixed_ips)
+ self.assertIsNone(router.external_network_id)
def test_construction_named(self):
router = Router(
self.assertEqual('hello', router.status)
self.assertEqual('1234', router.tenant_id)
self.assertEqual('yes', router.admin_state_up)
- self.assertEqual('no', router.external_gateway_info)
+ self.assertIsNone(router.external_fixed_ips)
+ self.assertIsNone(router.external_network_id)
+
+ def test_ext_gateway_named(self):
+ ext_gateway = {'network_id': '123',
+ 'external_fixed_ips': ['456', '789']}
+ router = Router(
+ external_fixed_ips=['456', '789'], external_network_id='123',
+ admin_state_up='yes', tenant_id='1234', status='hello', id='id',
+ name='name')
+ self.assertEqual('name', router.name)
+ self.assertEqual('id', router.id)
+ self.assertEqual('hello', router.status)
+ self.assertEqual('1234', router.tenant_id)
+ self.assertEqual('yes', router.admin_state_up)
+ self.assertEqual(['456', '789'], router.external_fixed_ips)
+ self.assertEqual('123', router.external_network_id)
+
+
+ def test_ext_net_ips_named(self):
+ ext_gateway = {'network_id': '123',
+ 'external_fixed_ips': ['456', '789']}
+ router = Router(
+ external_gateway_info=ext_gateway, admin_state_up='yes',
+ tenant_id='1234', status='hello', id='id', name='name')
+ self.assertEqual('name', router.name)
+ self.assertEqual('id', router.id)
+ self.assertEqual('hello', router.status)
+ self.assertEqual('1234', router.tenant_id)
+ self.assertEqual('yes', router.admin_state_up)
+ self.assertEqual(['456', '789'], router.external_fixed_ips)
+ self.assertEqual('123', router.external_network_id)
class InterfaceRouterDomainObjectTests(unittest.TestCase):
"""
router = neutron_utils.get_router(
self.__neutron, router_name=router_name)
- if router and router.external_gateway_info:
+ if router and router.external_network_id:
network = neutron_utils.get_network_by_id(
- self.__neutron,
- router.external_gateway_info['network_id'])
+ self.__neutron, router.external_network_id)
if network:
return network.name
return None
'subnet_name' and 'ip' values which will get mapped to
self.fixed_ips. These values will be directly
translated into the fixed_ips dict (optional)
- :param fixed_ips: A dict where the key is the subnet IDs and value is
- the IP address to assign to the port (optional and
- recommended to configure via ip_addrs instead)
:param security_groups: One or more security group IDs.
:param allowed_address_pairs: A dictionary containing a set of zero or
more allowed address pairs. An address
if 'port' in kwargs:
kwargs = kwargs['port']
- self.network = None
-
self.name = kwargs.get('name')
self.network_name = kwargs.get('network_name')
self.project_name = kwargs.get('project_name')
self.mac_address = kwargs.get('mac_address')
self.ip_addrs = kwargs.get('ip_addrs')
- self.fixed_ips = kwargs.get('fixed_ips')
self.security_groups = kwargs.get('security_groups')
self.allowed_address_pairs = kwargs.get('allowed_address_pairs')
self.opt_value = kwargs.get('opt_value')
raise PortSettingsError(
'The attribute network_name is required')
- def __set_fixed_ips(self, neutron):
+ def __get_fixed_ips(self, neutron):
"""
Sets the self.fixed_ips value
:param neutron: the Neutron client
:return: None
"""
- if not self.fixed_ips and self.ip_addrs:
- self.fixed_ips = list()
+
+ fixed_ips = list()
+ if self.ip_addrs:
for ip_addr_dict in self.ip_addrs:
subnet = neutron_utils.get_subnet(
neutron, subnet_name=ip_addr_dict['subnet_name'])
if subnet and 'ip' in ip_addr_dict:
- self.fixed_ips.append({'ip_address': ip_addr_dict['ip'],
- 'subnet_id': subnet.id})
+ fixed_ips.append({'ip_address': ip_addr_dict['ip'],
+ 'subnet_id': subnet.id})
else:
raise PortSettingsError(
'Invalid port configuration, subnet does not exist '
'with name - ' + ip_addr_dict['subnet_name'])
+ return fixed_ips
+
def dict_for_neutron(self, neutron, os_creds):
"""
Returns a dictionary object representing this object.
:param os_creds: the OpenStack credentials
:return: the dictionary object
"""
- self.__set_fixed_ips(neutron)
out = dict()
if project:
project_id = project.id
- if not self.network:
- self.network = neutron_utils.get_network(
- neutron, network_name=self.network_name, project_id=project_id)
- if not self.network:
+ network = neutron_utils.get_network(
+ neutron, network_name=self.network_name, project_id=project_id)
+ if not network:
raise PortSettingsError(
'Cannot locate network with name - ' + self.network_name)
- out['network_id'] = self.network.id
+ out['network_id'] = network.id
if self.admin_state_up is not None:
out['admin_state_up'] = self.admin_state_up
self.project_name)
if self.mac_address:
out['mac_address'] = self.mac_address
- if self.fixed_ips and len(self.fixed_ips) > 0:
- out['fixed_ips'] = self.fixed_ips
+
+ fixed_ips = self.__get_fixed_ips(neutron)
+ if fixed_ips and len(fixed_ips) > 0:
+ out['fixed_ips'] = fixed_ips
+
if self.security_groups:
out['security_groups'] = self.security_groups
if self.allowed_address_pairs and len(self.allowed_address_pairs) > 0:
self.project_name == other.project_name and
self.mac_address == other.mac_address and
self.ip_addrs == other.ip_addrs and
- self.fixed_ips == other.fixed_ips and
+ # self.fixed_ips == other.fixed_ips and
self.security_groups == other.security_groups and
self.allowed_address_pairs == other.allowed_address_pairs and
self.opt_value == other.opt_value and
'Error creating port with name - '
+ port_setting.name)
+ self.__router = neutron_utils.get_router_by_id(
+ self._neutron, self.__router.id)
return self.__router
def clean(self):
:param external_gateway: Name of the external network to which to route
:param admin_state_up: The administrative status of the router.
True = up / False = down (default True)
- :param external_fixed_ips: Dictionary containing the IP address
- parameters.
:param internal_subnets: List of subnet names to which to connect this
router for Floating IP purposes
:param port_settings: List of PortSettings objects
self.project_name = kwargs.get('project_name')
self.external_gateway = kwargs.get('external_gateway')
- self.admin_state_up = kwargs.get('admin_state_up')
+ self.admin_state_up = kwargs.get('admin_state_up', True)
self.enable_snat = kwargs.get('enable_snat')
- self.external_fixed_ips = kwargs.get('external_fixed_ips')
if kwargs.get('internal_subnets'):
self.internal_subnets = kwargs['internal_subnets']
else:
from snaps.openstack.create_flavor import OpenStackFlavor
from snaps.openstack.create_instance import OpenStackVmInstance
from snaps.openstack.create_keypairs import OpenStackKeypair
+from snaps.openstack.create_router import OpenStackRouter
from snaps.openstack.create_volume import OpenStackVolume
from snaps.openstack.create_volume_type import OpenStackVolumeType
from snaps.openstack.openstack_creator import OpenStackCloudObject
return out
+ def get_router_creators(self):
+ """
+ Returns a list of router creator objects as configured by the heat
+ template
+ :return: list() of OpenStackRouter objects
+ """
+
+ neutron = neutron_utils.neutron_client(self._os_creds)
+
+ out = list()
+ stack_routers = heat_utils.get_stack_routers(
+ self.__heat_cli, neutron, self.__stack)
+
+ for routers in stack_routers:
+ settings = settings_utils.create_router_settings(
+ neutron, routers)
+ creator = OpenStackRouter(self._os_creds, settings)
+ out.append(creator)
+ creator.initialize()
+
+ return out
+
def get_vm_inst_creators(self, heat_keypair_option=None):
"""
Returns a list of VM Instance creator objects as configured by the heat
self.assertIsNone(settings.project_name)
self.assertIsNone(settings.mac_address)
self.assertIsNone(settings.ip_addrs)
- self.assertIsNone(settings.fixed_ips)
self.assertIsNone(settings.security_groups)
self.assertIsNone(settings.allowed_address_pairs)
self.assertIsNone(settings.opt_value)
self.assertIsNone(settings.project_name)
self.assertIsNone(settings.mac_address)
self.assertIsNone(settings.ip_addrs)
- self.assertIsNone(settings.fixed_ips)
self.assertIsNone(settings.security_groups)
self.assertIsNone(settings.allowed_address_pairs)
self.assertIsNone(settings.opt_value)
def test_all(self):
ip_addrs = [{'subnet_name', 'foo-sub', 'ip', '10.0.0.10'}]
- fixed_ips = {'sub_id', '10.0.0.10'}
allowed_address_pairs = {'10.0.0.101', '1234.5678'}
settings = PortSettings(name='foo', network_name='bar',
admin_state_up=False,
project_name='foo-project',
mac_address='1234', ip_addrs=ip_addrs,
- fixed_ips=fixed_ips,
security_groups=['foo_grp_id'],
allowed_address_pairs=allowed_address_pairs,
opt_value='opt value', opt_name='opt name',
self.assertEqual('foo-project', settings.project_name)
self.assertEqual('1234', settings.mac_address)
self.assertEqual(ip_addrs, settings.ip_addrs)
- self.assertEqual(fixed_ips, settings.fixed_ips)
self.assertEqual(1, len(settings.security_groups))
self.assertEqual('foo_grp_id', settings.security_groups[0])
self.assertEqual(allowed_address_pairs, settings.allowed_address_pairs)
def test_config_all(self):
ip_addrs = [{'subnet_name', 'foo-sub', 'ip', '10.0.0.10'}]
- fixed_ips = {'sub_id', '10.0.0.10'}
allowed_address_pairs = {'10.0.0.101', '1234.5678'}
settings = PortSettings(
**{'name': 'foo', 'network_name': 'bar', 'admin_state_up': False,
'project_name': 'foo-project', 'mac_address': '1234',
- 'ip_addrs': ip_addrs,
- 'fixed_ips': fixed_ips, 'security_groups': ['foo_grp_id'],
+ 'ip_addrs': ip_addrs, 'security_groups': ['foo_grp_id'],
'allowed_address_pairs': allowed_address_pairs,
- 'opt_value': 'opt value',
- 'opt_name': 'opt name', 'device_owner': 'owner',
- 'device_id': 'device number'})
+ 'opt_value': 'opt value', 'opt_name': 'opt name',
+ 'device_owner': 'owner', 'device_id': 'device number'})
self.assertEqual('foo', settings.name)
self.assertEqual('bar', settings.network_name)
self.assertFalse(settings.admin_state_up)
self.assertEqual('foo-project', settings.project_name)
self.assertEqual('1234', settings.mac_address)
self.assertEqual(ip_addrs, settings.ip_addrs)
- self.assertEqual(fixed_ips, settings.fixed_ips)
self.assertEqual(1, len(settings.security_groups))
self.assertEqual('foo_grp_id', settings.security_groups[0])
self.assertEqual(allowed_address_pairs, settings.allowed_address_pairs)
from snaps.openstack.create_router import (
RouterSettings, RouterSettingsError)
from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
-from snaps.openstack.utils import neutron_utils
+from snaps.openstack.utils import neutron_utils, settings_utils
__author__ = 'mmakati'
self.assertEqual('foo', settings.name)
self.assertIsNone(settings.project_name)
self.assertIsNone(settings.external_gateway)
- self.assertIsNone(settings.admin_state_up)
+ self.assertTrue(settings.admin_state_up)
self.assertIsNone(settings.enable_snat)
- self.assertIsNone(settings.external_fixed_ips)
self.assertIsNotNone(settings.internal_subnets)
self.assertTrue(isinstance(settings.internal_subnets, list))
self.assertEqual(0, len(settings.internal_subnets))
self.assertEqual('foo', settings.name)
self.assertIsNone(settings.project_name)
self.assertIsNone(settings.external_gateway)
- self.assertIsNone(settings.admin_state_up)
+ self.assertTrue(settings.admin_state_up)
self.assertIsNone(settings.enable_snat)
- self.assertIsNone(settings.external_fixed_ips)
self.assertIsNotNone(settings.internal_subnets)
self.assertTrue(isinstance(settings.internal_subnets, list))
self.assertEqual(0, len(settings.internal_subnets))
port_settings = PortSettings(name='foo', network_name='bar')
settings = RouterSettings(
name='foo', project_name='bar', external_gateway='foo_gateway',
- admin_state_up=True, enable_snat=False, external_fixed_ips=['ip1'],
+ admin_state_up=True, enable_snat=False,
internal_subnets=['10.0.0.1/24'], interfaces=[port_settings])
self.assertEqual('foo', settings.name)
self.assertEqual('bar', settings.project_name)
self.assertEqual('foo_gateway', settings.external_gateway)
self.assertTrue(settings.admin_state_up)
self.assertFalse(settings.enable_snat)
- self.assertEqual(['ip1'], settings.external_fixed_ips)
self.assertIsNotNone(settings.internal_subnets)
self.assertTrue(isinstance(settings.internal_subnets, list))
self.assertEqual(1, len(settings.internal_subnets))
settings = RouterSettings(
**{'name': 'foo', 'project_name': 'bar',
'external_gateway': 'foo_gateway', 'admin_state_up': True,
- 'enable_snat': False, 'external_fixed_ips': ['ip1'],
- 'internal_subnets': ['10.0.0.1/24'],
+ 'enable_snat': False, 'internal_subnets': ['10.0.0.1/24'],
'interfaces':
[{'port': {'name': 'foo-port',
'network_name': 'bar-net'}}]})
self.assertEqual('foo_gateway', settings.external_gateway)
self.assertTrue(settings.admin_state_up)
self.assertFalse(settings.enable_snat)
- self.assertEqual(['ip1'], settings.external_fixed_ips)
self.assertIsNotNone(settings.internal_subnets)
self.assertTrue(isinstance(settings.internal_subnets, list))
self.assertEqual(1, len(settings.internal_subnets))
router_settings=router_settings)
self.assertIsNotNone(router)
- self.assertTrue(verify_router_attributes(
- router, self.router_creator, ext_gateway=self.ext_net_name))
+ self.assertEqual(self.router_creator.get_router(), router)
+
+ self.check_router_recreation(router, router_settings)
def test_create_router_admin_user_to_new_project(self):
"""
router_settings=router_settings)
self.assertIsNotNone(router)
- self.assertTrue(verify_router_attributes(
- router, self.router_creator, ext_gateway=self.ext_net_name))
+ self.assertEqual(self.router_creator.get_router(), router)
+
+ self.check_router_recreation(router, router_settings)
def test_create_router_new_user_to_admin_project(self):
"""
router_settings=router_settings)
self.assertIsNotNone(router)
- self.assertTrue(verify_router_attributes(
- router, self.router_creator, ext_gateway=self.ext_net_name))
+ self.assertEqual(self.router_creator.get_router(), router)
+
+ self.check_router_recreation(router, router_settings)
def test_create_delete_router(self):
"""
router_settings=router_settings)
self.assertIsNotNone(router)
- self.assertTrue(verify_router_attributes(router, self.router_creator,
- admin_state=False))
+ self.assertEqual(self.router_creator.get_router(), router)
+
+ self.check_router_recreation(router, router_settings)
def test_create_router_admin_state_True(self):
"""
Test creation of a basic router with admin state Up.
"""
- router_settings = RouterSettings(name=self.guid + '-pub-router',
- admin_state_up=True)
+ router_settings = RouterSettings(
+ name=self.guid + '-pub-router', admin_state_up=True)
- self.router_creator = create_router.OpenStackRouter(self.os_creds,
- router_settings)
+ self.router_creator = create_router.OpenStackRouter(
+ self.os_creds, router_settings)
self.router_creator.create()
- router = neutron_utils.get_router(self.neutron,
- router_settings=router_settings)
+ router = neutron_utils.get_router(
+ self.neutron, router_settings=router_settings)
self.assertIsNotNone(router)
- self.assertTrue(verify_router_attributes(router, self.router_creator,
- admin_state=True))
+ self.assertEqual(self.router_creator.get_router(), router)
+
+ self.check_router_recreation(router, router_settings)
def test_create_router_private_network(self):
"""
router_settings)
self.router_creator.create()
- router = neutron_utils.get_router(self.neutron,
- router_settings=router_settings)
+ router = neutron_utils.get_router(
+ self.neutron, router_settings=router_settings)
- self.assertTrue(verify_router_attributes(router, self.router_creator))
+ self.assertEqual(router, self.router_creator.get_router())
# Instantiate second identical creator to ensure a second router
# has not been created
router2 = router_creator2.create()
self.assertIsNotNone(self.router_creator.get_router(), router2)
+ self.check_router_recreation(router2, router_settings)
+
def test_create_router_external_network(self):
"""
Test creation of a router connected to an external network and a
router_settings = RouterSettings(
name=self.guid + '-pub-router', external_gateway=self.ext_net_name,
port_settings=port_settings)
- self.router_creator = create_router.OpenStackRouter(self.os_creds,
- router_settings)
+ self.router_creator = create_router.OpenStackRouter(
+ self.os_creds, router_settings)
self.router_creator.create()
- router = neutron_utils.get_router(self.neutron,
- router_settings=router_settings)
+ router = neutron_utils.get_router(
+ self.neutron, router_settings=router_settings)
+
+ self.assertEquals(router, self.router_creator.get_router())
- self.assertTrue(verify_router_attributes(
- router, self.router_creator, ext_gateway=self.ext_net_name))
+ self.check_router_recreation(router, router_settings)
+
+ def check_router_recreation(self, router, orig_settings):
+ """
+ Validates the derived RouterSettings with the original
+ :param router: the Router domain object to test
+ :param orig_settings: the original RouterSettings object that was
+ responsible for creating the router
+ :return: the derived RouterSettings object
+ """
+ derived_settings = settings_utils.create_router_settings(
+ self.neutron, router)
+ self.assertIsNotNone(derived_settings)
+ self.assertEqual(
+ orig_settings.enable_snat, derived_settings.enable_snat)
+ self.assertEqual(orig_settings.external_gateway,
+ derived_settings.external_gateway)
+ self.assertEqual(orig_settings.name, derived_settings.name)
+ self.assertEqual(orig_settings.internal_subnets,
+ derived_settings.internal_subnets)
+
+ if orig_settings.external_gateway:
+ self.assertEqual(len(orig_settings.port_settings),
+ len(derived_settings.port_settings))
+ else:
+ self.assertEqual(len(orig_settings.port_settings),
+ len(derived_settings.port_settings))
+
+ if len(orig_settings.port_settings) > 0:
+ self.assertEqual(orig_settings.port_settings[0].name,
+ derived_settings.port_settings[0].name)
+
+ if len(orig_settings.port_settings) > 1:
+ self.assertEqual(orig_settings.port_settings[1].name,
+ derived_settings.port_settings[1].name)
+
+ return derived_settings
class CreateRouterNegativeTests(OSIntegrationTestCase):
self.router_creator = create_router.OpenStackRouter(
self.os_creds, router_settings)
self.router_creator.create()
-
-
-def verify_router_attributes(router_operational, router_creator,
- admin_state=True, ext_gateway=None):
- """
- Helper function to validate the attributes of router created with the one
- operational
- :param router_operational: Operational Router object returned from neutron
- utils of type snaps.domain.Router
- :param router_creator: router_creator object returned from creating a
- router in the router test functions
- :param admin_state: True if router is expected to be Up, else False
- :param ext_gateway: None if router is not connected to external gateway
- :return:
- """
-
- router = router_creator.get_router()
-
- if not router_operational:
- return False
- elif not router_creator:
- return False
- elif not (router_operational.name == router_creator.router_settings.name):
- return False
- elif not (router_operational.id == router.id):
- return False
- elif not (router_operational.status == router.status):
- return False
- elif not (router_operational.tenant_id == router.tenant_id):
- return False
- elif not (admin_state == router_operational.admin_state_up):
- return False
- elif (ext_gateway is None) and \
- (router_operational.external_gateway_info is not None):
- return False
- elif ext_gateway is not None:
- if router_operational.external_gateway_info is None:
- return False
- return True
self.assertEqual(0, len(vm_settings.floating_ip_settings))
+class CreateStackRouterTests(OSIntegrationTestCase):
+ """
+ Tests for the CreateStack class defined in create_stack.py where the
+ target is a Network, Subnet, and Router
+ """
+
+ def setUp(self):
+ """
+ Instantiates the CreateStack object that is responsible for downloading
+ and creating an OS stack file within OpenStack
+ """
+ super(self.__class__, self).__start__()
+
+ self.guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+
+ self.heat_creds = self.admin_os_creds
+ self.heat_creds.project_name = self.admin_os_creds.project_name
+
+ self.heat_cli = heat_utils.heat_client(self.heat_creds)
+ self.neutron = neutron_utils.neutron_client(self.os_creds)
+ self.stack_creator = None
+
+ self.net_name = self.guid + '-net'
+ self.subnet_name = self.guid + '-subnet'
+ self.router_name = self.guid + '-router'
+
+ self.env_values = {
+ 'net_name': self.net_name,
+ 'subnet_name': self.subnet_name,
+ 'router_name': self.router_name,
+ 'external_net_name': self.ext_net_name}
+
+ self.heat_tmplt_path = pkg_resources.resource_filename(
+ 'snaps.openstack.tests.heat', 'router_heat_template.yaml')
+
+ stack_settings = StackSettings(
+ name=self.__class__.__name__ + '-' + str(self.guid) + '-stack',
+ template_path=self.heat_tmplt_path,
+ env_values=self.env_values)
+ self.stack_creator = create_stack.OpenStackHeatStack(
+ self.heat_creds, stack_settings)
+ self.created_stack = self.stack_creator.create()
+ self.assertIsNotNone(self.created_stack)
+
+ def tearDown(self):
+ """
+ Cleans the stack and downloaded stack file
+ """
+ if self.stack_creator:
+ try:
+ self.stack_creator.clean()
+ except:
+ pass
+
+ super(self.__class__, self).__clean__()
+
+ def test_retrieve_router_creator(self):
+ """
+ Tests the creation of an OpenStack stack from Heat template file and
+ the retrieval of an OpenStackRouter creator/state machine instance
+ """
+ router_creators = self.stack_creator.get_router_creators()
+ self.assertEqual(1, len(router_creators))
+
+ creator = router_creators[0]
+ self.assertEqual(self.router_name, creator.router_settings.name)
+
+ router = creator.get_router()
+
+ ext_net = neutron_utils.get_network(
+ self.neutron, network_name=self.ext_net_name)
+ self.assertEqual(ext_net.id, router.external_network_id)
+
+
class CreateStackVolumeTests(OSIntegrationTestCase):
"""
Tests for the CreateStack class as they pertain to volumes
--- /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.
+##############################################################################
+heat_template_version: 2015-04-30
+
+description: >
+ Sample template with two VMs instantiated against different images and
+ flavors on the same network and the first one has a floating IP
+
+parameters:
+ net_name:
+ type: string
+ label: Test network name
+ description: The name of the stack's network
+ default: test_net
+ subnet_name:
+ type: string
+ label: Test subnet name
+ description: The name of the stack's subnet
+ default: test_subnet
+ router_name:
+ type: string
+ label: Test router name
+ description: The name of the stack's router
+ default: mgmt_router
+ external_net_name:
+ type: string
+ description: Name of the external network which management network will connect to
+ default: external
+
+resources:
+ network:
+ type: OS::Neutron::Net
+ properties:
+ name: { get_param: net_name }
+
+ subnet:
+ type: OS::Neutron::Subnet
+ properties:
+ name: { get_param: subnet_name }
+ ip_version: 4
+ cidr: 10.1.2.0/24
+ network: { get_resource: network }
+
+ management_router:
+ type: OS::Neutron::Router
+ properties:
+ name: { get_param: router_name }
+ external_gateway_info:
+ network: { get_param: external_net_name }
+
+ management_router_interface:
+ type: OS::Neutron::RouterInterface
+ properties:
+ router: { get_resource: management_router }
+ subnet: { get_resource: subnet }
out = list()
resources = get_resources(heat_cli, stack, 'OS::Neutron::Net')
for resource in resources:
- network = neutron_utils.get_network_by_id(
- neutron, resource.id)
+ network = neutron_utils.get_network_by_id(neutron, resource.id)
if network:
out.append(network)
return out
+def get_stack_routers(heat_cli, neutron, stack):
+ """
+ Returns a list of Network domain objects deployed by this stack
+ :param heat_cli: the OpenStack heat client object
+ :param neutron: the OpenStack neutron client object
+ :param stack: the SNAPS-OO Stack domain object
+ :return: a list of Network objects
+ """
+
+ out = list()
+ resources = get_resources(heat_cli, stack, 'OS::Neutron::Router')
+ for resource in resources:
+ router = neutron_utils.get_router_by_id(neutron, resource.id)
+ if router:
+ out.append(router)
+
+ return out
+
+
def get_stack_servers(heat_cli, nova, stack):
"""
Returns a list of VMInst domain objects associated with a Stack
json_body = router_settings.dict_for_neutron(neutron, os_creds)
logger.info('Creating router with name - ' + router_settings.name)
os_router = neutron.create_router(json_body)
- return Router(**os_router['router'])
+ return __map_router(neutron, os_router['router'])
else:
logger.error("Failed to create router.")
raise NeutronException('Failed to create router')
"""
router = neutron.show_router(router_id)
if router:
- return Router(**router['router'])
+ return __map_router(neutron, router['router'])
def get_router(neutron, router_settings=None, router_name=None):
return None
routers = neutron.list_routers(**router_filter)
+
for routerInst in routers['routers']:
- return Router(**routerInst)
+ return __map_router(neutron, routerInst)
+
return None
+def __map_router(neutron, os_router):
+ """
+ Takes an OpenStack router instance and maps it to a SNAPS Router domain
+ object
+ :param neutron: the neutron client
+ :param os_router: the OpenStack Router object
+ :return:
+ """
+ device_ports = neutron.list_ports(
+ **{'device_id': os_router['id']})['ports']
+ port_subnets = list()
+
+ # Order by create date
+ sorted_ports = sorted(device_ports, key=lambda dev_port: dev_port['created_at'])
+
+ for port in sorted_ports:
+ subnets = list()
+ for fixed_ip in port['fixed_ips']:
+ subnet = get_subnet_by_id(neutron, fixed_ip['subnet_id'])
+ if subnet and subnet.network_id == port['network_id']:
+ subnets.append(subnet)
+ port_subnets.append((Port(**port), subnets))
+
+ os_router['port_subnets'] = port_subnets
+ return Router(**os_router)
+
+
def add_interface_router(neutron, router, subnet=None, port=None):
"""
Adds an interface router for OpenStack for either a subnet or port.
from snaps.openstack.create_keypairs import KeypairSettings
from snaps.openstack.create_network import (
PortSettings, SubnetSettings, NetworkSettings)
+from snaps.openstack.create_router import RouterSettings
from snaps.openstack.create_volume import VolumeSettings
from snaps.openstack.create_volume_type import (
VolumeTypeSettings, VolumeTypeEncryptionSettings, ControlLocation)
return out
+def create_router_settings(neutron, router):
+ """
+ Returns a RouterSettings object
+ :param neutron: the neutron client
+ :param router: a SNAPS-OO Router domain object
+ :return:
+ """
+ ext_net_name = None
+
+ if router.external_network_id:
+ network = neutron_utils.get_network_by_id(
+ neutron, router.external_network_id)
+ if network:
+ ext_net_name = network.name
+
+ ports_tuple_list = list()
+ if router.port_subnets:
+ for port, subnets in router.port_subnets:
+ network = neutron_utils.get_network_by_id(
+ neutron, port.network_id)
+
+ ip_addrs = list()
+ if network and router.external_fixed_ips:
+ for ext_fixed_ips in router.external_fixed_ips:
+ for subnet in subnets:
+ if ext_fixed_ips['subnet_id'] == subnet.id:
+ ip_addrs.append(ext_fixed_ips['ip_address'])
+ else:
+ for ip in port.ips:
+ ip_addrs.append(ip)
+
+ ip_list = list()
+ if len(ip_addrs) > 0:
+ for ip_addr in ip_addrs:
+ if isinstance(ip_addr, dict):
+ ip_list.append(ip_addr['ip_address'])
+ else:
+ ip_list.append(ip_addr)
+
+ ports_tuple_list.append((network, ip_list))
+
+ port_settings = __create_port_settings(neutron, ports_tuple_list)
+
+ filtered_settings = list()
+ for port_setting in port_settings:
+ if port_setting.network_name != ext_net_name:
+ filtered_settings.append(port_setting)
+
+ return RouterSettings(
+ name=router.name, external_gateway=ext_net_name,
+ admin_state_up=router.admin_state_up,
+ port_settings=filtered_settings)
+
+
def create_volume_settings(volume):
"""
Returns a VolumeSettings object
kwargs = dict()
kwargs['name'] = server.name
kwargs['flavor'] = flavor_name
+
+ net_tuples = list()
+ for net_name, ips in server.networks.items():
+ network = neutron_utils.get_network(neutron, network_name=net_name)
+ if network:
+ net_tuples.append((network, ips))
+
kwargs['port_settings'] = __create_port_settings(
- neutron, server.networks)
+ neutron, net_tuples)
kwargs['security_group_names'] = server.sec_grp_names
kwargs['floating_ip_settings'] = __create_floatingip_settings(
neutron, kwargs['port_settings'])
"""
Returns a list of port settings based on the networks parameter
:param neutron: the neutron client
- :param networks: a dict where the key is the network name and the value
- is a list of IP addresses
+ :param networks: a list of tuples where #1 is the SNAPS Network domain
+ object and #2 is a list of IP addresses
:return:
"""
out = list()
- for net_name, ips in networks.items():
- network = neutron_utils.get_network(neutron, network_name=net_name)
+ for network, ips in networks:
ports = neutron_utils.get_ports(neutron, network, ips)
for port in ports:
- kwargs = dict()
- if port.name:
- kwargs['name'] = port.name
- kwargs['network_name'] = network.name
- kwargs['mac_address'] = port.mac_address
- kwargs['allowed_address_pairs'] = port.allowed_address_pairs
- kwargs['admin_state_up'] = port.admin_state_up
- out.append(PortSettings(**kwargs))
+ if port.device_owner != 'network:dhcp':
+ ip_addrs = list()
+ for ip_dict in port.ips:
+ subnet = neutron_utils.get_subnet_by_id(
+ neutron, ip_dict['subnet_id'])
+ ip_addrs.append({'subnet_name': subnet.name,
+ 'ip': ip_dict['ip_address']})
+
+ kwargs = dict()
+ if port.name:
+ kwargs['name'] = port.name
+ kwargs['network_name'] = network.name
+ kwargs['mac_address'] = port.mac_address
+ kwargs['allowed_address_pairs'] = port.allowed_address_pairs
+ kwargs['admin_state_up'] = port.admin_state_up
+ kwargs['ip_addrs'] = ip_addrs
+ out.append(PortSettings(**kwargs))
return out
self.assertEqual(self.keypair_name, keypair2_settings.name)
+class HeatUtilsRouterTests(OSComponentTestCase):
+ """
+ Test Heat volume functionality
+ """
+
+ def setUp(self):
+ """
+ Instantiates OpenStack instances that cannot be spawned by Heat
+ """
+ guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
+ stack_name = guid + '-stack'
+
+ self.net_name = guid + '-net'
+ self.subnet_name = guid + '-subnet'
+ self.router_name = guid + '-router'
+
+ env_values = {
+ 'net_name': self.net_name,
+ 'subnet_name': self.subnet_name,
+ 'router_name': self.router_name,
+ 'external_net_name': self.ext_net_name}
+
+ heat_tmplt_path = pkg_resources.resource_filename(
+ 'snaps.openstack.tests.heat', 'router_heat_template.yaml')
+ self.stack_settings = StackSettings(
+ name=stack_name, template_path=heat_tmplt_path,
+ env_values=env_values)
+ self.stack = None
+ self.heat_client = heat_utils.heat_client(self.os_creds)
+ self.neutron = neutron_utils.neutron_client(self.os_creds)
+
+ def tearDown(self):
+ """
+ Cleans the image and downloaded image file
+ """
+ if self.stack:
+ try:
+ heat_utils.delete_stack(self.heat_client, self.stack)
+ except:
+ pass
+
+ def test_create_router_with_stack(self):
+ """
+ Tests the creation of an OpenStack router with Heat and the retrieval
+ of the Router Domain objects from heat_utils#get_stack_routers().
+ """
+ self.stack = heat_utils.create_stack(
+ self.heat_client, self.stack_settings)
+
+ # Wait until stack deployment has completed
+ end_time = time.time() + create_stack.STACK_COMPLETE_TIMEOUT
+ is_active = False
+ while time.time() < end_time:
+ status = heat_utils.get_stack_status(self.heat_client,
+ self.stack.id)
+ if status == create_stack.STATUS_CREATE_COMPLETE:
+ is_active = True
+ break
+ elif status == create_stack.STATUS_CREATE_FAILED:
+ is_active = False
+ break
+
+ time.sleep(3)
+
+ self.assertTrue(is_active)
+
+ routers = heat_utils.get_stack_routers(
+ self.heat_client, self.neutron, self.stack)
+
+ self.assertEqual(1, len(routers))
+
+ router = routers[0]
+ self.assertEqual(self.router_name, router.name)
+
+ ext_net = neutron_utils.get_network(
+ self.neutron, network_name=self.ext_net_name)
+ self.assertEqual(ext_net.id, router.external_network_id)
+
+
class HeatUtilsVolumeTests(OSComponentTestCase):
"""
Test Heat volume functionality
ext_net = neutron_utils.get_network(
self.neutron, network_name=self.ext_net_name)
self.assertEqual(
- self.router.external_gateway_info['network_id'], ext_net.id)
+ self.router.external_network_id, ext_net.id)
def test_create_router_empty_name(self):
"""
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)
self.glance = glance_utils.glance_client(self.os_creds)
self.neutron = neutron_utils.neutron_client(self.os_creds)
Code Review / snaps.git / commitdiff