import logging
from neutronclient.common.exceptions import NotFound
-
from snaps.openstack.utils import keystone_utils, neutron_utils
__author__ = 'spisarski'
def create(self, cleanup=False):
"""
- Responsible for creating not only the network but then a private subnet, router, and an interface to the router.
+ Responsible for creating not only the network but then a private
+ subnet, router, and an interface to the router.
:param cleanup: When true, only perform lookups for OpenStack objects.
:return: the created network object or None
"""
self.__neutron = neutron_utils.neutron_client(self.__os_creds)
- logger.info('Creating neutron network %s...' % self.network_settings.name)
- net_inst = neutron_utils.get_network(self.__neutron, self.network_settings.name,
- self.network_settings.get_project_id(self.__os_creds))
+ logger.info(
+ 'Creating neutron network %s...' % self.network_settings.name)
+ net_inst = neutron_utils.get_network(
+ self.__neutron, self.network_settings.name,
+ self.network_settings.get_project_id(self.__os_creds))
if net_inst:
self.__network = net_inst
else:
if not cleanup:
- self.__network = neutron_utils.create_network(self.__neutron, self.__os_creds,
- self.network_settings)
+ self.__network = neutron_utils.create_network(
+ self.__neutron, self.__os_creds, self.network_settings)
else:
- logger.info('Network does not exist and will not create as in cleanup mode')
+ logger.info(
+ 'Network does not exist and will not create as in cleanup'
+ ' mode')
return
- logger.debug("Network '%s' created successfully" % self.__network['network']['id'])
+ logger.debug(
+ "Network '%s' created successfully" % self.__network['network'][
+ 'id'])
logger.debug('Creating Subnets....')
for subnet_setting in self.network_settings.subnet_settings:
- sub_inst = neutron_utils.get_subnet_by_name(self.__neutron, subnet_setting.name)
+ sub_inst = neutron_utils.get_subnet_by_name(self.__neutron,
+ subnet_setting.name)
if sub_inst:
self.__subnets.append(sub_inst)
- logger.debug("Subnet '%s' created successfully" % sub_inst['subnet']['id'])
+ logger.debug(
+ "Subnet '%s' created successfully" % sub_inst['subnet'][
+ 'id'])
else:
if not cleanup:
- self.__subnets.append(neutron_utils.create_subnet(self.__neutron, subnet_setting,
- self.__os_creds, self.__network))
+ self.__subnets.append(
+ neutron_utils.create_subnet(self.__neutron,
+ subnet_setting,
+ self.__os_creds,
+ self.__network))
return self.__network
"""
for subnet in self.__subnets:
try:
- logger.info('Deleting subnet with name ' + subnet['subnet']['name'])
+ logger.info(
+ 'Deleting subnet with name ' + subnet['subnet']['name'])
neutron_utils.delete_subnet(self.__neutron, subnet)
except NotFound as e:
- logger.warning('Error deleting subnet with message - ' + str(e))
+ logger.warning(
+ 'Error deleting subnet with message - ' + str(e))
pass
self.__subnets = list()
Class representing a network configuration
"""
- def __init__(self, config=None, name=None, admin_state_up=True, shared=None, project_name=None,
- external=False, network_type=None, physical_network=None, subnet_settings=list()):
+ def __init__(self, **kwargs):
"""
Constructor - all parameters are optional
- :param config: Should be a dict object containing the configuration settings using the attribute names below
- as each member's the key and overrides any of the other parameters.
:param name: The network name.
- :param admin_state_up: The administrative status of the network. True = up / False = down (default True)
- :param shared: Boolean value indicating whether this network is shared across all projects/tenants. By default,
- only administrative users can change this value.
- :param project_name: Admin-only. The name of the project that will own the network. This project can be
- different from the project that makes the create network request. However, only
- administrative users can specify a project ID other than their own. You cannot change this
- value through authorization policies.
- :param external: when true, will setup an external network (default False).
+ :param admin_state_up: The administrative status of the network.
+ True = up / False = down (default True)
+ :param shared: Boolean value indicating whether this network is shared
+ across all projects/tenants. By default, only
+ administrative users can change this value.
+ :param project_name: Admin-only. The name of the project that will own
+ the network. This project can be different from
+ the project that makes the create network request.
+ However, only administrative users can specify a
+ project ID other than their own. You cannot change
+ this value through authorization policies.
+ :param external: when true, will setup an external network
+ (default False).
:param network_type: the type of network (i.e. vlan|flat).
- :param physical_network: the name of the physical network (this is required when network_type is 'flat')
- :param subnet_settings: List of SubnetSettings objects.
+ :param physical_network: the name of the physical network
+ (this is required when network_type is 'flat')
+ :param subnets or subnet_settings: List of SubnetSettings objects.
:return:
"""
self.project_id = None
- if config:
- self.name = config.get('name')
- if config.get('admin_state_up') is not None:
- self.admin_state_up = bool(config['admin_state_up'])
- else:
- self.admin_state_up = admin_state_up
-
- if config.get('shared') is not None:
- self.shared = bool(config['shared'])
- else:
- self.shared = None
-
- self.project_name = config.get('project_name')
-
- if config.get('external') is not None:
- self.external = bool(config.get('external'))
- else:
- self.external = external
+ self.name = kwargs.get('name')
+ if kwargs.get('admin_state_up') is not None:
+ self.admin_state_up = bool(kwargs['admin_state_up'])
+ else:
+ self.admin_state_up = True
- self.network_type = config.get('network_type')
- self.physical_network = config.get('physical_network')
+ if kwargs.get('shared') is not None:
+ self.shared = bool(kwargs['shared'])
+ else:
+ self.shared = None
- self.subnet_settings = list()
- if config.get('subnets'):
- for subnet_config in config['subnets']:
- self.subnet_settings.append(SubnetSettings(config=subnet_config['subnet']))
+ self.project_name = kwargs.get('project_name')
+ if kwargs.get('external') is not None:
+ self.external = bool(kwargs.get('external'))
else:
- self.name = name
- self.admin_state_up = admin_state_up
- self.shared = shared
- self.project_name = project_name
- self.external = external
- self.network_type = network_type
- self.physical_network = physical_network
- self.subnet_settings = subnet_settings
+ self.external = False
+
+ self.network_type = kwargs.get('network_type')
+ self.physical_network = kwargs.get('physical_network')
+
+ self.subnet_settings = list()
+ subnet_settings = kwargs.get('subnets')
+ if not subnet_settings:
+ subnet_settings = kwargs.get('subnet_settings')
+ if subnet_settings:
+ for subnet_config in subnet_settings:
+ if isinstance(subnet_config, SubnetSettings):
+ self.subnet_settings.append(subnet_config)
+ else:
+ self.subnet_settings.append(
+ SubnetSettings(**subnet_config['subnet']))
if not self.name or len(self.name) < 1:
raise Exception('Name required for networks')
else:
if self.project_name:
keystone = keystone_utils.keystone_client(os_creds)
- project = keystone_utils.get_project(keystone, self.project_name)
+ project = keystone_utils.get_project(keystone,
+ self.project_name)
if project:
return project.id
def dict_for_neutron(self, os_creds):
"""
Returns a dictionary object representing this object.
- This is meant to be converted into JSON designed for use by the Neutron API
-
+ This is meant to be converted into JSON designed for use by the Neutron
+ API
TODO - expand automated testing to exercise all parameters
:param os_creds: the OpenStack credentials
if project_id:
out['project_id'] = project_id
else:
- raise Exception('Could not find project ID for project named - ' + self.project_name)
+ raise Exception(
+ 'Could not find project ID for project named - ' +
+ self.project_name)
if self.network_type:
out['provider:network_type'] = self.network_type
if self.physical_network:
Class representing a subnet configuration
"""
- def __init__(self, config=None, cidr=None, ip_version=4, name=None, project_name=None, start=None,
- end=None, gateway_ip=None, enable_dhcp=None, dns_nameservers=None, host_routes=None, destination=None,
- nexthop=None, ipv6_ra_mode=None, ipv6_address_mode=None):
+ def __init__(self, **kwargs):
"""
Constructor - all parameters are optional except cidr (subnet mask)
- :param config: Should be a dict object containing the configuration settings using the attribute names below
- as each member's the key and overrides any of the other parameters.
:param cidr: The CIDR. REQUIRED if config parameter is None
:param ip_version: The IP version, which is 4 or 6.
:param name: The subnet name.
- :param project_name: The name of the project who owns the network. Only administrative users can specify a
- project ID other than their own. You cannot change this value through authorization
- policies.
+ :param project_name: The name of the project who owns the network.
+ Only administrative users can specify a project ID
+ other than their own. You cannot change this value
+ through authorization policies.
:param start: The start address for the allocation pools.
:param end: The end address for the allocation pools.
:param gateway_ip: The gateway IP address.
- :param enable_dhcp: Set to true if DHCP is enabled and false if DHCP is disabled.
- :param dns_nameservers: A list of DNS name servers for the subnet. Specify each name server as an IP address
- and separate multiple entries with a space. For example [8.8.8.7 8.8.8.8].
- :param host_routes: A list of host route dictionaries for the subnet. For example:
+ :param enable_dhcp: Set to true if DHCP is enabled and false if DHCP is
+ disabled.
+ :param dns_nameservers: A list of DNS name servers for the subnet.
+ Specify each name server as an IP address
+ and separate multiple entries with a space.
+ For example [8.8.8.7 8.8.8.8].
+ :param host_routes: A list of host route dictionaries for the subnet.
+ For example:
"host_routes":[
{
"destination":"0.0.0.0/0",
]
:param destination: The destination for static route
:param nexthop: The next hop for the destination.
- :param ipv6_ra_mode: A valid value is dhcpv6-stateful, dhcpv6-stateless, or slaac.
- :param ipv6_address_mode: A valid value is dhcpv6-stateful, dhcpv6-stateless, or slaac.
+ :param ipv6_ra_mode: A valid value is dhcpv6-stateful,
+ dhcpv6-stateless, or slaac.
+ :param ipv6_address_mode: A valid value is dhcpv6-stateful,
+ dhcpv6-stateless, or slaac.
:raise: Exception when config does not have or cidr values are None
"""
- if not dns_nameservers:
- dns_nameservers = ['8.8.8.8']
-
- if config:
- self.cidr = config['cidr']
- if config.get('ip_version'):
- self.ip_version = config['ip_version']
- else:
- self.ip_version = ip_version
-
- # Optional attributes that can be set after instantiation
- self.name = config.get('name')
- self.project_name = config.get('project_name')
- self.start = config.get('start')
- self.end = config.get('end')
- self.gateway_ip = config.get('gateway_ip')
- self.enable_dhcp = config.get('enable_dhcp')
-
- if config.get('dns_nameservers'):
- self.dns_nameservers = config.get('dns_nameservers')
- else:
- self.dns_nameservers = dns_nameservers
-
- self.host_routes = config.get('host_routes')
- self.destination = config.get('destination')
- self.nexthop = config.get('nexthop')
- self.ipv6_ra_mode = config.get('ipv6_ra_mode')
- self.ipv6_address_mode = config.get('ipv6_address_mode')
+ self.cidr = kwargs['cidr']
+ if kwargs.get('ip_version'):
+ self.ip_version = kwargs['ip_version']
+ else:
+ self.ip_version = 4
+
+ # Optional attributes that can be set after instantiation
+ self.name = kwargs.get('name')
+ self.project_name = kwargs.get('project_name')
+ self.start = kwargs.get('start')
+ self.end = kwargs.get('end')
+ self.gateway_ip = kwargs.get('gateway_ip')
+ self.enable_dhcp = kwargs.get('enable_dhcp')
+
+ if kwargs.get('dns_nameservers'):
+ self.dns_nameservers = kwargs.get('dns_nameservers')
else:
- # Required attributes
- self.cidr = cidr
- self.ip_version = ip_version
-
- # Optional attributes that can be set after instantiation
- self.name = name
- self.project_name = project_name
- self.start = start
- self.end = end
- self.gateway_ip = gateway_ip
- self.enable_dhcp = enable_dhcp
- self.dns_nameservers = dns_nameservers
- self.host_routes = host_routes
- self.destination = destination
- self.nexthop = nexthop
- self.ipv6_ra_mode = ipv6_ra_mode
- self.ipv6_address_mode = ipv6_address_mode
+ self.dns_nameservers = ['8.8.8.8']
+
+ self.host_routes = kwargs.get('host_routes')
+ self.destination = kwargs.get('destination')
+ self.nexthop = kwargs.get('nexthop')
+ self.ipv6_ra_mode = kwargs.get('ipv6_ra_mode')
+ self.ipv6_address_mode = kwargs.get('ipv6_address_mode')
if not self.name or not self.cidr:
raise Exception('Name and cidr required for subnets')
def dict_for_neutron(self, os_creds, network=None):
"""
Returns a dictionary object representing this object.
- This is meant to be converted into JSON designed for use by the Neutron API
+ This is meant to be converted into JSON designed for use by the Neutron
+ API
:param os_creds: the OpenStack credentials
- :param network: (Optional) the network object on which the subnet will be created
+ :param network: The network object on which the subnet will be created
+ (optional)
:return: the dictionary object
"""
out = {
if project_id:
out['project_id'] = project_id
else:
- raise Exception('Could not find project ID for project named - ' + self.project_name)
+ raise Exception(
+ 'Could not find project ID for project named - ' +
+ self.project_name)
if self.start and self.end:
out['allocation_pools'] = [{'start': self.start, 'end': self.end}]
if self.gateway_ip:
Class representing a port configuration
"""
- def __init__(self, config=None, name=None, network_name=None, admin_state_up=True, project_name=None,
- mac_address=None, ip_addrs=None, fixed_ips=None, security_groups=None, allowed_address_pairs=None,
- opt_value=None, opt_name=None, device_owner=None, device_id=None):
+ def __init__(self, **kwargs):
"""
Constructor - all parameters are optional
- :param config: Should be a dict object containing the configuration settings using the attribute names below
- as each member's the key and overrides any of the other parameters.
:param name: A symbolic name for the port.
- :param network_name: The name of the network on which to create the port.
- :param admin_state_up: A boolean value denoting the administrative status of the port. True = up / False = down
- :param project_name: The name of the project who owns the network. Only administrative users can specify a
- project ID other than their own. You cannot change this value through authorization
- policies.
- :param mac_address: The MAC address. If you specify an address that is not valid, a Bad Request (400) status
- code is returned. If you do not specify a MAC address, OpenStack Networking tries to
- allocate one. If a failure occurs, a Service Unavailable (503) status code is returned.
- :param ip_addrs: A list of dict objects where each contains two keys 'subnet_name' and 'ip' values which will
- get mapped to self.fixed_ips.
- These values will be directly translated into the fixed_ips dict
- :param fixed_ips: A dict where the key is the subnet IDs and value is the IP address to assign to the port
+ :param network_name: The name of the network on which to create the
+ port.
+ :param admin_state_up: A boolean value denoting the administrative
+ status of the port. True = up / False = down
+ :param project_name: The name of the project who owns the network.
+ Only administrative users can specify a project ID
+ other than their own. You cannot change this value
+ through authorization policies.
+ :param mac_address: The MAC address. If you specify an address that is
+ not valid, a Bad Request (400) status code is
+ returned. If you do not specify a MAC address,
+ OpenStack Networking tries to allocate one. If a
+ failure occurs, a Service Unavailable (503) status
+ code is returned.
+ :param ip_addrs: A list of dict objects where each contains two keys
+ 'subnet_name' and 'ip' values which will get mapped to
+ self.fixed_ips. These values will be directly
+ translated into the fixed_ips dict
+ :param fixed_ips: A dict where the key is the subnet IDs and value is
+ the IP address to assign to the port
: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
- pair contains an IP address and MAC address.
+ :param allowed_address_pairs: A dictionary containing a set of zero or
+ more allowed address pairs. An address
+ pair contains an IP address and MAC
+ address.
:param opt_value: The extra DHCP option value.
:param opt_name: The extra DHCP option name.
- :param device_owner: The ID of the entity that uses this port. For example, a DHCP agent.
- :param device_id: The ID of the device that uses this port. For example, a virtual server.
+ :param device_owner: The ID of the entity that uses this port.
+ For example, a DHCP agent.
+ :param device_id: The ID of the device that uses this port.
+ For example, a virtual server.
:return:
"""
self.network = None
- if config:
- self.name = config.get('name')
- self.network_name = config.get('network_name')
+ self.name = kwargs.get('name')
+ self.network_name = kwargs.get('network_name')
- if config.get('admin_state_up') is not None:
- self.admin_state_up = bool(config['admin_state_up'])
- else:
- self.admin_state_up = admin_state_up
-
- self.project_name = config.get('project_name')
- self.mac_address = config.get('mac_address')
- self.ip_addrs = config.get('ip_addrs')
- self.fixed_ips = config.get('fixed_ips')
- self.security_groups = config.get('security_groups')
- self.allowed_address_pairs = config.get('allowed_address_pairs')
- self.opt_value = config.get('opt_value')
- self.opt_name = config.get('opt_name')
- self.device_owner = config.get('device_owner')
- self.device_id = config.get('device_id')
+ if kwargs.get('admin_state_up') is not None:
+ self.admin_state_up = bool(kwargs['admin_state_up'])
else:
- self.name = name
- self.network_name = network_name
- self.admin_state_up = admin_state_up
- self.project_name = project_name
- self.mac_address = mac_address
- self.ip_addrs = ip_addrs
- self.fixed_ips = fixed_ips
- self.security_groups = security_groups
- self.allowed_address_pairs = allowed_address_pairs
- self.opt_value = opt_value
- self.opt_name = opt_name
- self.device_owner = device_owner
- self.device_id = device_id
+ self.admin_state_up = True
+
+ 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')
+ self.opt_name = kwargs.get('opt_name')
+ self.device_owner = kwargs.get('device_owner')
+ self.device_id = kwargs.get('device_id')
if not self.name or not self.network_name:
- raise Exception('The attributes neutron, name, and network_name are required for PortSettings')
+ raise Exception(
+ 'The attributes neutron, name, and network_name are required '
+ 'for PortSettings')
def __set_fixed_ips(self, neutron):
"""
self.fixed_ips = list()
for ip_addr_dict in self.ip_addrs:
- subnet = neutron_utils.get_subnet_by_name(neutron, ip_addr_dict['subnet_name'])
+ subnet = neutron_utils.get_subnet_by_name(neutron,
+ ip_addr_dict[
+ 'subnet_name'])
if subnet:
- self.fixed_ips.append({'ip_address': ip_addr_dict['ip'], 'subnet_id': subnet['subnet']['id']})
+ self.fixed_ips.append({'ip_address': ip_addr_dict['ip'],
+ 'subnet_id': subnet['subnet'][
+ 'id']})
else:
- raise Exception('Invalid port configuration, subnet does not exist with name - ' +
- ip_addr_dict['subnet_name'])
+ raise Exception(
+ 'Invalid port configuration, subnet does not exist '
+ 'with name - ' + ip_addr_dict['subnet_name'])
def dict_for_neutron(self, neutron, os_creds):
"""
Returns a dictionary object representing this object.
- This is meant to be converted into JSON designed for use by the Neutron API
+ This is meant to be converted into JSON designed for use by the Neutron
+ API
TODO - expand automated testing to exercise all parameters
:param neutron: the Neutron client
project_id = project.id
if not self.network:
- self.network = neutron_utils.get_network(neutron, self.network_name, project_id)
+ self.network = neutron_utils.get_network(neutron,
+ self.network_name,
+ project_id)
if not self.network:
- raise Exception('Cannot locate network with name - ' + self.network_name)
+ raise Exception(
+ 'Cannot locate network with name - ' + self.network_name)
out['network_id'] = self.network['network']['id']
if project_id:
out['project_id'] = project_id
else:
- raise Exception('Could not find project ID for project named - ' + self.project_name)
+ raise Exception(
+ 'Could not find project ID for project named - ' +
+ self.project_name)
if self.mac_address:
out['mac_address'] = self.mac_address
if self.fixed_ips and len(self.fixed_ips) > 0:
# 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 uuid
import unittest
+import uuid
-from snaps.openstack.create_network import OpenStackNetwork, NetworkSettings, SubnetSettings, PortSettings
from snaps.openstack import create_router
+from snaps.openstack.create_network import (OpenStackNetwork, NetworkSettings,
+ SubnetSettings, PortSettings)
from snaps.openstack.tests import openstack_tests
-from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase, OSComponentTestCase
+from snaps.openstack.tests.os_source_file_test import (OSIntegrationTestCase,
+ OSComponentTestCase)
from snaps.openstack.utils import neutron_utils
from snaps.openstack.utils.tests import neutron_utils_tests
def test_empty_config(self):
with self.assertRaises(Exception):
- NetworkSettings(config=dict())
+ NetworkSettings(**dict())
def test_name_only(self):
settings = NetworkSettings(name='foo')
self.assertEqual(0, len(settings.subnet_settings))
def test_config_with_name_only(self):
- settings = NetworkSettings(config={'name': 'foo'})
+ settings = NetworkSettings(**{'name': 'foo'})
self.assertEqual('foo', settings.name)
self.assertTrue(settings.admin_state_up)
self.assertIsNone(settings.shared)
def test_all(self):
sub_settings = SubnetSettings(name='foo-subnet', cidr='10.0.0.0/24')
- settings = NetworkSettings(name='foo', admin_state_up=False, shared=True, project_name='bar', external=True,
- network_type='flat', physical_network='phy', subnet_settings=[sub_settings])
+ settings = NetworkSettings(name='foo', admin_state_up=False,
+ shared=True, project_name='bar',
+ external=True,
+ network_type='flat', physical_network='phy',
+ subnet_settings=[sub_settings])
self.assertEqual('foo', settings.name)
self.assertFalse(settings.admin_state_up)
self.assertTrue(settings.shared)
self.assertEqual('foo-subnet', settings.subnet_settings[0].name)
def test_config_all(self):
- settings = NetworkSettings(config={'name': 'foo', 'admin_state_up': False, 'shared': True,
- 'project_name': 'bar', 'external': True, 'network_type': 'flat',
- 'physical_network': 'phy',
- 'subnets':
- [{'subnet': {'name': 'foo-subnet', 'cidr': '10.0.0.0/24'}}]})
+ settings = NetworkSettings(
+ **{'name': 'foo', 'admin_state_up': False, 'shared': True,
+ 'project_name': 'bar', 'external': True, 'network_type': 'flat',
+ 'physical_network': 'phy',
+ 'subnets':
+ [{'subnet': {'name': 'foo-subnet',
+ 'cidr': '10.0.0.0/24'}}]})
self.assertEqual('foo', settings.name)
self.assertFalse(settings.admin_state_up)
self.assertTrue(settings.shared)
def test_empty_config(self):
with self.assertRaises(Exception):
- SubnetSettings(config=dict())
+ SubnetSettings(**dict())
def test_name_only(self):
with self.assertRaises(Exception):
def test_config_with_name_only(self):
with self.assertRaises(Exception):
- SubnetSettings(config={'name': 'foo'})
+ SubnetSettings(**{'name': 'foo'})
def test_name_cidr_only(self):
settings = SubnetSettings(name='foo', cidr='10.0.0.0/24')
self.assertIsNone(settings.ipv6_address_mode)
def test_config_with_name_cidr_only(self):
- settings = SubnetSettings(config={'name': 'foo', 'cidr': '10.0.0.0/24'})
+ settings = SubnetSettings(**{'name': 'foo', 'cidr': '10.0.0.0/24'})
self.assertEqual('foo', settings.name)
self.assertEqual('10.0.0.0/24', settings.cidr)
self.assertEqual(4, settings.ip_version)
def test_all(self):
host_routes = {'destination': '0.0.0.0/0', 'nexthop': '123.456.78.9'}
- settings = SubnetSettings(name='foo', cidr='10.0.0.0/24', ip_version=6, project_name='bar-project',
- start='10.0.0.2', end='10.0.0.101', gateway_ip='10.0.0.1', enable_dhcp=False,
- dns_nameservers=['8.8.8.8'], host_routes=[host_routes], destination='dest',
- nexthop='hop', ipv6_ra_mode='dhcpv6-stateful', ipv6_address_mode='slaac')
+ settings = SubnetSettings(name='foo', cidr='10.0.0.0/24', ip_version=6,
+ project_name='bar-project',
+ start='10.0.0.2', end='10.0.0.101',
+ gateway_ip='10.0.0.1', enable_dhcp=False,
+ dns_nameservers=['8.8.8.8'],
+ host_routes=[host_routes],
+ destination='dest',
+ nexthop='hop',
+ ipv6_ra_mode='dhcpv6-stateful',
+ ipv6_address_mode='slaac')
self.assertEqual('foo', settings.name)
self.assertEqual('10.0.0.0/24', settings.cidr)
self.assertEqual(6, settings.ip_version)
def test_config_all(self):
host_routes = {'destination': '0.0.0.0/0', 'nexthop': '123.456.78.9'}
- settings = SubnetSettings(config={'name': 'foo', 'cidr': '10.0.0.0/24', 'ip_version': 6,
- 'project_name': 'bar-project', 'start': '10.0.0.2', 'end': '10.0.0.101',
- 'gateway_ip': '10.0.0.1', 'enable_dhcp': False,
- 'dns_nameservers': ['8.8.8.8'], 'host_routes': [host_routes],
- 'destination': 'dest', 'nexthop': 'hop', 'ipv6_ra_mode': 'dhcpv6-stateful',
- 'ipv6_address_mode': 'slaac'})
+ settings = SubnetSettings(
+ **{'name': 'foo', 'cidr': '10.0.0.0/24', 'ip_version': 6,
+ 'project_name': 'bar-project', 'start': '10.0.0.2',
+ 'end': '10.0.0.101',
+ 'gateway_ip': '10.0.0.1', 'enable_dhcp': False,
+ 'dns_nameservers': ['8.8.8.8'], 'host_routes': [host_routes],
+ 'destination': 'dest', 'nexthop': 'hop',
+ 'ipv6_ra_mode': 'dhcpv6-stateful',
+ 'ipv6_address_mode': 'slaac'})
self.assertEqual('foo', settings.name)
self.assertEqual('10.0.0.0/24', settings.cidr)
self.assertEqual(6, settings.ip_version)
def test_empty_config(self):
with self.assertRaises(Exception):
- PortSettings(config=dict())
+ PortSettings(**dict())
def test_name_only(self):
with self.assertRaises(Exception):
def test_config_name_only(self):
with self.assertRaises(Exception):
- PortSettings(config={'name': 'foo'})
+ PortSettings(**{'name': 'foo'})
def test_name_netname_only(self):
settings = PortSettings(name='foo', network_name='bar')
self.assertIsNone(settings.device_id)
def test_config_with_name_netname_only(self):
- settings = PortSettings(config={'name': 'foo', 'network_name': 'bar'})
+ settings = PortSettings(**{'name': 'foo', 'network_name': 'bar'})
self.assertEqual('foo', settings.name)
self.assertEqual('bar', settings.network_name)
self.assertTrue(settings.admin_state_up)
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', device_owner='owner',
+ 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',
+ device_owner='owner',
device_id='device number')
self.assertEqual('foo', settings.name)
self.assertEqual('bar', settings.network_name)
fixed_ips = {'sub_id', '10.0.0.10'}
allowed_address_pairs = {'10.0.0.101', '1234.5678'}
- settings = PortSettings(config={'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', 'device_owner': 'owner', 'device_id': 'device number'})
+ 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', 'device_owner': 'owner',
+ 'device_id': 'device number'})
self.assertEqual('foo', settings.name)
self.assertEqual('bar', settings.network_name)
self.assertFalse(settings.admin_state_up)
if len(self.net_creator.get_subnets()) > 0:
# Validate subnet has been deleted
neutron_utils_tests.validate_subnet(
- self.neutron, self.net_creator.network_settings.subnet_settings[0].name,
- self.net_creator.network_settings.subnet_settings[0].cidr, False)
+ self.neutron,
+ self.net_creator.network_settings.subnet_settings[0].name,
+ self.net_creator.network_settings.subnet_settings[0].cidr,
+ False)
if self.net_creator.get_network():
# Validate network has been deleted
- neutron_utils_tests.validate_network(self.neutron, self.net_creator.network_settings.name,
- False)
+ neutron_utils_tests.validate_network(
+ self.neutron, self.net_creator.network_settings.name,
+ False)
self.net_creator.clean()
super(self.__class__, self).__clean__()
Tests the creation of an OpenStack network without a router.
"""
# Create Nework
- self.net_creator = OpenStackNetwork(self.os_creds, self.net_config.network_settings)
+ self.net_creator = OpenStackNetwork(self.os_creds,
+ self.net_config.network_settings)
self.net_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(self.neutron, self.net_creator.network_settings.name, True)
+ neutron_utils_tests.validate_network(
+ self.neutron, self.net_creator.network_settings.name, True)
# Validate subnets
neutron_utils_tests.validate_subnet(
- self.neutron, self.net_creator.network_settings.subnet_settings[0].name,
+ self.neutron,
+ self.net_creator.network_settings.subnet_settings[0].name,
self.net_creator.network_settings.subnet_settings[0].cidr, True)
def test_create_delete_network(self):
"""
- Tests the creation of an OpenStack network, it's deletion, then cleanup.
+ Tests the creation of an OpenStack network, it's deletion, then cleanup
"""
# Create Nework
- self.net_creator = OpenStackNetwork(self.os_creds, self.net_config.network_settings)
+ self.net_creator = OpenStackNetwork(self.os_creds,
+ self.net_config.network_settings)
self.net_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(self.neutron, self.net_creator.network_settings.name, True)
+ neutron_utils_tests.validate_network(
+ self.neutron, self.net_creator.network_settings.name, True)
- neutron_utils.delete_network(self.neutron, self.net_creator.get_network())
- self.assertIsNone(neutron_utils.get_network(self.neutron, self.net_creator.network_settings.name))
+ neutron_utils.delete_network(self.neutron,
+ self.net_creator.get_network())
+ self.assertIsNone(neutron_utils.get_network(
+ self.neutron, self.net_creator.network_settings.name))
# This shall not throw an exception here
self.net_creator.clean()
Tests the creation of an OpenStack network with a router.
"""
# Create Network
- self.net_creator = OpenStackNetwork(self.os_creds, self.net_config.network_settings)
+ self.net_creator = OpenStackNetwork(self.os_creds,
+ self.net_config.network_settings)
self.net_creator.create()
# Create Router
- self.router_creator = create_router.OpenStackRouter(self.os_creds, self.net_config.router_settings)
+ self.router_creator = create_router.OpenStackRouter(
+ self.os_creds, self.net_config.router_settings)
self.router_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(self.neutron, self.net_creator.network_settings.name, True)
+ neutron_utils_tests.validate_network(
+ self.neutron, self.net_creator.network_settings.name, True)
# Validate subnets
neutron_utils_tests.validate_subnet(
- self.neutron, self.net_creator.network_settings.subnet_settings[0].name,
+ self.neutron,
+ self.net_creator.network_settings.subnet_settings[0].name,
self.net_creator.network_settings.subnet_settings[0].cidr, True)
# Validate routers
- neutron_utils_tests.validate_router(self.neutron, self.router_creator.router_settings.name, True)
+ neutron_utils_tests.validate_router(
+ self.neutron, self.router_creator.router_settings.name, True)
- neutron_utils_tests.validate_interface_router(self.router_creator.get_internal_router_interface(),
- self.router_creator.get_router(),
- self.net_creator.get_subnets()[0])
+ neutron_utils_tests.validate_interface_router(
+ self.router_creator.get_internal_router_interface(),
+ self.router_creator.get_router(),
+ self.net_creator.get_subnets()[0])
def test_create_networks_same_name(self):
"""
- Tests the creation of an OpenStack network and ensures that the OpenStackNetwork object will not create
- a second.
+ Tests the creation of an OpenStack network and ensures that the
+ OpenStackNetwork object will not create a second.
"""
# Create Nework
- self.net_creator = OpenStackNetwork(self.os_creds, self.net_config.network_settings)
+ self.net_creator = OpenStackNetwork(self.os_creds,
+ self.net_config.network_settings)
self.net_creator.create()
- self.net_creator2 = OpenStackNetwork(self.os_creds, self.net_config.network_settings)
+ self.net_creator2 = OpenStackNetwork(self.os_creds,
+ self.net_config.network_settings)
self.net_creator2.create()
self.assertEqual(self.net_creator.get_network()['network']['id'],
class CreateNetworkTypeTests(OSComponentTestCase):
"""
- Test for the CreateNework class defined in create_nework.py for testing creating networks of different types
+ Test for the CreateNework class defined in create_nework.py for testing
+ creating networks of different types
"""
def setUp(self):
if len(self.net_creator.get_subnets()) > 0:
# Validate subnet has been deleted
neutron_utils_tests.validate_subnet(
- self.neutron, self.net_creator.network_settings.subnet_settings[0].name,
- self.net_creator.network_settings.subnet_settings[0].cidr, False)
+ self.neutron,
+ self.net_creator.network_settings.subnet_settings[0].name,
+ self.net_creator.network_settings.subnet_settings[0].cidr,
+ False)
if self.net_creator.get_network():
# Validate network has been deleted
- neutron_utils_tests.validate_network(self.neutron, self.net_creator.network_settings.name,
- False)
+ neutron_utils_tests.validate_network(
+ self.neutron, self.net_creator.network_settings.name,
+ False)
self.net_creator.clean()
+
# TODO - determine why this is not working on Newton
# - Unable to create the network. No tenant network is available for allocation.
# def test_create_network_type_vlan(self):
"""
# Create Network
network_type = 'vxlan'
- net_settings = NetworkSettings(name=self.net_config.network_settings.name,
- subnet_settings=self.net_config.network_settings.subnet_settings,
- network_type=network_type)
+ net_settings = NetworkSettings(
+ name=self.net_config.network_settings.name,
+ subnet_settings=self.net_config.network_settings.subnet_settings,
+ network_type=network_type)
# When setting the network_type, creds must be admin
self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
network = self.net_creator.create()
# Validate network was created
- neutron_utils_tests.validate_network(self.neutron, net_settings.name, True)
+ neutron_utils_tests.validate_network(self.neutron, net_settings.name,
+ True)
- self.assertEqual(network_type, network['network']['provider:network_type'])
+ self.assertEqual(network_type,
+ network['network']['provider:network_type'])
# TODO - determine what value we need to place into physical_network
# - Do not know what vaule to place into the 'physical_network' setting.
def test_create_network_type_foo(self):
"""
- Tests the creation of an OpenStack network of type foo which should raise an exception.
+ Tests the creation of an OpenStack network of type foo which should
+ raise an exception.
"""
# Create Network
network_type = 'foo'
- net_settings = NetworkSettings(name=self.net_config.network_settings.name,
- subnet_settings=self.net_config.network_settings.subnet_settings,
- network_type=network_type)
+ net_settings = NetworkSettings(
+ name=self.net_config.network_settings.name,
+ subnet_settings=self.net_config.network_settings.subnet_settings,
+ network_type=network_type)
self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
with self.assertRaises(Exception):
self.net_creator.create()
Code Review / snaps.git / commitdiff