+---------------------------------------+---------------+-----------------------------------------------------------+
| Test Name | API | Description |
+=======================================+===============+===========================================================+
-| test_derive_net_settings_no_subnet | Neutron 2 | Tests to ensure that derived NetworkSettings from an |
+| test_derive_net_settings_no_subnet | Neutron 2 | Tests to ensure that derived NetworkConfig from an |
| | | OpenStack network are correct without a subnet |
+---------------------------------------+---------------+-----------------------------------------------------------+
-| test_derive_net_settings_two_subnets | Neutron 2 | Tests to ensure that derived NetworkSettings from an |
+| test_derive_net_settings_two_subnets | Neutron 2 | Tests to ensure that derived NetworkConfig from an |
| | | OpenStack network are correct with two subnets |
+---------------------------------------+---------------+-----------------------------------------------------------+
- Network - snaps.openstack.create\_network.OpenStackNetwork
- - snaps.openstack.create\_network.NetworkSettings
+ - snaps.config_network.NetworkConfig
- name - the name of the network (required)
- admin\_state\_up - flag denoting the administrative status of
- segmentation\_id - the id of the segmentation (required
when network\_type is 'vlan')
- subnet\_settings (list of optional
- snaps.openstack.create\_network.SubnetSettings objects)
+ snaps.config.network.SubnetConfig objects)
- cidr - the subnet's CIDR (required)
- ip\_version - 4 or 6 (default=4)
.. code:: python
- from snaps.openstack.create_network import NetworkSettings, SubnetSettings, OpenStackNetwork
+ from snaps.config.network import NetworkConfig, SubnetConfig
+ from snaps.openstack.create_network import OpenStackNetwork
- subnet_settings = SubnetSettings(name='subnet-name', cidr='10.0.0.0/24')
- network_settings = NetworkSettings(name='network-name', subnet_settings=[subnet_settings])
+ subnet_settings = SubnetConfig(name='subnet-name', cidr='10.0.0.0/24')
+ network_settings = NetworkConfig(name='network-name', subnet_settings=[subnet_settings])
network_creator = OpenStackNetwork(os_creds, network_settings)
network_creator.create()
.. code:: python
+ from snaps.config.network import SubnetConfig
+ from snaps.config.rule import RuleConfig
from snaps.openstack.create_security_group import SecurityGroupSettings, SecurityGroupRuleSettings, Direction, OpenStackSecurityGroup
- rule_settings = SubnetSettings(name='subnet-name', cidr='10.0.0.0/24')
- network_settings = NetworkSettings(name='network-name', subnet_settings=[subnet_settings])
+ rule_settings = RuleConfig(name='subnet-name', cidr='10.0.0.0/24')
+ network_settings = SubnetConfig(name='network-name', subnet_settings=[subnet_settings])
sec_grp_name = 'sec-grp-name'
rule_settings = SecurityGroupRuleSettings(name=sec_grp_name, direction=Direction.ingress)
- internal\_subnets - list of subnet names to which this router
will connect (optional)
- port\_settings (list of optional
- snaps.openstack.create\_router.PortSettings objects) - creates
+ snaps.config.network.PortConfig objects) - creates
custom ports to internal subnets (similar to internal\_subnets
with more control)
- name - the name of the VM (required)
- flavor - the name of the flavor (required)
- port\_settings - list of
- snaps.openstack.create\_network.PortSettings objects where each
+ snaps.config.network.PortConfig objects where each
denote a NIC (see above in create router section for details)
API does not require, but newer NFVIs now require VMs have at
least one network
.. code:: python
from snaps.openstack.create_instance import VmInstanceSettings, FloatingIpSettings, OpenStackVmInstance
- from snaps.openstack.create_network import PortSettings
+ from snaps.config.network import PortConfig
- port_settings = PortSettings(name='port-name', network_name=network_settings.name)
+ port_settings = PortConfig(name='port-name', network_name=network_settings.name)
floating_ip_settings = FloatingIpSettings(name='fip1', port_name=port_settings.name, router_name=router_settings.name)
instance_settings = VmInstanceSettings(name='vm-name', flavor='flavor_settings.name', port_settings=[port_settings],
floating_ip_settings=[floating_ip_settings])
Ensures that all required members are included when constructing a
Role domain object
+NetworkConfigUnitTests
+----------------------
+
+Ensures that all required members are included when constructing a
+NetworkConfig object
+
NetworkSettingsUnitTests
------------------------
Ensures that all required members are included when constructing a
-NetworkSettings object
+deprecated NetworkSettings object
NetworkObjectTests
------------------
Ensures that all required members are included when constructing a
Network domain object
+SubnetConfigUnitTests
+---------------------
+
+Ensures that all required members are included when constructing a
+SubnetConfig object
+
SubnetSettingsUnitTests
-----------------------
Ensures that all required members are included when constructing a
-SubnetSettings object
+deprecated SubnetSettings object
SubnetObjectTests
-----------------
Ensures that all required members are included when constructing a
Subnet domain object
+PortConfigUnitTests
+-------------------
+
+Ensures that all required members are included when constructing a
+PortConfig object
+
PortSettingsUnitTests
---------------------
Ensures that all required members are included when constructing a
-PortSettings object
+deprecated PortSettings object
PortDomainObjectTests
---------------------
from snaps.openstack.os_credentials import OSCreds, ProxySettings
-proxy_settings = ProxySettings(host='10.197.123.27', port='3128',
- ssh_proxy_cmd='/usr/local/bin/corkscrew 10.197.123.27 3128 %h %p')
+proxy_settings = ProxySettings(
+ host='10.197.123.27', port='3128',
+ ssh_proxy_cmd='/usr/local/bin/corkscrew 10.197.123.27 3128 %h %p')
-os_creds = OSCreds(username='admin', password='cable123', auth_url='http://192.168.67.10:5000/v2.0/',
- project_name='admin', proxy_settings=proxy_settings)
+os_creds = OSCreds(
+ username='admin', password='cable123',
+ auth_url='http://192.168.67.10:5000/v2.0/', project_name='admin',
+ proxy_settings=proxy_settings)
# Images
# Network
-from snaps.openstack.create_network import NetworkSettings, SubnetSettings, OpenStackNetwork
+from snaps.config.network import NetworkConfig, SubnetConfig
+from snaps.openstack.create_network import OpenStackNetwork
-subnet_settings = SubnetSettings(name='test-subnet', cidr='10.0.0.1/24')
-network_settings = NetworkSettings(name='test-net', subnet_settings=[subnet_settings])
+subnet_settings = SubnetConfig(name='test-subnet', cidr='10.0.0.1/24')
+network_settings = NetworkConfig(
+ name='test-net', subnet_settings=[subnet_settings])
network = OpenStackNetwork(os_creds, network_settings)
network.create()
flavor.create()
# Instances
-from snaps.openstack.create_network import PortSettings
-from snaps.openstack.create_instance import VmInstanceSettings, OpenStackVmInstance
-
-port_settings = PortSettings(name='test-port', network_name=network_settings.name)
-instance_settings = VmInstanceSettings(name='test-inst', flavor=flavor_settings.name, port_settings=[port_settings])
+from snaps.config.network import PortConfig
+from snaps.openstack.create_instance import (
+ VmInstanceSettings, OpenStackVmInstance)
+
+port_settings = PortConfig(
+ name='test-port', network_name=network_settings.name)
+instance_settings = VmInstanceSettings(
+ name='test-inst', flavor=flavor_settings.name,
+ port_settings=[port_settings])
vm_inst = OpenStackVmInstance(os_creds, instance_settings, image_settings)
vm_inst.create(block=True)
from snaps.config.flavor import FlavorConfig
from snaps.config.image import ImageConfig
from snaps.config.keypair import KeypairConfig
+from snaps.config.network import PortConfig, NetworkConfig
from snaps.config.project import ProjectConfig
from snaps.config.qos import QoSConfig
from snaps.config.router import RouterConfig
from snaps.openstack.create_image import OpenStackImage
from snaps.openstack.create_instance import VmInstanceSettings
from snaps.openstack.create_keypairs import OpenStackKeypair
-from snaps.openstack.create_network import (
- PortSettings, NetworkSettings, OpenStackNetwork)
+from snaps.openstack.create_network import OpenStackNetwork
from snaps.openstack.create_project import OpenStackProject
from snaps.openstack.create_qos import OpenStackQoS
from snaps.openstack.create_router import OpenStackRouter
"""
out = list()
for port_config in config:
- out.append(PortSettings(**port_config.get('port')))
+ out.append(PortConfig(**port_config.get('port')))
return out
# Create networks
creators.append(__create_instances(
- os_creds_dict, OpenStackNetwork, NetworkSettings,
+ os_creds_dict, OpenStackNetwork, NetworkConfig,
os_config.get('networks'), 'network', clean, users_dict))
# Create routers
--- /dev/null
+# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
+# and others. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at:
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import enum
+
+from snaps.openstack.utils import keystone_utils, neutron_utils
+
+
+class NetworkConfig(object):
+ """
+ Class representing a network configuration
+ """
+
+ def __init__(self, **kwargs):
+ """
+ Constructor - all parameters are optional
+ :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 network_type: the type of network (i.e. vlan|flat).
+ :param physical_network: the name of the physical network
+ (required when network_type is 'flat')
+ :param segmentation_id: the id of the segmentation
+ (this is required when network_type is 'vlan')
+ :param subnets or subnet_settings: List of SubnetConfig objects.
+ :return:
+ """
+
+ self.project_id = None
+
+ 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
+
+ if kwargs.get('shared') is not None:
+ self.shared = bool(kwargs['shared'])
+ else:
+ self.shared = None
+
+ self.project_name = kwargs.get('project_name')
+
+ if kwargs.get('external') is not None:
+ self.external = bool(kwargs.get('external'))
+ else:
+ self.external = False
+
+ self.network_type = kwargs.get('network_type')
+ self.physical_network = kwargs.get('physical_network')
+ self.segmentation_id = kwargs.get('segmentation_id')
+
+ self.subnet_settings = list()
+ subnet_settings = kwargs.get('subnets')
+ if not subnet_settings:
+ subnet_settings = kwargs.get('subnet_settings', list())
+ if subnet_settings:
+ for subnet_config in subnet_settings:
+ if isinstance(subnet_config, SubnetConfig):
+ self.subnet_settings.append(subnet_config)
+ else:
+ self.subnet_settings.append(
+ SubnetConfig(**subnet_config['subnet']))
+
+ if not self.name or len(self.name) < 1:
+ raise NetworkConfigError('Name required for networks')
+
+ def get_project_id(self, os_creds):
+ """
+ Returns the project ID for a given project_name or None
+ :param os_creds: the credentials required for keystone client retrieval
+ :return: the ID or None
+ """
+ if self.project_id:
+ return self.project_id
+ else:
+ if self.project_name:
+ keystone = keystone_utils.keystone_client(os_creds)
+ project = keystone_utils.get_project(
+ keystone=keystone, project_name=self.project_name)
+ if project:
+ return project.id
+
+ return None
+
+ 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
+ TODO - expand automated testing to exercise all parameters
+
+ :param os_creds: the OpenStack credentials
+ :return: the dictionary object
+ """
+ out = dict()
+
+ if self.name:
+ out['name'] = self.name
+ if self.admin_state_up is not None:
+ out['admin_state_up'] = self.admin_state_up
+ if self.shared:
+ out['shared'] = self.shared
+ if self.project_name:
+ project_id = self.get_project_id(os_creds)
+ if project_id:
+ out['tenant_id'] = project_id
+ else:
+ raise NetworkConfigError(
+ '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:
+ out['provider:physical_network'] = self.physical_network
+ if self.segmentation_id:
+ out['provider:segmentation_id'] = self.segmentation_id
+ if self.external:
+ out['router:external'] = self.external
+ return {'network': out}
+
+
+class NetworkConfigError(Exception):
+ """
+ Exception to be thrown when networks settings attributes are incorrect
+ """
+
+
+class IPv6Mode(enum.Enum):
+ """
+ A rule's direction
+ """
+ slaac = 'slaac'
+ stateful = 'dhcpv6-stateful'
+ stateless = 'dhcpv6-stateless'
+
+
+class SubnetConfig(object):
+ """
+ Class representing a subnet configuration
+ """
+
+ def __init__(self, **kwargs):
+ """
+ Constructor - all parameters are optional except cidr (subnet mask)
+ :param name: The subnet name (required)
+ :param cidr: The CIDR (required)
+ :param ip_version: The IP version, which is 4 or 6 (required)
+ :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 (optional)
+ :param start: The start address for the allocation pools (optional)
+ :param end: The end address for the allocation pools (optional)
+ :param gateway_ip: The gateway IP address (optional)
+ :param enable_dhcp: Set to true if DHCP is enabled and false if DHCP is
+ disabled (optional)
+ :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]
+ (default '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",
+ "nexthop":"123.456.78.9"
+ },
+ {
+ "destination":"192.168.0.0/24",
+ "nexthop":"192.168.0.1"
+ }
+ ]
+ :param destination: The destination for static route (optional)
+ :param nexthop: The next hop for the destination (optional)
+ :param ipv6_ra_mode: an instance of the IPv6Mode enum
+ (optional when enable_dhcp is True)
+ :param ipv6_address_mode: an instance of the IPv6Mode enum
+ (optional when enable_dhcp is True)
+ :raise: SubnetConfigError when config does not have or cidr values
+ are None
+ """
+ self.cidr = kwargs.get('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 'dns_nameservers' in kwargs:
+ self.dns_nameservers = kwargs.get('dns_nameservers')
+ else:
+ if self.ip_version == 4:
+ self.dns_nameservers = ['8.8.8.8']
+ else:
+ self.dns_nameservers = list()
+
+ self.host_routes = kwargs.get('host_routes')
+ self.destination = kwargs.get('destination')
+ self.nexthop = kwargs.get('nexthop')
+ self.ipv6_ra_mode = map_mode(kwargs.get('ipv6_ra_mode'))
+ self.ipv6_address_mode = map_mode(kwargs.get('ipv6_address_mode'))
+
+ if not self.name or not self.cidr:
+ raise SubnetConfigError('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
+ :param os_creds: the OpenStack credentials
+ :param network: The network object on which the subnet will be created
+ (optional)
+ :return: the dictionary object
+ """
+ out = {
+ 'cidr': self.cidr,
+ 'ip_version': self.ip_version,
+ }
+
+ if network:
+ out['network_id'] = network.id
+ if self.name:
+ out['name'] = self.name
+ if self.project_name:
+ keystone = keystone_utils.keystone_client(os_creds)
+ project = keystone_utils.get_project(
+ keystone=keystone, project_name=self.project_name)
+ project_id = None
+ if project:
+ project_id = project.id
+ if project_id:
+ out['tenant_id'] = project_id
+ else:
+ raise SubnetConfigError(
+ '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:
+ out['gateway_ip'] = self.gateway_ip
+ if self.enable_dhcp is not None:
+ out['enable_dhcp'] = self.enable_dhcp
+ if self.dns_nameservers and len(self.dns_nameservers) > 0:
+ out['dns_nameservers'] = self.dns_nameservers
+ if self.host_routes and len(self.host_routes) > 0:
+ out['host_routes'] = self.host_routes
+ if self.destination:
+ out['destination'] = self.destination
+ if self.nexthop:
+ out['nexthop'] = self.nexthop
+ if self.ipv6_ra_mode:
+ out['ipv6_ra_mode'] = self.ipv6_ra_mode.value
+ if self.ipv6_address_mode:
+ out['ipv6_address_mode'] = self.ipv6_address_mode.value
+ return out
+
+
+def map_mode(mode):
+ """
+ Takes a the direction value maps it to the Direction enum. When None return
+ None
+ :param mode: the mode value
+ :return: the IPv6Mode enum object
+ :raise: SubnetConfigError if value is invalid
+ """
+ if not mode:
+ return None
+ if isinstance(mode, IPv6Mode):
+ return mode
+ elif isinstance(mode, str):
+ mode_str = str(mode)
+ if mode_str == 'slaac':
+ return IPv6Mode.slaac
+ elif mode_str == 'dhcpv6-stateful':
+ return IPv6Mode.stateful
+ elif mode_str == 'stateful':
+ return IPv6Mode.stateful
+ elif mode_str == 'dhcpv6-stateless':
+ return IPv6Mode.stateless
+ elif mode_str == 'stateless':
+ return IPv6Mode.stateless
+ else:
+ raise SubnetConfigError('Invalid mode - ' + mode_str)
+ else:
+ return map_mode(mode.value)
+
+
+class SubnetConfigError(Exception):
+ """
+ Exception to be thrown when subnet settings attributes are incorrect
+ """
+
+
+class PortConfig(object):
+ """
+ Class representing a port configuration
+ """
+
+ def __init__(self, **kwargs):
+ """
+ Constructor
+ :param name: A symbolic name for the port (optional).
+ :param network_name: The name of the network on which to create the
+ port (required).
+ :param admin_state_up: A boolean value denoting the administrative
+ status of the port (default = True)
+ :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 (optional)
+ :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 (optional)
+ :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 (optional)
+ :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 (optional)
+ :param opt_value: The extra DHCP option value (optional)
+ :param opt_name: The extra DHCP option name (optional)
+ :param device_owner: The ID of the entity that uses this port.
+ For example, a DHCP agent (optional)
+ :param device_id: The ID of the device that uses this port.
+ For example, a virtual server (optional)
+ :return:
+ """
+ if 'port' in kwargs:
+ kwargs = kwargs['port']
+
+ self.name = kwargs.get('name')
+ self.network_name = kwargs.get('network_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.project_name = kwargs.get('project_name')
+ self.mac_address = kwargs.get('mac_address')
+ self.ip_addrs = kwargs.get('ip_addrs')
+ 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.network_name:
+ raise PortConfigError(
+ 'The attribute network_name is required')
+
+ def __get_fixed_ips(self, neutron):
+ """
+ Sets the self.fixed_ips value
+ :param neutron: the Neutron client
+ :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})
+ else:
+ raise PortConfigError(
+ '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.
+ 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
+ :param os_creds: the OpenStack credentials
+ :return: the dictionary object
+ """
+
+ out = dict()
+
+ project_id = None
+ if self.project_name:
+ keystone = keystone_utils.keystone_client(os_creds)
+ project = keystone_utils.get_project(
+ keystone=keystone, project_name=self.project_name)
+ if project:
+ project_id = project.id
+
+ network = neutron_utils.get_network(
+ neutron, network_name=self.network_name, project_id=project_id)
+ if not network:
+ raise PortConfigError(
+ 'Cannot locate network with name - ' + self.network_name)
+
+ out['network_id'] = network.id
+
+ if self.admin_state_up is not None:
+ out['admin_state_up'] = self.admin_state_up
+ if self.name:
+ out['name'] = self.name
+ if self.project_name:
+ if project_id:
+ out['tenant_id'] = project_id
+ else:
+ raise PortConfigError(
+ 'Could not find project ID for project named - ' +
+ self.project_name)
+ if self.mac_address:
+ out['mac_address'] = self.mac_address
+
+ 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:
+ out['allowed_address_pairs'] = self.allowed_address_pairs
+ if self.opt_value:
+ out['opt_value'] = self.opt_value
+ if self.opt_name:
+ out['opt_name'] = self.opt_name
+ if self.device_owner:
+ out['device_owner'] = self.device_owner
+ if self.device_id:
+ out['device_id'] = self.device_id
+ return {'port': out}
+
+ def __eq__(self, other):
+ return (self.name == other.name and
+ self.network_name == other.network_name and
+ self.admin_state_up == other.admin_state_up and
+ 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.security_groups == other.security_groups and
+ self.allowed_address_pairs == other.allowed_address_pairs and
+ self.opt_value == other.opt_value and
+ self.opt_name == other.opt_name and
+ self.device_owner == other.device_owner and
+ self.device_id == other.device_id)
+
+
+class PortConfigError(Exception):
+ """
+ Exception to be thrown when port settings attributes are incorrect
+ """
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-from snaps.openstack.create_network import PortSettings
+from snaps.config.network import PortConfig
from snaps.openstack.utils import neutron_utils, keystone_utils
True = up / False = down (default True)
:param internal_subnets: List of subnet names to which to connect this
router for Floating IP purposes
- :param port_settings: List of PortSettings objects
+ :param port_settings: List of PortConfig objects
:return:
"""
self.name = kwargs.get('name')
if kwargs.get('interfaces', kwargs.get('port_settings')):
interfaces = kwargs.get('interfaces', kwargs.get('port_settings'))
for interface in interfaces:
- if isinstance(interface, PortSettings):
+ if isinstance(interface, PortConfig):
self.port_settings.append(interface)
else:
self.port_settings.append(
- PortSettings(**interface['port']))
+ PortConfig(**interface['port']))
if not self.name:
raise RouterConfigError('Name is required')
--- /dev/null
+# Copyright (c) 2017 Cable Television Laboratories, Inc. ("CableLabs")
+# and others. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at:
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+import unittest
+
+from snaps.config.network import (
+ NetworkConfigError, NetworkConfig, SubnetConfig, SubnetConfigError,
+ IPv6Mode, PortConfig, PortConfigError)
+
+
+class NetworkConfigUnitTests(unittest.TestCase):
+ """
+ Tests the construction of the NetworkConfig class
+ """
+
+ def test_no_params(self):
+ with self.assertRaises(NetworkConfigError):
+ NetworkConfig()
+
+ def test_empty_config(self):
+ with self.assertRaises(NetworkConfigError):
+ NetworkConfig(**dict())
+
+ def test_name_only(self):
+ settings = NetworkConfig(name='foo')
+ self.assertEqual('foo', settings.name)
+ self.assertTrue(settings.admin_state_up)
+ self.assertIsNone(settings.shared)
+ self.assertIsNone(settings.project_name)
+ self.assertFalse(settings.external)
+ self.assertIsNone(settings.network_type)
+ self.assertIsNone(settings.segmentation_id)
+ self.assertEqual(0, len(settings.subnet_settings))
+
+ def test_config_with_name_only(self):
+ settings = NetworkConfig(**{'name': 'foo'})
+ self.assertEqual('foo', settings.name)
+ self.assertTrue(settings.admin_state_up)
+ self.assertIsNone(settings.shared)
+ self.assertIsNone(settings.project_name)
+ self.assertFalse(settings.external)
+ self.assertIsNone(settings.network_type)
+ self.assertIsNone(settings.segmentation_id)
+ self.assertEqual(0, len(settings.subnet_settings))
+
+ def test_all(self):
+ sub_settings = SubnetConfig(name='foo-subnet', cidr='10.0.0.0/24')
+ settings = NetworkConfig(
+ name='foo', admin_state_up=False, shared=True, project_name='bar',
+ external=True, network_type='vlan', physical_network='phy',
+ segmentation_id=2366, subnet_settings=[sub_settings])
+ self.assertEqual('foo', settings.name)
+ self.assertFalse(settings.admin_state_up)
+ self.assertTrue(settings.shared)
+ self.assertEqual('bar', settings.project_name)
+ self.assertTrue(settings.external)
+ self.assertEqual('vlan', settings.network_type)
+ self.assertEqual('phy', settings.physical_network)
+ self.assertEqual(2366, settings.segmentation_id)
+ self.assertEqual(1, len(settings.subnet_settings))
+ self.assertEqual('foo-subnet', settings.subnet_settings[0].name)
+
+ def test_config_all(self):
+ settings = NetworkConfig(
+ **{'name': 'foo', 'admin_state_up': False, 'shared': True,
+ 'project_name': 'bar', 'external': True, 'network_type': 'vlan',
+ 'physical_network': 'phy',
+ 'segmentation_id': 2366,
+ '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)
+ self.assertEqual('bar', settings.project_name)
+ self.assertTrue(settings.external)
+ self.assertEqual('vlan', settings.network_type)
+ self.assertEqual('phy', settings.physical_network)
+ self.assertEqual(2366, settings.segmentation_id)
+ self.assertEqual(1, len(settings.subnet_settings))
+ self.assertEqual('foo-subnet', settings.subnet_settings[0].name)
+
+
+class SubnetConfigUnitTests(unittest.TestCase):
+ """
+ Tests the construction of the SubnetConfig class
+ """
+
+ def test_no_params(self):
+ with self.assertRaises(SubnetConfigError):
+ SubnetConfig()
+
+ def test_empty_config(self):
+ with self.assertRaises(SubnetConfigError):
+ SubnetConfig(**dict())
+
+ def test_name_only(self):
+ with self.assertRaises(SubnetConfigError):
+ SubnetConfig(name='foo')
+
+ def test_config_with_name_only(self):
+ with self.assertRaises(SubnetConfigError):
+ SubnetConfig(**{'name': 'foo'})
+
+ def test_name_cidr_only(self):
+ settings = SubnetConfig(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)
+ self.assertIsNone(settings.project_name)
+ self.assertIsNone(settings.start)
+ self.assertIsNone(settings.end)
+ self.assertIsNone(settings.enable_dhcp)
+ self.assertEqual(1, len(settings.dns_nameservers))
+ self.assertEqual('8.8.8.8', settings.dns_nameservers[0])
+ self.assertIsNone(settings.host_routes)
+ self.assertIsNone(settings.destination)
+ self.assertIsNone(settings.nexthop)
+ self.assertIsNone(settings.ipv6_ra_mode)
+ self.assertIsNone(settings.ipv6_address_mode)
+
+ def test_config_with_name_cidr_only(self):
+ settings = SubnetConfig(**{'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)
+ self.assertIsNone(settings.project_name)
+ self.assertIsNone(settings.start)
+ self.assertIsNone(settings.end)
+ self.assertIsNone(settings.gateway_ip)
+ self.assertIsNone(settings.enable_dhcp)
+ self.assertEqual(1, len(settings.dns_nameservers))
+ self.assertEqual('8.8.8.8', settings.dns_nameservers[0])
+ self.assertIsNone(settings.host_routes)
+ self.assertIsNone(settings.destination)
+ self.assertIsNone(settings.nexthop)
+ self.assertIsNone(settings.ipv6_ra_mode)
+ self.assertIsNone(settings.ipv6_address_mode)
+
+ def test_all_string_enums(self):
+ host_routes = {'destination': '0.0.0.0/0', 'nexthop': '123.456.78.9'}
+ settings = SubnetConfig(
+ 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)
+ self.assertEqual('bar-project', settings.project_name)
+ self.assertEqual('10.0.0.2', settings.start)
+ self.assertEqual('10.0.0.101', settings.end)
+ self.assertEqual('10.0.0.1', settings.gateway_ip)
+ self.assertEqual(False, settings.enable_dhcp)
+ self.assertEqual(1, len(settings.dns_nameservers))
+ self.assertEqual('8.8.8.8', settings.dns_nameservers[0])
+ self.assertEqual(1, len(settings.host_routes))
+ self.assertEqual(host_routes, settings.host_routes[0])
+ self.assertEqual('dest', settings.destination)
+ self.assertEqual('hop', settings.nexthop)
+ self.assertEqual(IPv6Mode.stateful, settings.ipv6_ra_mode)
+ self.assertEqual(IPv6Mode.slaac, settings.ipv6_address_mode)
+
+ def test_all_type_enums(self):
+ host_routes = {'destination': '0.0.0.0/0', 'nexthop': '123.456.78.9'}
+ settings = SubnetConfig(
+ 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=IPv6Mode.stateful,
+ ipv6_address_mode=IPv6Mode.slaac)
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('10.0.0.0/24', settings.cidr)
+ self.assertEqual(6, settings.ip_version)
+ self.assertEqual('bar-project', settings.project_name)
+ self.assertEqual('10.0.0.2', settings.start)
+ self.assertEqual('10.0.0.101', settings.end)
+ self.assertEqual('10.0.0.1', settings.gateway_ip)
+ self.assertEqual(False, settings.enable_dhcp)
+ self.assertEqual(1, len(settings.dns_nameservers))
+ self.assertEqual('8.8.8.8', settings.dns_nameservers[0])
+ self.assertEqual(1, len(settings.host_routes))
+ self.assertEqual(host_routes, settings.host_routes[0])
+ self.assertEqual('dest', settings.destination)
+ self.assertEqual('hop', settings.nexthop)
+ self.assertEqual(IPv6Mode.stateful, settings.ipv6_ra_mode)
+ self.assertEqual(IPv6Mode.slaac, settings.ipv6_address_mode)
+
+ def test_config_all(self):
+ host_routes = {'destination': '0.0.0.0/0', 'nexthop': '123.456.78.9'}
+ settings = SubnetConfig(
+ **{'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-stateless',
+ 'ipv6_address_mode': 'slaac'})
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('10.0.0.0/24', settings.cidr)
+ self.assertEqual(6, settings.ip_version)
+ self.assertEqual('bar-project', settings.project_name)
+ self.assertEqual('10.0.0.2', settings.start)
+ self.assertEqual('10.0.0.101', settings.end)
+ self.assertEqual('10.0.0.1', settings.gateway_ip)
+ self.assertEqual(False, settings.enable_dhcp)
+ self.assertEqual(1, len(settings.dns_nameservers))
+ self.assertEqual('8.8.8.8', settings.dns_nameservers[0])
+ self.assertEqual(1, len(settings.host_routes))
+ self.assertEqual(host_routes, settings.host_routes[0])
+ self.assertEqual('dest', settings.destination)
+ self.assertEqual('hop', settings.nexthop)
+ self.assertEqual(IPv6Mode.stateless, settings.ipv6_ra_mode)
+ self.assertEqual(IPv6Mode.slaac, settings.ipv6_address_mode)
+
+
+class PortConfigUnitTests(unittest.TestCase):
+ """
+ Tests the construction of the PortConfig class
+ """
+
+ def test_no_params(self):
+ with self.assertRaises(PortConfigError):
+ PortConfig()
+
+ def test_empty_config(self):
+ with self.assertRaises(PortConfigError):
+ PortConfig(**dict())
+
+ def test_name_only(self):
+ with self.assertRaises(PortConfigError):
+ PortConfig(name='foo')
+
+ def test_config_name_only(self):
+ with self.assertRaises(PortConfigError):
+ PortConfig(**{'name': 'foo'})
+
+ def test_name_netname_only(self):
+ settings = PortConfig(name='foo', network_name='bar')
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('bar', settings.network_name)
+ self.assertTrue(settings.admin_state_up)
+ self.assertIsNone(settings.project_name)
+ self.assertIsNone(settings.mac_address)
+ self.assertIsNone(settings.ip_addrs)
+ self.assertIsNone(settings.security_groups)
+ self.assertIsNone(settings.allowed_address_pairs)
+ self.assertIsNone(settings.opt_value)
+ self.assertIsNone(settings.opt_name)
+ self.assertIsNone(settings.device_owner)
+ self.assertIsNone(settings.device_id)
+
+ def test_config_with_name_netname_only(self):
+ settings = PortConfig(**{'name': 'foo', 'network_name': 'bar'})
+ self.assertEqual('foo', settings.name)
+ self.assertEqual('bar', settings.network_name)
+ self.assertTrue(settings.admin_state_up)
+ self.assertIsNone(settings.project_name)
+ self.assertIsNone(settings.mac_address)
+ self.assertIsNone(settings.ip_addrs)
+ self.assertIsNone(settings.security_groups)
+ self.assertIsNone(settings.allowed_address_pairs)
+ self.assertIsNone(settings.opt_value)
+ self.assertIsNone(settings.opt_name)
+ self.assertIsNone(settings.device_owner)
+ self.assertIsNone(settings.device_id)
+
+ def test_all(self):
+ ip_addrs = [{'subnet_name', 'foo-sub', 'ip', '10.0.0.10'}]
+ allowed_address_pairs = {'10.0.0.101', '1234.5678'}
+
+ settings = PortConfig(
+ name='foo', network_name='bar', admin_state_up=False,
+ project_name='foo-project', mac_address='1234', 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')
+ 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(1, len(settings.security_groups))
+ self.assertEqual('foo_grp_id', settings.security_groups[0])
+ self.assertEqual(allowed_address_pairs, settings.allowed_address_pairs)
+ self.assertEqual('opt value', settings.opt_value)
+ self.assertEqual('opt name', settings.opt_name)
+ self.assertEqual('owner', settings.device_owner)
+ self.assertEqual('device number', settings.device_id)
+
+ def test_config_all(self):
+ ip_addrs = [{'subnet_name', 'foo-sub', 'ip', '10.0.0.10'}]
+ allowed_address_pairs = {'10.0.0.101', '1234.5678'}
+
+ settings = PortConfig(
+ **{'name': 'foo', 'network_name': 'bar', 'admin_state_up': False,
+ 'project_name': 'foo-project', 'mac_address': '1234',
+ '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'})
+ 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(1, len(settings.security_groups))
+ self.assertEqual('foo_grp_id', settings.security_groups[0])
+ self.assertEqual(allowed_address_pairs, settings.allowed_address_pairs)
+ self.assertEqual('opt value', settings.opt_value)
+ self.assertEqual('opt name', settings.opt_name)
+ self.assertEqual('owner', settings.device_owner)
+ self.assertEqual('device number', settings.device_id)
# limitations under the License.
import unittest
+from snaps.config.network import PortConfig
from snaps.config.router import RouterConfig, RouterConfigError
-from snaps.openstack.create_network import PortSettings
class RouterConfigUnitTests(unittest.TestCase):
self.assertEqual(0, len(settings.port_settings))
def test_all(self):
- port_settings = PortSettings(name='foo', network_name='bar')
+ port_settings = PortConfig(name='foo', network_name='bar')
settings = RouterConfig(
name='foo', project_name='bar', external_gateway='foo_gateway',
admin_state_up=True, enable_snat=False,
self.assertTrue(isinstance(settings.internal_subnets, list))
self.assertEqual(1, len(settings.internal_subnets))
self.assertEqual(['10.0.0.1/24'], settings.internal_subnets)
- self.assertEqual([PortSettings(**{'name': 'foo-port',
- 'network_name': 'bar-net'})],
+ self.assertEqual([PortConfig(**{'name': 'foo-port',
+ 'network_name': 'bar-net'})],
settings.port_settings)
from neutronclient.common.exceptions import PortNotFoundClient
from novaclient.exceptions import NotFound, BadRequest
-from snaps.openstack.create_network import PortSettings
+from snaps.config.network import PortConfig
from snaps.openstack.openstack_creator import OpenStackComputeObject
from snaps.openstack.utils import glance_utils, cinder_utils
from snaps.openstack.utils import neutron_utils
if port_settings:
for port_setting in port_settings:
if isinstance(port_setting, dict):
- self.port_settings.append(PortSettings(**port_setting))
- elif isinstance(port_setting, PortSettings):
+ self.port_settings.append(PortConfig(**port_setting))
+ elif isinstance(port_setting, PortConfig):
self.port_settings.append(port_setting)
if kwargs.get('security_group_names'):
import enum
from neutronclient.common.exceptions import NetworkNotFoundClient
+from snaps.config.network import NetworkConfig, SubnetConfig, PortConfig
from snaps.openstack.openstack_creator import OpenStackNetworkObject
-from snaps.openstack.utils import keystone_utils, neutron_utils
+from snaps.openstack.utils import neutron_utils
__author__ = 'spisarski'
return self.__network
-class NetworkSettings:
+class NetworkSettings(NetworkConfig):
"""
- Class representing a network configuration
+ Class to hold the configuration settings required for creating OpenStack
+ Network objects
+ deprecated
"""
def __init__(self, **kwargs):
- """
- Constructor - all parameters are optional
- :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 network_type: the type of network (i.e. vlan|flat).
- :param physical_network: the name of the physical network
- (required when network_type is 'flat')
- :param segmentation_id: the id of the segmentation
- (this is required when network_type is 'vlan')
- :param subnets or subnet_settings: List of SubnetSettings objects.
- :return:
- """
-
- self.project_id = None
-
- 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
-
- if kwargs.get('shared') is not None:
- self.shared = bool(kwargs['shared'])
- else:
- self.shared = None
-
- self.project_name = kwargs.get('project_name')
-
- if kwargs.get('external') is not None:
- self.external = bool(kwargs.get('external'))
- else:
- self.external = False
-
- self.network_type = kwargs.get('network_type')
- self.physical_network = kwargs.get('physical_network')
- self.segmentation_id = kwargs.get('segmentation_id')
-
- self.subnet_settings = list()
- subnet_settings = kwargs.get('subnets')
- if not subnet_settings:
- subnet_settings = kwargs.get('subnet_settings', list())
- 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 NetworkSettingsError('Name required for networks')
-
- def get_project_id(self, os_creds):
- """
- Returns the project ID for a given project_name or None
- :param os_creds: the credentials required for keystone client retrieval
- :return: the ID or None
- """
- if self.project_id:
- return self.project_id
- else:
- if self.project_name:
- keystone = keystone_utils.keystone_client(os_creds)
- project = keystone_utils.get_project(
- keystone=keystone, project_name=self.project_name)
- if project:
- return project.id
-
- return None
-
- 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
- TODO - expand automated testing to exercise all parameters
-
- :param os_creds: the OpenStack credentials
- :return: the dictionary object
- """
- out = dict()
-
- if self.name:
- out['name'] = self.name
- if self.admin_state_up is not None:
- out['admin_state_up'] = self.admin_state_up
- if self.shared:
- out['shared'] = self.shared
- if self.project_name:
- project_id = self.get_project_id(os_creds)
- if project_id:
- out['tenant_id'] = project_id
- else:
- raise NetworkSettingsError(
- '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:
- out['provider:physical_network'] = self.physical_network
- if self.segmentation_id:
- out['provider:segmentation_id'] = self.segmentation_id
- if self.external:
- out['router:external'] = self.external
- return {'network': out}
-
-
-class NetworkSettingsError(Exception):
- """
- Exception to be thrown when networks settings attributes are incorrect
- """
+ from warnings import warn
+ warn('Use snaps.config.network.NetworkConfig instead',
+ DeprecationWarning)
+ super(self.__class__, self).__init__(**kwargs)
class IPv6Mode(enum.Enum):
"""
A rule's direction
+ deprecated - use snaps.config.network.IPv6Mode
"""
slaac = 'slaac'
stateful = 'dhcpv6-stateful'
stateless = 'dhcpv6-stateless'
-class SubnetSettings:
+class SubnetSettings(SubnetConfig):
"""
- Class representing a subnet configuration
+ Class to hold the configuration settings required for creating OpenStack
+ Subnet objects
+ deprecated
"""
def __init__(self, **kwargs):
- """
- Constructor - all parameters are optional except cidr (subnet mask)
- :param name: The subnet name (required)
- :param cidr: The CIDR (required)
- :param ip_version: The IP version, which is 4 or 6 (required)
- :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 (optional)
- :param start: The start address for the allocation pools (optional)
- :param end: The end address for the allocation pools (optional)
- :param gateway_ip: The gateway IP address (optional)
- :param enable_dhcp: Set to true if DHCP is enabled and false if DHCP is
- disabled (optional)
- :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]
- (default '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",
- "nexthop":"123.456.78.9"
- },
- {
- "destination":"192.168.0.0/24",
- "nexthop":"192.168.0.1"
- }
- ]
- :param destination: The destination for static route (optional)
- :param nexthop: The next hop for the destination (optional)
- :param ipv6_ra_mode: an instance of the IPv6Mode enum
- (optional when enable_dhcp is True)
- :param ipv6_address_mode: an instance of the IPv6Mode enum
- (optional when enable_dhcp is True)
- :raise: SubnetSettingsError when config does not have or cidr values
- are None
- """
- self.cidr = kwargs.get('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 'dns_nameservers' in kwargs:
- self.dns_nameservers = kwargs.get('dns_nameservers')
- else:
- if self.ip_version == 4:
- self.dns_nameservers = ['8.8.8.8']
- else:
- self.dns_nameservers = list()
+ from warnings import warn
+ warn('Use snaps.config.network.SubnetConfig instead',
+ DeprecationWarning)
+ super(self.__class__, self).__init__(**kwargs)
- self.host_routes = kwargs.get('host_routes')
- self.destination = kwargs.get('destination')
- self.nexthop = kwargs.get('nexthop')
- self.ipv6_ra_mode = map_mode(kwargs.get('ipv6_ra_mode'))
- self.ipv6_address_mode = map_mode(kwargs.get('ipv6_address_mode'))
- if not self.name or not self.cidr:
- raise SubnetSettingsError('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
- :param os_creds: the OpenStack credentials
- :param network: The network object on which the subnet will be created
- (optional)
- :return: the dictionary object
- """
- out = {
- 'cidr': self.cidr,
- 'ip_version': self.ip_version,
- }
-
- if network:
- out['network_id'] = network.id
- if self.name:
- out['name'] = self.name
- if self.project_name:
- keystone = keystone_utils.keystone_client(os_creds)
- project = keystone_utils.get_project(
- keystone=keystone, project_name=self.project_name)
- project_id = None
- if project:
- project_id = project.id
- if project_id:
- out['tenant_id'] = project_id
- else:
- raise SubnetSettingsError(
- '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:
- out['gateway_ip'] = self.gateway_ip
- if self.enable_dhcp is not None:
- out['enable_dhcp'] = self.enable_dhcp
- if self.dns_nameservers and len(self.dns_nameservers) > 0:
- out['dns_nameservers'] = self.dns_nameservers
- if self.host_routes and len(self.host_routes) > 0:
- out['host_routes'] = self.host_routes
- if self.destination:
- out['destination'] = self.destination
- if self.nexthop:
- out['nexthop'] = self.nexthop
- if self.ipv6_ra_mode:
- out['ipv6_ra_mode'] = self.ipv6_ra_mode.value
- if self.ipv6_address_mode:
- out['ipv6_address_mode'] = self.ipv6_address_mode.value
- return out
-
-
-def map_mode(mode):
+class PortSettings(PortConfig):
"""
- Takes a the direction value maps it to the Direction enum. When None return
- None
- :param mode: the mode value
- :return: the IPv6Mode enum object
- :raise: SubnetSettingsError if value is invalid
- """
- if not mode:
- return None
- if isinstance(mode, IPv6Mode):
- return mode
- else:
- mode_str = str(mode)
- if mode_str == 'slaac':
- return IPv6Mode.slaac
- elif mode_str == 'dhcpv6-stateful':
- return IPv6Mode.stateful
- elif mode_str == 'stateful':
- return IPv6Mode.stateful
- elif mode_str == 'dhcpv6-stateless':
- return IPv6Mode.stateless
- elif mode_str == 'stateless':
- return IPv6Mode.stateless
- else:
- raise SubnetSettingsError('Invalid mode - ' + mode_str)
-
-
-class SubnetSettingsError(Exception):
- """
- Exception to be thrown when subnet settings attributes are incorrect
- """
-
-
-class PortSettings:
- """
- Class representing a port configuration
+ Class to hold the configuration settings required for creating OpenStack
+ Subnet objects
+ deprecated
"""
def __init__(self, **kwargs):
- """
- Constructor
- :param name: A symbolic name for the port (optional).
- :param network_name: The name of the network on which to create the
- port (required).
- :param admin_state_up: A boolean value denoting the administrative
- status of the port (default = True)
- :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 (optional)
- :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 (optional)
- :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 (optional)
- :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 (optional)
- :param opt_value: The extra DHCP option value (optional)
- :param opt_name: The extra DHCP option name (optional)
- :param device_owner: The ID of the entity that uses this port.
- For example, a DHCP agent (optional)
- :param device_id: The ID of the device that uses this port.
- For example, a virtual server (optional)
- :return:
- """
- if 'port' in kwargs:
- kwargs = kwargs['port']
-
- self.name = kwargs.get('name')
- self.network_name = kwargs.get('network_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.project_name = kwargs.get('project_name')
- self.mac_address = kwargs.get('mac_address')
- self.ip_addrs = kwargs.get('ip_addrs')
- 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.network_name:
- raise PortSettingsError(
- 'The attribute network_name is required')
-
- def __get_fixed_ips(self, neutron):
- """
- Sets the self.fixed_ips value
- :param neutron: the Neutron client
- :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})
- 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.
- 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
- :param os_creds: the OpenStack credentials
- :return: the dictionary object
- """
-
- out = dict()
-
- project_id = None
- if self.project_name:
- keystone = keystone_utils.keystone_client(os_creds)
- project = keystone_utils.get_project(
- keystone=keystone, project_name=self.project_name)
- if project:
- project_id = project.id
-
- 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'] = network.id
-
- if self.admin_state_up is not None:
- out['admin_state_up'] = self.admin_state_up
- if self.name:
- out['name'] = self.name
- if self.project_name:
- if project_id:
- out['tenant_id'] = project_id
- else:
- raise PortSettingsError(
- 'Could not find project ID for project named - ' +
- self.project_name)
- if self.mac_address:
- out['mac_address'] = self.mac_address
-
- 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:
- out['allowed_address_pairs'] = self.allowed_address_pairs
- if self.opt_value:
- out['opt_value'] = self.opt_value
- if self.opt_name:
- out['opt_name'] = self.opt_name
- if self.device_owner:
- out['device_owner'] = self.device_owner
- if self.device_id:
- out['device_id'] = self.device_id
- return {'port': out}
-
- def __eq__(self, other):
- return (self.name == other.name and
- self.network_name == other.network_name and
- self.admin_state_up == other.admin_state_up and
- 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.security_groups == other.security_groups and
- self.allowed_address_pairs == other.allowed_address_pairs and
- self.opt_value == other.opt_value and
- self.opt_name == other.opt_name and
- self.device_owner == other.device_owner and
- self.device_id == other.device_id)
-
-
-class PortSettingsError(Exception):
- """
- Exception to be thrown when port settings attributes are incorrect
- """
+ from warnings import warn
+ warn('Use snaps.config.network.PortConfig instead',
+ DeprecationWarning)
+ super(self.__class__, self).__init__(**kwargs)
self.__heat_cli, neutron, self.__stack)
for stack_network in stack_networks:
- net_settings = settings_utils.create_network_settings(
+ net_settings = settings_utils.create_network_config(
neutron, stack_network)
net_creator = OpenStackNetwork(self._os_creds, net_settings)
out.append(net_creator)
from novaclient.exceptions import BadRequest
from snaps import file_utils
-from snaps.config.volume import VolumeConfig
-from snaps.config.router import RouterConfig
+from snaps.config.flavor import FlavorConfig
+from snaps.config.image import ImageConfig
from snaps.config.keypair import KeypairConfig
+from snaps.config.network import PortConfig, NetworkConfig, SubnetConfig
+from snaps.config.router import RouterConfig
+from snaps.config.volume import VolumeConfig
from snaps.openstack import create_network, create_router
-from snaps.config.flavor import FlavorConfig
from snaps.openstack.create_flavor import OpenStackFlavor
-from snaps.config.image import ImageConfig
from snaps.openstack.create_image import OpenStackImage
from snaps.openstack.create_instance import (
VmInstanceSettings, OpenStackVmInstance, FloatingIpSettings,
VmInstanceSettingsError, FloatingIpSettingsError)
from snaps.openstack.create_keypairs import OpenStackKeypair
-from snaps.openstack.create_network import (
- OpenStackNetwork, PortSettings, NetworkSettings, SubnetSettings)
+from snaps.openstack.create_network import OpenStackNetwork
from snaps.openstack.create_router import OpenStackRouter
from snaps.openstack.create_security_group import (
SecurityGroupSettings, OpenStackSecurityGroup, SecurityGroupRuleSettings,
VmInstanceSettings(config={'name': 'foo', 'flavor': 'bar'})
def test_name_flavor_port_only(self):
- port_settings = PortSettings(name='foo-port', network_name='bar-net')
+ port_settings = PortConfig(name='foo-port', network_name='bar-net')
settings = VmInstanceSettings(name='foo', flavor='bar',
port_settings=[port_settings])
self.assertEqual('foo', settings.name)
self.assertIsNone(settings.volume_names)
def test_config_with_name_flavor_port_only(self):
- port_settings = PortSettings(name='foo-port', network_name='bar-net')
+ port_settings = PortConfig(name='foo-port', network_name='bar-net')
settings = VmInstanceSettings(
**{'name': 'foo', 'flavor': 'bar', 'ports': [port_settings]})
self.assertEqual('foo', settings.name)
self.assertIsNone(settings.volume_names)
def test_all(self):
- port_settings = PortSettings(name='foo-port', network_name='bar-net')
+ port_settings = PortConfig(name='foo-port', network_name='bar-net')
fip_settings = FloatingIpSettings(name='foo-fip', port_name='bar-port',
router_name='foo-bar-router')
self.assertEqual('vol1', settings.volume_names[0])
def test_config_all(self):
- port_settings = PortSettings(name='foo-port', network_name='bar-net')
+ port_settings = PortConfig(name='foo-port', network_name='bar-net')
fip_settings = FloatingIpSettings(name='foo-fip', port_name='bar-port',
router_name='foo-bar-router')
self.priv_net_config = openstack_tests.get_priv_net_config(
net_name=guid + '-priv-net', subnet_name=guid + '-priv-subnet')
- self.port_settings = PortSettings(
+ self.port_settings = PortConfig(
name=self.port_1_name,
network_name=self.priv_net_config.network_settings.name)
self.os_creds, net_config.network_settings)
self.network_creator.create()
- self.port_settings = PortSettings(
+ self.port_settings = PortConfig(
name=guid + '-port',
network_name=net_config.network_settings.name)
"""
ip_1 = '10.55.1.100'
sub_settings = self.pub_net_config.network_settings.subnet_settings
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=self.port_1_name,
network_name=self.pub_net_config.network_settings.name,
ip_addrs=[
Tests the ability to access a VM via SSH and a floating IP when it has
been assigned prior to being active.
"""
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=self.port_1_name,
network_name=self.pub_net_config.network_settings.name)
Tests the ability to access a VM via SSH and a floating IP when it has
been assigned prior to being active.
"""
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=self.port_1_name,
network_name=self.pub_net_config.network_settings.name)
Tests the ability to access a VM via SSH and a floating IP via a
creator that is identical to the original creator.
"""
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=self.port_1_name,
network_name=self.pub_net_config.network_settings.name)
Tests the ability to assign an IPv4 floating IP to an IPv6 overlay
network when the external network does not have an IPv6 subnet.
"""
- subnet_settings = SubnetSettings(
+ subnet_settings = SubnetConfig(
name=self.guid + '-subnet', cidr='1:1:0:0:0:0:0:0/64',
ip_version=6)
- network_settings = NetworkSettings(
+ network_settings = NetworkConfig(
name=self.guid + '-net', subnet_settings=[subnet_settings])
router_settings = RouterConfig(
name=self.guid + '-router', external_gateway=self.ext_net_name,
self.os_creds, router_settings)
self.router_creator.create()
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=self.port1_name, network_name=network_settings.name)
instance_settings = VmInstanceSettings(
Tests the ability to assign an IPv4 floating IP to an IPv6 overlay
network when the external network does not have an IPv6 subnet.
"""
- subnet4_settings = SubnetSettings(
+ subnet4_settings = SubnetConfig(
name=self.guid + '-subnet4', cidr='10.0.1.0/24',
ip_version=4)
- subnet6_settings = SubnetSettings(
+ subnet6_settings = SubnetConfig(
name=self.guid + '-subnet6', cidr='1:1:0:0:0:0:0:0/64',
ip_version=6)
- network_settings = NetworkSettings(
+ network_settings = NetworkConfig(
name=self.guid + '-net',
subnet_settings=[subnet4_settings, subnet6_settings])
router_settings = RouterConfig(
self.os_creds, router_settings)
self.router_creator.create()
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=self.port1_name, network_name=network_settings.name)
instance_settings = VmInstanceSettings(
"""
ip = '10.55.0.101'
sub_settings = self.net_config.network_settings.subnet_settings
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=self.port_1_name,
network_name=self.net_config.network_settings.name,
ip_addrs=[{'subnet_name': sub_settings[0].name, 'ip': ip}])
"""
ip = '10.66.0.101'
sub_settings = self.net_config.network_settings.subnet_settings
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=self.port_1_name,
network_name=self.net_config.network_settings.name,
ip_addrs=[{'subnet_name': sub_settings[0].name, 'ip': ip}])
the MAC address is assigned.
"""
mac_addr = '0a:1b:2c:3d:4e:5f'
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=self.port_1_name,
network_name=self.net_config.network_settings.name,
mac_address=mac_addr)
invalid MAC address value is being
assigned. This should raise an Exception
"""
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=self.port_1_name,
network_name=self.net_config.network_settings.name,
mac_address='foo')
ip = '10.55.0.101'
mac_addr = '0a:1b:2c:3d:4e:5f'
sub_settings = self.net_config.network_settings.subnet_settings
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=self.port_1_name,
network_name=self.net_config.network_settings.name,
mac_address=mac_addr,
ip = '10.55.0.101'
mac_addr = '0a:1b:2c:3d:4e:5f'
pair = {'ip_address': ip, 'mac_address': mac_addr}
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=self.port_1_name,
network_name=self.net_config.network_settings.name,
allowed_address_pairs=[pair])
pair = {'ip_address': ip, 'mac_address': mac_addr}
pairs = set()
pairs.add((ip, mac_addr))
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=self.port_1_name,
network_name=self.net_config.network_settings.name,
allowed_address_pairs=[pair])
pair = {'ip_address': ip, 'mac_address': mac_addr}
pairs = set()
pairs.add((ip, mac_addr))
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=self.port_1_name,
network_name=self.net_config.network_settings.name,
allowed_address_pairs=[pair])
for zone in zone_hosts:
inst_name = self.vm_inst_name + '-' + zone
ctr += 1
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=self.port_base_name + '-' + str(ctr),
network_name=self.priv_net_config.network_settings.name)
ports_settings = []
ctr = 1
for network_creator in self.network_creators:
- ports_settings.append(PortSettings(
+ ports_settings.append(PortConfig(
name=self.guid + '-port-' + str(ctr),
network_name=network_creator.network_settings.name))
ctr += 1
metadata=self.flavor_metadata))
self.flavor_creator.create()
- self.port_settings = PortSettings(
+ self.port_settings = PortConfig(
name=self.guid + '-port',
network_name=net_config.network_settings.name)
except Exception as e:
self.os_creds, net_config.network_settings)
self.network_creator.create()
- self.port_settings = PortSettings(
+ self.port_settings = PortConfig(
name=guid + '-port',
network_name=net_config.network_settings.name)
except Exception as e:
self.priv_net_config = openstack_tests.get_priv_net_config(
net_name=self.guid + '-priv-net',
subnet_name=self.guid + '-priv-subnet')
- self.port_settings = PortSettings(
+ self.port_settings = PortConfig(
name=self.port_1_name,
network_name=self.priv_net_config.network_settings.name)
self.vm_inst2_name = self.guid + '-inst2'
self.port_1_name = self.guid + '-vm1-port'
self.port_2_name = self.guid + '-vm2-port'
- self.net_config_1 = NetworkSettings(
+ self.net_config_1 = NetworkConfig(
name=self.guid + '-net1',
subnet_settings=[
- create_network.SubnetSettings(
+ create_network.SubnetConfig(
cidr=cidr1, name=self.guid + '-subnet1',
gateway_ip=static_gateway_ip1)])
- self.net_config_2 = NetworkSettings(
+ self.net_config_2 = NetworkConfig(
name=self.guid + '-net2',
subnet_settings=[
- create_network.SubnetSettings(
+ create_network.SubnetConfig(
cidr=cidr2, name=self.guid + '-subnet2',
gateway_ip=static_gateway_ip2)])
network_creator.create()
port_settings = [
- create_network.PortSettings(
+ create_network.PortConfig(
name=self.guid + '-router-port1',
ip_addrs=[{
'subnet_name':
}],
network_name=self.net_config_1.name,
project_name=self.os_creds.project_name),
- create_network.PortSettings(
+ create_network.PortConfig(
name=self.guid + '-router-port2',
ip_addrs=[{
'subnet_name':
ports_settings = []
ctr = 1
for network_creator in self.network_creators:
- ports_settings.append(PortSettings(
+ ports_settings.append(PortConfig(
name=self.guid + '-port-' + str(ctr),
network_name=network_creator.network_settings.name))
ctr += 1
name=self.vm_inst1_name,
flavor=self.flavor_creator.flavor_settings.name,
userdata=_get_ping_userdata(self.ip2),
- port_settings=[PortSettings(
+ port_settings=[PortConfig(
name=self.port_1_name,
ip_addrs=[{
'subnet_name':
name=self.vm_inst2_name,
flavor=self.flavor_creator.flavor_settings.name,
userdata=_get_ping_userdata(self.ip1),
- port_settings=[PortSettings(
+ port_settings=[PortConfig(
name=self.port_2_name,
ip_addrs=[{
'subnet_name':
self.os_creds, net_config.network_settings)
self.network_creator.create()
- self.port_settings = PortSettings(
+ self.port_settings = PortConfig(
name=guid + '-port',
network_name=net_config.network_settings.name)
import unittest
import uuid
+from snaps.config.network import (
+ NetworkConfig, SubnetConfig, SubnetConfigError, NetworkConfigError,
+ PortConfigError, IPv6Mode)
from snaps.openstack import create_router
from snaps.openstack.create_network import (
- OpenStackNetwork, NetworkSettings, SubnetSettings, PortSettings,
- NetworkSettingsError, SubnetSettingsError, PortSettingsError, IPv6Mode)
+ 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.utils import neutron_utils
from snaps.openstack.utils.tests import neutron_utils_tests
+from snaps.openstack.create_network import IPv6Mode as IPv6Mode_old
__author__ = 'spisarski'
"""
def test_no_params(self):
- with self.assertRaises(NetworkSettingsError):
+ with self.assertRaises(NetworkConfigError):
NetworkSettings()
def test_empty_config(self):
- with self.assertRaises(NetworkSettingsError):
+ with self.assertRaises(NetworkConfigError):
NetworkSettings(**dict())
def test_name_only(self):
"""
def test_no_params(self):
- with self.assertRaises(SubnetSettingsError):
+ with self.assertRaises(SubnetConfigError):
SubnetSettings()
def test_empty_config(self):
- with self.assertRaises(SubnetSettingsError):
+ with self.assertRaises(SubnetConfigError):
SubnetSettings(**dict())
def test_name_only(self):
- with self.assertRaises(SubnetSettingsError):
+ with self.assertRaises(SubnetConfigError):
SubnetSettings(name='foo')
def test_config_with_name_only(self):
- with self.assertRaises(SubnetSettingsError):
+ with self.assertRaises(SubnetConfigError):
SubnetSettings(**{'name': 'foo'})
def test_name_cidr_only(self):
def test_all_string_enums(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)
self.assertEqual(host_routes, settings.host_routes[0])
self.assertEqual('dest', settings.destination)
self.assertEqual('hop', settings.nexthop)
- self.assertEqual(IPv6Mode.stateful, settings.ipv6_ra_mode)
- self.assertEqual(IPv6Mode.slaac, settings.ipv6_address_mode)
+ self.assertEqual(IPv6Mode_old.stateful.value,
+ settings.ipv6_ra_mode.value)
+ self.assertEqual(IPv6Mode_old.slaac.value,
+ settings.ipv6_address_mode.value)
def test_all_type_enums(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=IPv6Mode.stateful,
- ipv6_address_mode=IPv6Mode.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=IPv6Mode_old.stateful,
+ ipv6_address_mode=IPv6Mode.slaac)
self.assertEqual('foo', settings.name)
self.assertEqual('10.0.0.0/24', settings.cidr)
self.assertEqual(6, settings.ip_version)
self.assertEqual(host_routes, settings.host_routes[0])
self.assertEqual('dest', settings.destination)
self.assertEqual('hop', settings.nexthop)
- self.assertEqual(IPv6Mode.stateful, settings.ipv6_ra_mode)
- self.assertEqual(IPv6Mode.slaac, settings.ipv6_address_mode)
+ self.assertEqual(IPv6Mode.stateful.value, settings.ipv6_ra_mode.value)
+ self.assertEqual(IPv6Mode.slaac.value,
+ settings.ipv6_address_mode.value)
def test_config_all(self):
host_routes = {'destination': '0.0.0.0/0', 'nexthop': '123.456.78.9'}
self.assertEqual(host_routes, settings.host_routes[0])
self.assertEqual('dest', settings.destination)
self.assertEqual('hop', settings.nexthop)
- self.assertEqual(IPv6Mode.stateless, settings.ipv6_ra_mode)
- self.assertEqual(IPv6Mode.slaac, settings.ipv6_address_mode)
+ self.assertEqual(IPv6Mode.stateless.value, settings.ipv6_ra_mode.value)
+ self.assertEqual(IPv6Mode.slaac.value,
+ settings.ipv6_address_mode.value)
class PortSettingsUnitTests(unittest.TestCase):
"""
def test_no_params(self):
- with self.assertRaises(PortSettingsError):
+ with self.assertRaises(PortConfigError):
PortSettings()
def test_empty_config(self):
- with self.assertRaises(PortSettingsError):
+ with self.assertRaises(PortConfigError):
PortSettings(**dict())
def test_name_only(self):
- with self.assertRaises(PortSettingsError):
+ with self.assertRaises(PortConfigError):
PortSettings(name='foo')
def test_config_name_only(self):
- with self.assertRaises(PortSettingsError):
+ with self.assertRaises(PortConfigError):
PortSettings(**{'name': 'foo'})
def test_name_netname_only(self):
Tests the creation of an OpenStack network without a router.
"""
# Create Network
- subnet_settings = SubnetSettings(
+ subnet_settings = SubnetConfig(
name=self.guid + '-subnet', cidr='1:1:0:0:0:0:0:0/64',
ip_version=6)
- network_settings = NetworkSettings(
+ network_settings = NetworkConfig(
name=self.guid + '-net', subnet_settings=[subnet_settings])
self.net_creator = OpenStackNetwork(self.os_creds, network_settings)
Tests the creation of an OpenStack network without a router.
"""
# Create Network
- subnet4_settings = SubnetSettings(
+ subnet4_settings = SubnetConfig(
name=self.guid + '-subnet4', cidr='10.0.1.0/24', ip_version=4)
- subnet6_settings = SubnetSettings(
+ subnet6_settings = SubnetConfig(
name=self.guid + '-subnet6', cidr='1:1:0:0:0:0:0:0/64',
ip_version=6)
- network_settings = NetworkSettings(
+ network_settings = NetworkConfig(
name=self.guid + '-net',
subnet_settings=[subnet4_settings, subnet6_settings])
"""
# Create Network
network_type = 'vlan'
- net_settings = NetworkSettings(
+ net_settings = NetworkConfig(
name=self.net_config.network_settings.name,
subnet_settings=self.net_config.network_settings.subnet_settings,
network_type=network_type)
physical_network = 'datacentre'
segmentation_id = 2366
- net_settings = NetworkSettings(
+ net_settings = NetworkConfig(
name=self.net_config.network_settings.name,
subnet_settings=self.net_config.network_settings.subnet_settings,
network_type=network_type,
"""
# Create Network
network_type = 'vxlan'
- net_settings = NetworkSettings(
+ net_settings = NetworkConfig(
name=self.net_config.network_settings.name,
subnet_settings=self.net_config.network_settings.subnet_settings,
network_type=network_type)
# This value must be variable to work on all OpenStack pods
physical_network = 'datacentre'
- net_settings = NetworkSettings(
+ net_settings = NetworkConfig(
name=self.net_config.network_settings.name,
subnet_settings=self.net_config.network_settings.subnet_settings,
network_type=network_type, physical_network=physical_network)
"""
# Create Network
network_type = 'foo'
- net_settings = NetworkSettings(
+ net_settings = NetworkConfig(
name=self.net_config.network_settings.name,
subnet_settings=self.net_config.network_settings.subnet_settings,
network_type=network_type)
import unittest
import uuid
+from snaps.config.network import PortConfig, NetworkConfig
from snaps.config.router import RouterConfigError, RouterConfig
from snaps.openstack import create_network
from snaps.openstack import create_router
-from snaps.openstack.create_network import (
- NetworkSettings, PortSettings)
from snaps.openstack.create_network import OpenStackNetwork
from snaps.openstack.create_router import RouterSettings
from snaps.openstack.tests.os_source_file_test import OSIntegrationTestCase
self.assertEqual(0, len(settings.port_settings))
def test_all(self):
- port_settings = PortSettings(name='foo', network_name='bar')
+ port_settings = PortConfig(name='foo', network_name='bar')
settings = RouterSettings(
name='foo', project_name='bar', external_gateway='foo_gateway',
admin_state_up=True, enable_snat=False,
self.assertTrue(isinstance(settings.internal_subnets, list))
self.assertEqual(1, len(settings.internal_subnets))
self.assertEqual(['10.0.0.1/24'], settings.internal_subnets)
- self.assertEqual([PortSettings(**{'name': 'foo-port',
- 'network_name': 'bar-net'})],
- settings.port_settings)
+ self.assertEqual(
+ [PortConfig(**{'name': 'foo-port', 'network_name': 'bar-net'})],
+ settings.port_settings)
class CreateRouterSuccessTests(OSIntegrationTestCase):
Test creation of a router connected with two private networks and no
external gateway
"""
- network_settings1 = NetworkSettings(
+ network_settings1 = NetworkConfig(
name=self.guid + '-pub-net1',
subnet_settings=[
- create_network.SubnetSettings(
+ create_network.SubnetConfig(
cidr=cidr1, name=self.guid + '-pub-subnet1',
gateway_ip=static_gateway_ip1)])
- network_settings2 = NetworkSettings(
+ network_settings2 = NetworkConfig(
name=self.guid + '-pub-net2',
subnet_settings=[
- create_network.SubnetSettings(
+ create_network.SubnetConfig(
cidr=cidr2, name=self.guid + '-pub-subnet2',
gateway_ip=static_gateway_ip2)])
self.network_creator2.create()
port_settings = [
- create_network.PortSettings(
+ create_network.PortConfig(
name=self.guid + '-port1',
ip_addrs=[{
'subnet_name':
}],
network_name=network_settings1.name,
project_name=self.os_creds.project_name),
- create_network.PortSettings(
+ create_network.PortConfig(
name=self.guid + '-port2',
ip_addrs=[{
'subnet_name': network_settings2.subnet_settings[0].name,
Test creation of a router connected to an external network and a
private network.
"""
- network_settings = NetworkSettings(
+ network_settings = NetworkConfig(
name=self.guid + '-pub-net1',
subnet_settings=[
- create_network.SubnetSettings(
+ create_network.SubnetConfig(
cidr=cidr1, name=self.guid + '-pub-subnet1',
gateway_ip=static_gateway_ip1)])
self.network_creator1 = OpenStackNetwork(self.os_creds,
self.network_creator1.create()
port_settings = [
- create_network.PortSettings(
+ create_network.PortConfig(
name=self.guid + '-port1',
ip_addrs=[{
'subnet_name': network_settings.subnet_settings[0].name,
import pkg_resources
from snaps import file_utils
from snaps.config.image import ImageConfig
+from snaps.config.network import NetworkConfig, SubnetConfig
from snaps.config.router import RouterConfig
-from snaps.openstack.create_network import NetworkSettings, SubnetSettings
from snaps.openstack.os_credentials import OSCreds, ProxySettings
__author__ = 'spisarski'
router_name=None, external_gateway=None):
if subnet_name and subnet_cidr:
- self.network_settings = NetworkSettings(
+ self.network_settings = NetworkConfig(
name=net_name, subnet_settings=[
- SubnetSettings(cidr=subnet_cidr, name=subnet_name)])
+ SubnetConfig(cidr=subnet_cidr, name=subnet_name)])
else:
- self.network_settings = NetworkSettings(name=net_name)
+ self.network_settings = NetworkConfig(name=net_name)
if router_name:
if subnet_name:
else the query will use just the name from the network_name parameter.
When the project_id is included, that will be added to the query filter.
:param neutron: the client
- :param network_settings: the NetworkSettings object used to create filter
+ :param network_settings: the NetworkConfig object used to create filter
:param network_name: the name of the network to retrieve
:param project_id: the id of the network's project
:return: a SNAPS-OO Network domain object
"""
Returns the first port object (dictionary) found for the given query
:param neutron: the client
- :param port_settings: the PortSettings object used for generating the query
+ :param port_settings: the PortConfig object used for generating the query
:param port_name: if port_settings is None, this name is the value to place
into the query
:return: a SNAPS-OO Port domain object
from snaps import file_utils
from snaps.config.flavor import FlavorConfig
from snaps.config.keypair import KeypairConfig
+from snaps.config.network import SubnetConfig, PortConfig, NetworkConfig
from snaps.config.router import RouterConfig
from snaps.config.volume import VolumeConfig
from snaps.config.volume_type import (
ControlLocation, VolumeTypeEncryptionConfig, VolumeTypeConfig)
from snaps.openstack.create_instance import (
VmInstanceSettings, FloatingIpSettings)
-from snaps.openstack.create_network import (
- PortSettings, SubnetSettings, NetworkSettings)
from snaps.openstack.create_security_group import (
SecurityGroupSettings, SecurityGroupRuleSettings)
from snaps.openstack.utils import (
neutron_utils, nova_utils, heat_utils, glance_utils)
-def create_network_settings(neutron, network):
+def create_network_config(neutron, network):
"""
- Returns a NetworkSettings object
+ Returns a NetworkConfig object
:param neutron: the neutron client
:param network: a SNAPS-OO Network domain object
:return:
"""
- return NetworkSettings(
+ return NetworkConfig(
name=network.name, network_type=network.type,
- subnet_settings=create_subnet_settings(neutron, network))
+ subnet_settings=create_subnet_config(neutron, network))
def create_security_group_settings(neutron, security_group):
"""
- Returns a NetworkSettings object
+ Returns a NetworkConfig object
:param neutron: the neutron client
:param security_group: a SNAPS-OO SecurityGroup domain object
:return:
rule_settings=rule_settings)
-def create_subnet_settings(neutron, network):
+def create_subnet_config(neutron, network):
"""
- Returns a list of SubnetSettings objects for a given network
+ Returns a list of SubnetConfig objects for a given network
:param neutron: the OpenStack neutron client
:param network: the SNAPS-OO Network domain object
:return: a list
kwargs['host_routes'] = subnet.host_routes
kwargs['ipv6_ra_mode'] = subnet.ipv6_ra_mode
kwargs['ipv6_address_mode'] = subnet.ipv6_address_mode
- out.append(SubnetSettings(**kwargs))
+ out.append(SubnetConfig(**kwargs))
return out
ports_tuple_list.append((network, ip_list))
- port_settings = __create_port_settings(neutron, ports_tuple_list)
+ port_settings = __create_port_config(neutron, ports_tuple_list)
filtered_settings = list()
for port_setting in port_settings:
def create_vm_inst_settings(nova, neutron, server):
"""
- Returns a NetworkSettings object
+ Returns a NetworkConfig object
:param nova: the nova client
:param neutron: the neutron client
:param server: a SNAPS-OO VmInst domain object
if network:
net_tuples.append((network, ips))
- kwargs['port_settings'] = __create_port_settings(
+ kwargs['port_settings'] = __create_port_config(
neutron, net_tuples)
kwargs['security_group_names'] = server.sec_grp_names
kwargs['floating_ip_settings'] = __create_floatingip_settings(
return VmInstanceSettings(**kwargs)
-def __create_port_settings(neutron, networks):
+def __create_port_config(neutron, networks):
"""
Returns a list of port settings based on the networks parameter
:param neutron: the neutron client
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))
+ out.append(PortConfig(**kwargs))
return out
Returns a list of FloatingIPSettings objects as they pertain to an
existing deployed server instance
:param neutron: the neutron client
- :param port_settings: list of SNAPS-OO PortSettings objects
+ :param port_settings: list of SNAPS-OO PortConfig objects
:return: a list of FloatingIPSettings objects or an empty list if no
floating IPs have been created
"""
self.assertEqual(1, len(networks))
self.assertEqual(self.network_name, networks[0].name)
- network_settings = settings_utils.create_network_settings(
+ network_settings = settings_utils.create_network_config(
neutron, networks[0])
self.assertIsNotNone(network_settings)
self.assertEqual(self.network_name, network_settings.name)
from neutronclient.common.exceptions import NotFound, BadRequest
-from snaps.openstack import create_router
-from snaps.openstack.create_network import NetworkSettings, SubnetSettings, \
- PortSettings
-from snaps.openstack.create_security_group import SecurityGroupSettings, \
- SecurityGroupRuleSettings, Direction
+from snaps.config.network import NetworkConfig, SubnetConfig, PortConfig
+from snaps.openstack.create_security_group import (
+ SecurityGroupSettings, SecurityGroupRuleSettings, Direction)
from snaps.openstack.tests import openstack_tests
from snaps.openstack.tests import validation_utils
from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
with self.assertRaises(Exception):
self.network = neutron_utils.create_network(
self.neutron, self.os_creds,
- network_settings=NetworkSettings(name=''))
+ network_settings=NetworkConfig(name=''))
def test_create_network_null_name(self):
"""
with self.assertRaises(Exception):
self.network = neutron_utils.create_network(
self.neutron, self.os_creds,
- network_settings=NetworkSettings())
+ network_settings=NetworkConfig())
class NeutronUtilsSubnetTests(OSComponentTestCase):
self.neutron, self.net_config.network_settings.name, True))
with self.assertRaises(Exception):
- SubnetSettings(cidr=self.net_config.subnet_cidr)
+ SubnetConfig(cidr=self.net_config.subnet_cidr)
def test_create_subnet_empty_name(self):
"""
Tests the neutron_utils.create_network() with an IPv6 subnet where DHCP
is True and IPv6 modes are slaac
"""
- sub_setting = SubnetSettings(
+ sub_setting = SubnetConfig(
name=self.guid + '-subnet', cidr='1:1:0:0:0:0:0:0/64',
ip_version=6, dns_nameservers=['2620:0:ccc:0:0:0:0:2'],
gateway_ip='1:1:0:0:0:0:0:1', start='1:1::ff', end='1:1::ffff',
enable_dhcp=True, ipv6_ra_mode='slaac', ipv6_address_mode='slaac')
- self.network_settings = NetworkSettings(
+ self.network_settings = NetworkConfig(
name=self.guid + '-net', subnet_settings=[sub_setting])
self.network = neutron_utils.create_network(
Tests the neutron_utils.create_network() with an IPv6 subnet where DHCP
is True and IPv6 modes are stateful
"""
- sub_setting = SubnetSettings(
+ sub_setting = SubnetConfig(
name=self.guid + '-subnet', cidr='1:1:0:0:0:0:0:0/64',
ip_version=6, dns_nameservers=['2620:0:ccc:0:0:0:0:2'],
gateway_ip='1:1:0:0:0:0:0:1', start='1:1::ff', end='1:1::ffff',
enable_dhcp=True, ipv6_ra_mode='dhcpv6-stateful',
ipv6_address_mode='dhcpv6-stateful')
- self.network_settings = NetworkSettings(
+ self.network_settings = NetworkConfig(
name=self.guid + '-net', subnet_settings=[sub_setting])
self.network = neutron_utils.create_network(
Tests the neutron_utils.create_network() when DHCP is enabled and
the RA and address modes are both 'slaac'
"""
- sub_setting = SubnetSettings(
+ sub_setting = SubnetConfig(
name=self.guid + '-subnet', cidr='1:1:0:0:0:0:0:0/64',
ip_version=6, dns_nameservers=['2620:0:ccc:0:0:0:0:2'],
gateway_ip='1:1:0:0:0:0:0:1', start='1:1::ff', end='1:1::ffff',
enable_dhcp=True, ipv6_ra_mode='dhcpv6-stateless',
ipv6_address_mode='dhcpv6-stateless')
- self.network_settings = NetworkSettings(
+ self.network_settings = NetworkConfig(
name=self.guid + '-net', subnet_settings=[sub_setting])
self.network = neutron_utils.create_network(
Tests the neutron_utils.create_network() for a BadRequest when
DHCP is not enabled and the RA and address modes are both 'slaac'
"""
- sub_setting = SubnetSettings(
+ sub_setting = SubnetConfig(
name=self.guid + '-subnet', cidr='1:1:0:0:0:0:0:0/64',
ip_version=6, dns_nameservers=['2620:0:ccc:0:0:0:0:2'],
gateway_ip='1:1:0:0:0:0:0:1', start='1:1::ff', end='1:1::ffff',
enable_dhcp=False, ipv6_ra_mode='slaac', ipv6_address_mode='slaac')
- self.network_settings = NetworkSettings(
+ self.network_settings = NetworkConfig(
name=self.guid + '-net', subnet_settings=[sub_setting])
with self.assertRaises(BadRequest):
with an invalid start IP to ensure Neutron assigns it the smallest IP
possible
"""
- sub_setting = SubnetSettings(
+ sub_setting = SubnetConfig(
name=self.guid + '-subnet', cidr='1:1::/48', ip_version=6,
start='foo')
- self.network_settings = NetworkSettings(
+ self.network_settings = NetworkConfig(
name=self.guid + '-net', subnet_settings=[sub_setting])
self.network = neutron_utils.create_network(
with an invalid end IP to ensure Neutron assigns it the largest IP
possible
"""
- sub_setting = SubnetSettings(
+ sub_setting = SubnetConfig(
name=self.guid + '-subnet', cidr='1:1::/48', ip_version=6,
end='bar')
- self.network_settings = NetworkSettings(
+ self.network_settings = NetworkConfig(
name=self.guid + '-net', subnet_settings=[sub_setting])
self.network = neutron_utils.create_network(
Tests the neutron_utils.create_network() for a BadRequest when
the subnet CIDR is invalid
"""
- sub_setting = SubnetSettings(
+ sub_setting = SubnetConfig(
name=self.guid + '-subnet', cidr='1:1:1:/48', ip_version=6)
- self.network_settings = NetworkSettings(
+ self.network_settings = NetworkConfig(
name=self.guid + '-net', subnet_settings=[sub_setting])
with self.assertRaises(BadRequest):
Tests the neutron_utils.create_network() for a BadRequest when
the subnet gateway IP is invalid
"""
- sub_setting = SubnetSettings(
+ sub_setting = SubnetConfig(
name=self.guid + '-subnet', cidr='1:1::/48', ip_version=6,
gateway_ip='1:2::1')
- self.network_settings = NetworkSettings(
+ self.network_settings = NetworkConfig(
name=self.guid + '-net', subnet_settings=[sub_setting])
with self.assertRaises(BadRequest):
Tests the neutron_utils.create_network() for a BadRequest when
the DNS IP is invalid
"""
- sub_setting = SubnetSettings(
+ sub_setting = SubnetConfig(
name=self.guid + '-subnet', cidr='1:1::/48', ip_version=6,
dns_nameservers=['foo'])
- self.network_settings = NetworkSettings(
+ self.network_settings = NetworkConfig(
name=self.guid + '-net', subnet_settings=[sub_setting])
with self.assertRaises(BadRequest):
self.neutron, subnet_setting.name, subnet_setting.cidr, True))
self.port = neutron_utils.create_port(
- self.neutron, self.os_creds, PortSettings(
+ self.neutron, self.os_creds, PortConfig(
name=self.port_name,
ip_addrs=[{
'subnet_name': subnet_setting.name,
subnet_setting.cidr, True))
self.port = neutron_utils.create_port(
- self.neutron, self.os_creds, PortSettings(
+ self.neutron, self.os_creds, PortConfig(
name=self.port_name,
network_name=self.net_config.network_settings.name,
ip_addrs=[{
self.port = neutron_utils.create_port(
self.neutron, self.os_creds,
- PortSettings(
+ PortConfig(
network_name=self.net_config.network_settings.name,
ip_addrs=[{
'subnet_name': subnet_setting.name,
with self.assertRaises(Exception):
self.port = neutron_utils.create_port(
self.neutron, self.os_creds,
- PortSettings(
+ PortConfig(
name=self.port_name,
network_name=self.net_config.network_settings.name,
ip_addrs=[{
with self.assertRaises(Exception):
self.port = neutron_utils.create_port(
self.neutron, self.os_creds,
- PortSettings(
+ PortConfig(
name=self.port_name,
network_name=self.net_config.network_settings.name,
ip_addrs=[{
with self.assertRaises(Exception):
self.port = neutron_utils.create_port(
self.neutron, self.os_creds,
- PortSettings(
+ PortConfig(
name=self.port_name,
network_name=self.net_config.network_settings.name,
ip_addrs=[{
with self.assertRaises(Exception):
self.port = neutron_utils.create_port(
self.neutron, self.os_creds,
- PortSettings(
+ PortConfig(
name=self.port_name,
network_name=self.net_config.network_settings.name,
ip_addrs=[{
import os
from snaps import file_utils
-from snaps.config.volume import VolumeConfig
from snaps.config.flavor import FlavorConfig
+from snaps.config.network import PortConfig
+from snaps.config.volume import VolumeConfig
from snaps.openstack import create_instance
from snaps.openstack.create_flavor import OpenStackFlavor
from snaps.openstack.create_image import OpenStackImage
from snaps.openstack.create_instance import (
VmInstanceSettings, OpenStackVmInstance)
-from snaps.openstack.create_network import OpenStackNetwork, PortSettings
+from snaps.openstack.create_network import OpenStackNetwork
from snaps.openstack.create_volume import OpenStackVolume
from snaps.openstack.tests import openstack_tests
from snaps.openstack.tests.os_source_file_test import OSComponentTestCase
name=guid + '-flavor-name', ram=256, disk=10, vcpus=1))
self.flavor_creator.create()
- port_settings = PortSettings(name=guid + '-port',
- network_name=network_settings.name)
+ port_settings = PortConfig(
+ name=guid + '-port', network_name=network_settings.name)
self.port = neutron_utils.create_port(
self.neutron, self.os_creds, port_settings)
self.os_creds, volume_settings)
self.volume_creator.create(block=True)
- port_settings = PortSettings(
+ port_settings = PortConfig(
name=guid + '-port', network_name=network_settings.name)
instance_settings = VmInstanceSettings(
name=guid + '-vm_inst',
import os
import uuid
+from snaps.config.network import SubnetConfig, NetworkConfig, PortConfig
from snaps.config.flavor import FlavorConfig
from snaps.config.keypair import KeypairConfig
from snaps.config.qos import Consumer
from snaps.openstack import (
create_image, create_network, create_router, create_flavor,
create_keypairs, create_instance)
-from snaps.openstack.create_network import (
- NetworkSettings, OpenStackNetwork, SubnetSettings)
+from snaps.openstack.create_qos import Consumer
+from snaps.openstack.create_network import OpenStackNetwork
from snaps.openstack.create_security_group import (
SecurityGroupRuleSettings, Direction, Protocol, OpenStackSecurityGroup,
SecurityGroupSettings)
class SettingsUtilsNetworkingTests(OSComponentTestCase):
"""
- Tests the ability to reverse engineer NetworkSettings objects from existing
+ Tests the ability to reverse engineer NetworkConfig objects from existing
networks deployed to OpenStack
"""
def test_derive_net_settings_no_subnet(self):
"""
- Validates the utility function settings_utils#create_network_settings
- returns an acceptable NetworkSettings object and ensures that the
+ Validates the utility function settings_utils#create_network_config
+ returns an acceptable NetworkConfig object and ensures that the
new settings object will not cause the new OpenStackNetwork instance
to create another network
"""
- net_settings = NetworkSettings(name=self.network_name)
+ net_settings = NetworkConfig(name=self.network_name)
self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
network = self.net_creator.create()
- derived_settings = settings_utils.create_network_settings(
+ derived_settings = settings_utils.create_network_config(
self.neutron, network)
self.assertIsNotNone(derived_settings)
def test_derive_net_settings_two_subnets(self):
"""
- Validates the utility function settings_utils#create_network_settings
- returns an acceptable NetworkSettings object
+ Validates the utility function settings_utils#create_network_config
+ returns an acceptable NetworkConfig object
"""
subnet_settings = list()
- subnet_settings.append(SubnetSettings(name='sub1', cidr='10.0.0.0/24'))
- subnet_settings.append(SubnetSettings(name='sub2', cidr='10.0.1.0/24'))
- net_settings = NetworkSettings(name=self.network_name,
- subnet_settings=subnet_settings)
+ subnet_settings.append(SubnetConfig(name='sub1', cidr='10.0.0.0/24'))
+ subnet_settings.append(SubnetConfig(name='sub2', cidr='10.0.1.0/24'))
+ net_settings = NetworkConfig(
+ name=self.network_name, subnet_settings=subnet_settings)
self.net_creator = OpenStackNetwork(self.os_creds, net_settings)
network = self.net_creator.create()
- derived_settings = settings_utils.create_network_settings(
+ derived_settings = settings_utils.create_network_config(
self.neutron, network)
self.assertIsNotNone(derived_settings)
# Create instance
ports_settings = list()
ports_settings.append(
- create_network.PortSettings(
+ PortConfig(
name=self.port_1_name,
network_name=self.pub_net_config.network_settings.name))
import pkg_resources
from scp import SCPClient
+from snaps.config.flavor import FlavorConfig
from snaps.config.keypair import KeypairConfig
+from snaps.config.network import PortConfig
-from snaps.config.flavor import FlavorConfig
from snaps.openstack import create_flavor
from snaps.openstack import create_image
from snaps.openstack import create_instance
# Create instance
ports_settings = list()
ports_settings.append(
- create_network.PortSettings(
+ PortConfig(
name=self.port_1_name,
network_name=self.pub_net_config.network_settings.name))
import logging
import unittest
+from snaps.config.tests.network_tests import (
+ NetworkConfigUnitTests, SubnetConfigUnitTests, PortConfigUnitTests)
from snaps.config.tests.volume_tests import VolumeConfigUnitTests
from snaps.config.tests.volume_type_tests import VolumeTypeConfigUnitTests
from snaps.config.tests.qos_tests import QoSConfigUnitTests
NetworkQuotasDomainObjectTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
RoleDomainObjectTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ NetworkConfigUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
NetworkSettingsUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
NetworkObjectTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ SubnetConfigUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
SubnetSettingsUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
SubnetObjectTests))
+ suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ PortConfigUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
PortSettingsUnitTests))
suite.addTest(unittest.TestLoader().loadTestsFromTestCase(
Code Review / snaps.git / commitdiff