name: mgr-router
external_gateway: external
internal_subnets:
- - mgr-subnet
- - mgr-subnet-2
+ - subnet:
+ project_name: admin
+ network_name: mgr-net
+ subnet_name: mgr-subnet
+ - subnet:
+ project_name: admin
+ network_name: mgr-net
+ subnet_name: mgr-subnet-2
interfaces:
- port:
name: mgr-router-to-site1
name: site1-router
external_gateway: external
internal_subnets:
- - site1-subnet
+ - subnet:
+ project_name: admin
+ network_name: site1-net
+ subnet_name: site1-subnet
- router:
name: site2-router
external_gateway: external
internal_subnets:
- - site2-subnet
+ - subnet:
+ project_name: admin
+ network_name: site2-net
+ subnet_name: site2-subnet
- router:
name: site-to-site-router
interfaces:
name: ext-net-router
external_gateway: ext-net
internal_subnets:
- - internal-subnet
+ - subnet:
+ project_name: admin
+ network_name: internal-net
+ subnet_name: internal-subnet
keypairs:
- keypair:
name: ext-net-kp
name: {{ router_name }}
external_gateway: {{ ext_net }}
internal_subnets:
- - {{ subnet_name }}
+ - subnet:
+ project_name: {{ admin_proj }}
+ network_name: {{ net_name }}
+ subnet_name: {{ subnet_name }}
keypairs:
- keypair:
os_user:
name: simple-router
external_gateway: external
internal_subnets:
- - simple-subnet
+ - subnet:
+ project_name: admin
+ network_name: simple-net
+ subnet_name: simple-subnet
keypairs:
- keypair:
name: simple-kp
name: router-1
external_gateway: external
internal_subnets:
- - subnet-1
+ - subnet:
+ project_name: admin
+ network_name: net-1
+ subnet_name: subnet-1
keypairs:
- keypair:
name: two-net
name: router-1
external_gateway: external
internal_subnets:
- - subnet-1
+ - subnet:
+ project_name: admin
+ network_name: net-1
+ subnet_name: subnet-1
keypairs:
- keypair:
name: simple
raise PortConfigError(
'The attribute network_name is required')
- def __get_fixed_ips(self, neutron):
+ def __get_fixed_ips(self, neutron, network):
"""
Sets the self.fixed_ips value
:param neutron: the Neutron client
+ :param network: the SNAPS-OO network domain object
:return: None
"""
-
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:
- fixed_ips.append({'ip_address': ip_addr_dict['ip'],
- 'subnet_id': subnet.id})
+ neutron, network, subnet_name=ip_addr_dict['subnet_name'])
+ if subnet:
+ if 'ip' in ip_addr_dict:
+ fixed_ips.append({'ip_address': ip_addr_dict['ip'],
+ 'subnet_id': subnet.id})
else:
raise PortConfigError(
'Invalid port configuration, subnet does not exist '
TODO - expand automated testing to exercise all parameters
:param neutron: the Neutron client
+ :param keystone: the Keystone client
:param os_creds: the OpenStack credentials
:return: the dictionary object
"""
-
out = dict()
-
session = keystone_utils.keystone_session(os_creds)
keystone = keystone_utils.keystone_client(os_creds, session)
+
+ project_name = os_creds.project_name
+ if self.project_name:
+ project_name = project_name
try:
network = neutron_utils.get_network(
neutron, keystone, network_name=self.network_name,
- project_name=self.project_name)
+ project_name=project_name)
finally:
- keystone_utils.close_session(session)
+ if session:
+ keystone_utils.close_session(session)
if not network:
raise PortConfigError(
'Cannot locate network with name - ' + self.network_name
- + ' in project - ' + str(self.project_name))
+ + ' in project - ' + str(project_name))
out['network_id'] = network.id
if self.mac_address:
out['mac_address'] = self.mac_address
- fixed_ips = self.__get_fixed_ips(neutron)
+ fixed_ips = self.__get_fixed_ips(neutron, network)
if fixed_ips and len(fixed_ips) > 0:
out['fixed_ips'] = fixed_ips
:param admin_state_up: The administrative status of the router.
True = up / False = down (default True)
:param internal_subnets: List of subnet names to which to connect this
- router for Floating IP purposes
+ router (this way is deprecated).
+ *** NEW WAY ***
+ List of dict where the key is 'subnet' that
+ contains members with the following keys:
+ project_name, network_name, and subnet_name
:param port_settings: List of PortConfig objects
:return:
"""
self.enable_snat = kwargs.get('enable_snat')
if kwargs.get('internal_subnets'):
self.internal_subnets = kwargs['internal_subnets']
+ if isinstance(self.internal_subnets, dict):
+ if 'subnet' not in self.internal_subnets:
+ raise RouterConfigError(
+ 'subnet is a required key to internal_subnets')
+ if 'project_name' not in self.internal_subnets['subnet']:
+ raise RouterConfigError(
+ 'subnet.project is a required key to subnet')
+ if 'network_name' not in self.internal_subnets['subnet']:
+ raise RouterConfigError(
+ 'network_name is a required key to subnet')
+ if 'subnet_name' not in self.internal_subnets['subnet']:
+ raise RouterConfigError(
+ 'subnet_name is a required key to subnet')
else:
self.internal_subnets = list()
with self.assertRaises(RouterConfigError):
RouterConfig(**dict())
+ def test_bad_internal_subnets_bad_key(self):
+ with self.assertRaises(RouterConfigError):
+ RouterConfig(name='foo', internal_subnets={'foo': 'bar'})
+
+ def test_bad_internal_subnets_no_project(self):
+ with self.assertRaises(RouterConfigError):
+ RouterConfig(name='foo', internal_subnets={
+ 'subnet': {'subnet_name': 'bar', 'network_name': 'foo'}})
+
+ def test_bad_internal_subnets_no_network(self):
+ with self.assertRaises(RouterConfigError):
+ RouterConfig(name='foo', internal_subnets={
+ 'subnet': {'subnet_name': 'bar', 'project_name': 'foo'}})
+
+ def test_bad_internal_subnets_no_subnet(self):
+ with self.assertRaises(RouterConfigError):
+ RouterConfig(name='foo', internal_subnets={
+ 'subnet': {'project_name': 'bar', 'network_name': 'foo'}})
+
def test_name_only(self):
settings = RouterConfig(name='foo')
self.assertEqual('foo', settings.name)
self.assertTrue(isinstance(settings.port_settings, list))
self.assertEqual(0, len(settings.port_settings))
- def test_all(self):
+ def test_all_internal_subnets_str(self):
port_settings = PortConfig(name='foo', network_name='bar')
settings = RouterConfig(
name='foo', project_name='bar', external_gateway='foo_gateway',
self.assertEqual(['10.0.0.1/24'], settings.internal_subnets)
self.assertEqual([port_settings], settings.port_settings)
- def test_config_all(self):
+ def test_all_internal_subnets_dict(self):
+ port_settings = PortConfig(name='foo', network_name='bar')
+ int_subs = {'subnet': {
+ 'project_name': 'proj_a', 'network_name': 'net_name',
+ 'subnet_name': 'sub_name'}}
+ settings = RouterConfig(
+ name='foo', project_name='bar', external_gateway='foo_gateway',
+ admin_state_up=True, enable_snat=False,
+ internal_subnets=int_subs,
+ 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.assertIsNotNone(settings.internal_subnets)
+ self.assertTrue(isinstance(settings.internal_subnets, dict))
+ self.assertEqual(1, len(settings.internal_subnets))
+ self.assertEqual(int_subs, settings.internal_subnets)
+ self.assertEqual([port_settings], settings.port_settings)
+
+ def test_config_all_internal_subnets_str(self):
+ int_subs = {'subnet': {
+ 'project_name': 'proj_a', 'network_name': 'net_name',
+ 'subnet_name': 'sub_name'}}
settings = RouterConfig(
**{'name': 'foo', 'project_name': 'bar',
'external_gateway': 'foo_gateway', 'admin_state_up': True,
- 'enable_snat': False, 'internal_subnets': ['10.0.0.1/24'],
+ 'enable_snat': False,
+ 'internal_subnets': int_subs,
'interfaces':
[{'port': {'name': 'foo-port',
'network_name': 'bar-net'}}]})
self.assertTrue(settings.admin_state_up)
self.assertFalse(settings.enable_snat)
self.assertIsNotNone(settings.internal_subnets)
- self.assertTrue(isinstance(settings.internal_subnets, list))
+ self.assertTrue(isinstance(settings.internal_subnets, dict))
self.assertEqual(1, len(settings.internal_subnets))
- self.assertEqual(['10.0.0.1/24'], settings.internal_subnets)
+ self.assertEqual(int_subs, settings.internal_subnets)
self.assertEqual([PortConfig(**{'name': 'foo-port',
'network_name': 'bar-net'})],
settings.port_settings)
"""
self.initialize()
- if len(self.__ports) == 0:
+ if len(self.__ports) != len(self.instance_settings.port_settings):
self.__ports = self.__create_ports(
self.instance_settings.port_settings)
if not self.__vm:
for port_setting in port_settings:
port = neutron_utils.get_port(
- self.__neutron, self.__keystone, port_settings=port_setting)
+ self.__neutron, self.__keystone, port_settings=port_setting,
+ project_name=self._os_creds.project_name)
if not port:
port = neutron_utils.create_port(
self.__neutron, self._os_creds, port_setting)
- if port:
- ports.append((port_setting.name, port))
+ if port:
+ ports.append((port_setting.name, port))
return ports
port = self.get_port_by_name(port_name)
if port:
if subnet_name:
+ network = neutron_utils.get_network_by_id(
+ self.__neutron, port.network_id)
subnet = neutron_utils.get_subnet(
- self.__neutron, subnet_name=subnet_name)
+ self.__neutron, network, subnet_name=subnet_name)
if not subnet:
logger.warning('Cannot retrieve port IP as subnet could '
'not be located with name - %s',
from snaps.config.router import RouterConfig
from snaps.openstack.openstack_creator import OpenStackNetworkObject
-from snaps.openstack.utils import neutron_utils, keystone_utils
+from snaps.openstack.utils import neutron_utils
__author__ = 'spisarski'
self.router_settings.name, e)
if self.__router:
- for internal_subnet_name in self.router_settings.internal_subnets:
- internal_subnet = neutron_utils.get_subnet(
- self._neutron, subnet_name=internal_subnet_name)
+ for sub_config in self.router_settings.internal_subnets:
+ internal_subnet = self.__get_internal_subnet(sub_config)
if internal_subnet:
self.__internal_subnets.append(internal_subnet)
else:
raise RouterCreationError(
- 'Subnet not found with name ' + internal_subnet_name)
+ 'Subnet not found with name ' + internal_subnet.name)
for port_setting in self.router_settings.port_settings:
port = neutron_utils.get_port(
self.__router = neutron_utils.create_router(
self._neutron, self._os_creds, self.router_settings)
- for internal_subnet_name in self.router_settings.internal_subnets:
- internal_subnet = neutron_utils.get_subnet(
- self._neutron, subnet_name=internal_subnet_name)
+ for sub_config in self.router_settings.internal_subnets:
+ internal_subnet = self.__get_internal_subnet(sub_config)
if internal_subnet:
self.__internal_subnets.append(internal_subnet)
if internal_subnet:
self.__internal_router_interface = router_intf
else:
raise RouterCreationError(
- 'Subnet not found with name ' + internal_subnet_name)
+ 'Subnet not found with name ' + internal_subnet.name)
for port_setting in self.router_settings.port_settings:
port = neutron_utils.get_port(
self._neutron, self.__router.id)
return self.__router
+ def __get_internal_subnet(self, sub_config):
+ """
+ returns the Subnet domain object from the subnet configurator
+ :param sub_config:
+ :return:
+ """
+ if isinstance(sub_config, str):
+ return neutron_utils.get_subnet_by_name(
+ self._neutron, self._keystone,
+ subnet_name=sub_config)
+ if isinstance(sub_config, dict):
+ sub_dict = sub_config['subnet']
+ network = neutron_utils.get_network(
+ self._neutron, self._keystone,
+ network_name=sub_dict['network_name'],
+ project_name=sub_dict['project_name'])
+ if network:
+ return neutron_utils.get_subnet(
+ self._neutron, network,
+ subnet_name=sub_dict['subnet_name'])
+
def clean(self):
"""
Removes and deletes all items created in reverse order.
self.inst_creator = None
self.priv_net_config = openstack_tests.get_priv_net_config(
+ project_name=self.os_creds.project_name,
net_name=guid + '-priv-net',
subnet_name=guid + '-priv-subnet',
netconf_override=self.netconf_override)
"""
# Create Network
net_config = openstack_tests.get_priv_net_config(
+ project_name=self.os_creds.project_name,
net_name=self.guid + '-pub-net',
subnet_name=self.guid + '-pub-subnet',
router_name=self.guid + '-pub-router',
"""
# Create Network
net_config = openstack_tests.get_priv_net_config(
+ project_name=self.os_creds.project_name,
net_name=self.guid + '-pub-net',
subnet_name=self.guid + '-pub-subnet',
router_name=self.guid + '-pub-router',
self.inst_creators = list()
self.pub_net_config = openstack_tests.get_pub_net_config(
+ project_name=self.os_creds.project_name,
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
router_name=guid + '-pub-router', external_net=self.ext_net_name,
netconf_override=self.netconf_override)
name=self.guid + '-net', subnet_settings=[subnet_settings])
router_settings = RouterConfig(
name=self.guid + '-router', external_gateway=self.ext_net_name,
- internal_subnets=[subnet_settings.name])
+ internal_subnets=[{'subnet': {
+ 'project_name': self.os_creds.project_name,
+ 'network_name': network_settings.name,
+ 'subnet_name': subnet_settings.name}}])
# Create Network
self.network_creator = OpenStackNetwork(
subnet_settings=[subnet4_settings, subnet6_settings])
router_settings = RouterConfig(
name=self.guid + '-router', external_gateway=self.ext_net_name,
- internal_subnets=[subnet4_settings.name])
+ internal_subnets=[{'subnet': {
+ 'project_name': self.os_creds.project_name,
+ 'network_name': network_settings.name,
+ 'subnet_name': subnet4_settings.name}}])
# Create Network
self.network_creator = OpenStackNetwork(
self.inst_creator = None
self.net_config = openstack_tests.get_priv_net_config(
+ project_name=self.os_creds.project_name,
net_name=self.guid + '-pub-net',
subnet_name=self.guid + '-pub-subnet',
router_name=self.guid + '-pub-router',
self.inst_creators = list()
self.priv_net_config = openstack_tests.get_priv_net_config(
+ project_name=self.os_creds.project_name,
net_name=guid + '-priv-net', subnet_name=guid + '-priv-subnet',
netconf_override=self.netconf_override)
self.floating_ip_name = self.guid + 'fip1'
net_config = openstack_tests.get_priv_net_config(
+ project_name=self.os_creds.project_name,
net_name=self.guid + '-pub-net',
subnet_name=self.guid + '-pub-subnet',
router_name=self.guid + '-pub-router',
self.nova = nova_utils.nova_client(self.os_creds, self.os_session)
net_config = openstack_tests.get_priv_net_config(
+ project_name=self.os_creds.project_name,
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
router_name=guid + '-pub-router', external_net=self.ext_net_name,
netconf_override=self.netconf_override)
self.inst_creator = None
self.priv_net_config = openstack_tests.get_priv_net_config(
+ project_name=self.os_creds.project_name,
net_name=self.guid + '-priv-net',
subnet_name=self.guid + '-priv-subnet')
self.port_settings = PortConfig(
name=guid + '-image', image_metadata=self.image_metadata)
net_config = openstack_tests.get_priv_net_config(
+ project_name=self.os_creds.project_name,
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
router_name=guid + '-pub-router', external_net=self.ext_net_name,
netconf_override=self.netconf_override)
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.net_config = openstack_tests.get_pub_net_config(
+ project_name=self.os_creds.project_name,
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
router_name=guid + '-pub-router', external_net=self.ext_net_name,
netconf_override=self.netconf_override)
# Create Network
self.net_creator = OpenStackNetwork(
self.os_creds, self.net_config.network_settings)
- self.net_creator.create()
+ network = self.net_creator.create()
# Validate network was created
self.assertTrue(neutron_utils_tests.validate_network(
# Validate subnets
self.assertTrue(neutron_utils_tests.validate_subnet(
- self.neutron,
+ self.neutron, network,
self.net_creator.network_settings.subnet_settings[0].name,
self.net_creator.network_settings.subnet_settings[0].cidr, True))
# Create Network
self.net_creator = OpenStackNetwork(
self.os_creds, self.net_config.network_settings)
- self.net_creator.create()
+ network = self.net_creator.create()
# Create Router
self.router_creator = create_router.OpenStackRouter(
# Validate subnets
self.assertTrue(neutron_utils_tests.validate_subnet(
- self.neutron,
+ self.neutron, network,
self.net_creator.network_settings.subnet_settings[0].name,
self.net_creator.network_settings.subnet_settings[0].cidr, True))
# Validate subnets
self.assertTrue(neutron_utils_tests.validate_subnet(
- self.neutron,
+ self.neutron, out_net,
self.net_creator.network_settings.subnet_settings[0].name,
self.net_creator.network_settings.subnet_settings[0].cidr, True))
# Validate subnets
self.assertTrue(neutron_utils_tests.validate_subnet(
- self.neutron,
+ self.neutron, out_net,
self.net_creator.network_settings.subnet_settings[0].name,
self.net_creator.network_settings.subnet_settings[0].cidr, True))
# Validate subnets
self.assertTrue(neutron_utils_tests.validate_subnet(
- self.neutron,
+ self.neutron, out_net,
self.net_creator.network_settings.subnet_settings[0].name,
self.net_creator.network_settings.subnet_settings[0].cidr, True))
"""
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.net_config = openstack_tests.get_pub_net_config(
+ project_name=self.os_creds.project_name,
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet')
self.neutron = neutron_utils.neutron_client(
guid = self.__class__.__name__ + '-' + str(uuid.uuid4())
self.net_config = openstack_tests.get_pub_net_config(
+ project_name=self.os_creds.project_name,
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet')
self.neutron = neutron_utils.neutron_client(
self.assertEqual(1, len(net_creators[0].get_network().subnets))
subnet = net_creators[0].get_network().subnets[0]
subnet_by_name = neutron_utils.get_subnet(
- neutron, subnet_name=subnet.name)
+ neutron, net_creators[0].get_network(), subnet_name=subnet.name)
self.assertEqual(subnet, subnet_by_name)
subnet_by_id = neutron_utils.get_subnet_by_id(neutron, subnet.id)
public=public)
-def get_priv_net_config(net_name, subnet_name, router_name=None,
+def get_priv_net_config(project_name, net_name, subnet_name, router_name=None,
cidr='10.55.0.0/24', external_net=None,
netconf_override=None):
- return OSNetworkConfig(net_name, subnet_name, cidr, router_name,
- external_gateway=external_net,
- netconf_override=netconf_override)
+ return OSNetworkConfig(
+ project_name, net_name, subnet_name, cidr, router_name,
+ external_gateway=external_net, netconf_override=netconf_override)
-def get_pub_net_config(net_name, subnet_name=None, router_name=None,
- cidr='10.55.1.0/24', external_net=None,
- netconf_override=None):
- return OSNetworkConfig(net_name, subnet_name, cidr, router_name,
- external_gateway=external_net,
+def get_pub_net_config(
+ project_name, net_name, subnet_name=None, router_name=None,
+ cidr='10.55.1.0/24', external_net=None, netconf_override=None):
+ return OSNetworkConfig(project_name, net_name, subnet_name, cidr,
+ router_name, external_gateway=external_net,
netconf_override=netconf_override)
physical_network and segmentation_id
"""
- def __init__(self, net_name, subnet_name=None, subnet_cidr=None,
- router_name=None, external_gateway=None,
+ def __init__(self, project_name, net_name, subnet_name=None,
+ subnet_cidr=None, router_name=None, external_gateway=None,
netconf_override=None):
"""
:param netconf_override: dict() containing the reconfigured
if subnet_name:
self.router_settings = RouterConfig(
name=router_name, external_gateway=external_gateway,
- internal_subnets=[subnet_name])
+ internal_subnets=[{'subnet': {
+ 'project_name': project_name,
+ 'network_name': net_name,
+ 'subnet_name': subnet_name}}])
else:
self.router_settings = RouterConfig(
name=router_name, external_gateway=external_gateway)
creator.create()
out[inst_config['name']] = creator
+ else:
+ raise Exception('Unable to instantiate creator')
logger.info('Initialized configured %ss', config_key)
neutron.delete_subnet(subnet.id)
-def get_subnet(neutron, subnet_settings=None, subnet_name=None):
+def get_subnet(neutron, network, subnet_settings=None, subnet_name=None):
"""
Returns the first subnet object that fits the query else None including
if subnet_settings or subnet_name parameters are None.
:param neutron: the client
+ :param network: the associated SNAPS-OO Network domain object
:param subnet_settings: the subnet settings of the object to retrieve
:param subnet_name: the name of the subnet to retrieve
:return: a SNAPS-OO Subnet domain object or None
"""
- sub_filter = dict()
+ sub_filter = {'network_id': network.id}
if subnet_settings:
sub_filter['name'] = subnet_settings.name
sub_filter['cidr'] = subnet_settings.cidr
return Subnet(**subnet)
+def get_subnet_by_name(neutron, keystone, subnet_name, project_name=None):
+ """
+ Returns the first subnet object that fits the query else None including
+ if subnet_settings or subnet_name parameters are None.
+ :param neutron: the Neutron client
+ :param keystone: the Keystone client
+ :param subnet_name: the name of the subnet to retrieve
+ :param project_name: the name of the associated project to the subnet to
+ retrieve
+ :return: a SNAPS-OO Subnet domain object or None
+ """
+ project = None
+ if project_name:
+ project = keystone_utils.get_project(
+ keystone, project_name=project_name)
+ if project:
+ sub_filter = {'name': subnet_name, 'project_id': project.id}
+ subnets = neutron.list_subnets(**sub_filter)
+ for subnet in subnets['subnets']:
+ return Subnet(**subnet)
+ else:
+ sub_filter = {'name': subnet_name}
+ subnets = neutron.list_subnets(**sub_filter)
+ for subnet in subnets['subnets']:
+ return Subnet(**subnet)
+
+
def get_subnet_by_id(neutron, subnet_id):
"""
Returns a SNAPS-OO Subnet domain object for a given ID
return None
groups = neutron.list_security_groups(**sec_grp_filter)
- group = None
for group in groups['security_groups']:
if project_name:
project = keystone_utils.get_project_by_id(
keystone, group['tenant_id'])
if project and project_name == project.name:
- break
+ return __map_os_security_group(neutron, group)
else:
- break
- if group:
- return __map_os_security_group(neutron, group)
+ return __map_os_security_group(neutron, group)
def __map_os_security_group(neutron, os_sec_grp):
self.os_creds, self.os_session)
self.network = None
self.net_config = openstack_tests.get_pub_net_config(
+ project_name=self.os_creds.project_name,
net_name=guid + '-pub-net')
def tearDown(self):
self.os_creds, self.os_session)
self.network = None
self.net_config = openstack_tests.get_pub_net_config(
+ project_name=self.os_creds.project_name,
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
external_net=self.ext_net_name)
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
- self.neutron, subnet_setting.name, subnet_setting.cidr, True))
+ self.neutron, self.network, subnet_setting.name,
+ subnet_setting.cidr, True))
subnet_query1 = neutron_utils.get_subnet(
- self.neutron, subnet_name=subnet_setting.name)
+ self.neutron, self.network, subnet_name=subnet_setting.name)
self.assertEqual(self.network.subnets[0], subnet_query1)
subnet_query2 = neutron_utils.get_subnets_by_network(self.neutron,
self.assertEqual(1, len(subnet_query2))
self.assertEqual(self.network.subnets[0], subnet_query2[0])
+ subnet_query3 = neutron_utils.get_subnet_by_name(
+ self.neutron, self.keystone, subnet_setting.name,
+ self.os_creds.project_name)
+ self.assertIsNotNone(subnet_query3)
+ self.assertEqual(self.network.subnets[0], subnet_query3)
+
def test_create_subnet_null_name(self):
"""
Tests the neutron_utils.create_neutron_subnet() function for an
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
- self.neutron, subnet_setting.name, subnet_setting.cidr, True))
+ self.neutron, self.network, subnet_setting.name,
+ subnet_setting.cidr, True))
self.assertFalse(validate_subnet(
- self.neutron, '', subnet_setting.cidr, True))
+ self.neutron, self.network, '', subnet_setting.cidr, True))
subnet_query1 = neutron_utils.get_subnet(
- self.neutron, subnet_name=subnet_setting.name)
+ self.neutron, self.network, subnet_name=subnet_setting.name)
self.assertEqual(self.network.subnets[0], subnet_query1)
- subnet_query2 = neutron_utils.get_subnets_by_network(self.neutron,
- self.network)
+ subnet_query2 = neutron_utils.get_subnets_by_network(
+ self.neutron, self.network)
self.assertIsNotNone(subnet_query2)
self.assertEqual(1, len(subnet_query2))
self.assertEqual(self.network.subnets[0], subnet_query2[0])
self.router = None
self.interface_router = None
self.net_config = openstack_tests.get_pub_net_config(
+ project_name=self.os_creds.project_name,
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
router_name=guid + '-pub-router', external_net=self.ext_net_name)
"""
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.net_config = openstack_tests.OSNetworkConfig(
- self.net_config.network_settings.name, subnet_setting.name,
- subnet_setting.cidr, self.net_config.router_settings.name,
- self.ext_net_name)
+ self.os_creds.project_name, self.net_config.network_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(
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
- self.neutron, subnet_setting.name, subnet_setting.cidr, True))
+ self.neutron, self.network, subnet_setting.name,
+ subnet_setting.cidr, True))
self.router = neutron_utils.create_router(
self.neutron, self.os_creds, self.net_config.router_settings)
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
- self.neutron, subnet_setting.name, subnet_setting.cidr, True))
+ self.neutron, self.network, subnet_setting.name,
+ subnet_setting.cidr, True))
with self.assertRaises(NeutronException):
self.interface_router = neutron_utils.add_interface_router(
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
- self.neutron, subnet_setting.name, subnet_setting.cidr, True))
+ self.neutron, self.network, subnet_setting.name,
+ subnet_setting.cidr, True))
self.port = neutron_utils.create_port(
self.neutron, self.os_creds, PortConfig(
self.os_creds.project_name))
subnet_setting = self.net_config.network_settings.subnet_settings[0]
- self.assertTrue(validate_subnet(self.neutron, subnet_setting.name,
- subnet_setting.cidr, True))
+ self.assertTrue(validate_subnet(
+ self.neutron, self.network, subnet_setting.name,
+ subnet_setting.cidr, True))
self.port = neutron_utils.create_port(
self.neutron, self.os_creds, PortConfig(
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
- self.neutron, subnet_setting.name, subnet_setting.cidr, True))
+ self.neutron, self.network, subnet_setting.name,
+ subnet_setting.cidr, True))
self.port = neutron_utils.create_port(
self.neutron, self.os_creds,
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
- self.neutron, subnet_setting.name, subnet_setting.cidr, True))
+ self.neutron, self.network, subnet_setting.name,
+ subnet_setting.cidr, True))
with self.assertRaises(Exception):
self.port = neutron_utils.create_port(
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
- self.neutron, subnet_setting.name, subnet_setting.cidr, True))
+ self.neutron, self.network, subnet_setting.name,
+ subnet_setting.cidr, True))
with self.assertRaises(Exception):
self.port = neutron_utils.create_port(
subnet_setting = self.net_config.network_settings.subnet_settings[0]
self.assertTrue(validate_subnet(
- self.neutron, subnet_setting.name, subnet_setting.cidr, True))
+ self.neutron, self.network, subnet_setting.name,
+ subnet_setting.cidr, True))
with self.assertRaises(Exception):
self.port = neutron_utils.create_port(
return False
-def validate_subnet(neutron, name, cidr, exists):
+def validate_subnet(neutron, network, 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.
:param neutron: The neutron client
+ :param network: The SNAPS-OO Network domain object
:param name: The expected subnet name
:param cidr: The expected CIDR value
:param exists: Whether or not the network name should exist or not
:return: True/False
"""
- subnet = neutron_utils.get_subnet(neutron, subnet_name=name)
+ subnet = neutron_utils.get_subnet(
+ neutron, network, subnet_name=name)
if exists and subnet and subnet.name == name:
return subnet.cidr == cidr
if not exists and not subnet:
self.image_creator.create()
network_settings = openstack_tests.get_priv_net_config(
- guid + '-net', guid + '-subnet').network_settings
+ self.os_creds.project_name, guid + '-net',
+ guid + '-subnet').network_settings
self.network_creator = OpenStackNetwork(
self.os_creds, network_settings)
self.network_creator.create()
self.image_creator.create()
network_settings = openstack_tests.get_priv_net_config(
- guid + '-net', guid + '-subnet').network_settings
+ self.os_creds.project_name, guid + '-net',
+ guid + '-subnet').network_settings
self.network_creator = OpenStackNetwork(
self.os_creds, network_settings)
self.network_creator.create()
# First network is public
self.pub_net_config = openstack_tests.get_pub_net_config(
+ project_name=self.os_creds.project_name,
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
router_name=guid + '-pub-router',
external_net=self.ext_net_name)
# First network is public
self.pub_net_config = openstack_tests.get_pub_net_config(
+ project_name=self.os_creds.project_name,
net_name=guid + '-pub-net', subnet_name=guid + '-pub-subnet',
router_name=guid + '-pub-router',
external_net=self.ext_net_name)