from sqlalchemy.orm import exc
from sqlalchemy.sql import and_
+from neutron.api.v2 import attributes
from neutron.common import exceptions as n_exc
import neutron.db.api as db
from neutron.db import models_v2
msg = _("Invalid segment range. example range: 500-550")
raise n_exc.InvalidInput(error_message=msg)
+ def _validate_multicast_ip_range(self, network_profile):
+ """
+ Validate multicast ip range values.
+
+ :param network_profile: network profile object
+ """
+ try:
+ min_ip, max_ip = (network_profile
+ ['multicast_ip_range'].split('-', 1))
+ except ValueError:
+ msg = _("Invalid multicast ip address range. "
+ "example range: 224.1.1.1-224.1.1.10")
+ LOG.error(msg)
+ raise n_exc.InvalidInput(error_message=msg)
+ for ip in [min_ip, max_ip]:
+ try:
+ if not netaddr.IPAddress(ip).is_multicast():
+ msg = _("%s is not a valid multicast ip address") % ip
+ LOG.error(msg)
+ raise n_exc.InvalidInput(error_message=msg)
+ except netaddr.AddrFormatError:
+ msg = _("%s is not a valid ip address") % ip
+ LOG.error(msg)
+ raise n_exc.InvalidInput(error_message=msg)
+ if netaddr.IPAddress(min_ip) > netaddr.IPAddress(max_ip):
+ msg = (_("Invalid multicast IP range '%(min_ip)s-%(max_ip)s':"
+ " Range should be from low address to high address") %
+ {'min_ip': min_ip, 'max_ip': max_ip})
+ LOG.error(msg)
+ raise n_exc.InvalidInput(error_message=msg)
+
def _validate_network_profile(self, net_p):
"""
Validate completeness of a network profile arguments.
if segment_type == c_const.NETWORK_TYPE_OVERLAY:
if net_p['sub_type'] != c_const.NETWORK_SUBTYPE_NATIVE_VXLAN:
net_p['multicast_ip_range'] = '0.0.0.0'
+ else:
+ multicast_ip_range = net_p.get("multicast_ip_range")
+ if not attributes.is_attr_set(multicast_ip_range):
+ msg = _("Argument multicast_ip_range missing"
+ " for VXLAN multicast network profile")
+ LOG.error(msg)
+ raise n_exc.InvalidInput(error_message=msg)
+ self._validate_multicast_ip_range(net_p)
else:
net_p['multicast_ip_range'] = '0.0.0.0'
'segment_range': {'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': ''},
'multicast_ip_range': {'allow_post': True, 'allow_put': True,
- 'is_visible': True, 'default': '0.0.0.0'},
+ 'is_visible': True,
+ 'default': attributes.ATTR_NOT_SPECIFIED},
'multicast_ip_index': {'allow_post': False, 'allow_put': False,
'is_visible': False, 'default': '0'},
'physical_network': {'allow_post': True, 'allow_put': False,
:param network: network dict
:param network_profile: network profile dict
"""
- body = {'publishName': network['name'],
+ body = {'publishName': network['id'],
'description': network['name'],
'id': network['id'],
'tenantId': network['tenant_id'],
profile = n1kv_db_v2.get_network_profile(
db_session, network[n1kv.PROFILE_ID])
n1kvclient = n1kv_client.Client()
- body = {'publishName': network['name'],
+ body = {'description': network['name'],
'id': network['id'],
'networkSegmentPool': profile['id'],
'vlan': network[providernet.SEGMENTATION_ID],
net['id'], del_segments)
self._extend_network_dict_provider(context, net)
self._extend_network_dict_profile(context, net)
- if binding.network_type not in [c_const.NETWORK_TYPE_MULTI_SEGMENT]:
- self._send_update_network_request(context, net, add_segments,
- del_segments)
- LOG.debug(_("Updated network: %s"), net['id'])
- return net
+ if binding.network_type != c_const.NETWORK_TYPE_MULTI_SEGMENT:
+ self._send_update_network_request(context, net, add_segments,
+ del_segments)
+ LOG.debug(_("Updated network: %s"), net['id'])
+ return net
def delete_network(self, context, id):
"""
net_profile_id,
network_profile))
self._send_update_network_profile_request(net_p)
- return net_p
+ return net_p
netp['network_profile']['physical_network'] = 'phys1'
elif segment_type == 'overlay':
netp['network_profile']['segment_range'] = '10000-10010'
- netp['network_profile']['sub_type'] = 'enhanced'
+ netp['network_profile']['sub_type'] = 'enhanced' or 'native_vxlan'
+ netp['network_profile']['multicast_ip_range'] = ("224.1.1.1-"
+ "224.1.1.10")
return netp
def test_create_network_profile_plugin(self):
update_res = update_req.get_response(self.ext_api)
self.assertEqual(update_res.status_int, 400)
+ def test_create_overlay_network_profile_invalid_multicast_fail(self):
+ net_p_dict = self._prepare_net_profile_data('overlay')
+ data = {'network_profile': {'sub_type': 'native_vxlan',
+ 'multicast_ip_range': '1.1.1.1'}}
+ net_p_req = self.new_create_request('network_profiles', data,
+ net_p_dict)
+ res = net_p_req.get_response(self.ext_api)
+ self.assertEqual(res.status_int, 400)
+
+ def test_create_overlay_network_profile_no_multicast_fail(self):
+ net_p_dict = self._prepare_net_profile_data('overlay')
+ data = {'network_profile': {'sub_type': 'native_vxlan',
+ 'multicast_ip_range': ''}}
+ net_p_req = self.new_create_request('network_profiles', data,
+ net_p_dict)
+ res = net_p_req.get_response(self.ext_api)
+ self.assertEqual(res.status_int, 400)
+
+ def test_create_overlay_network_profile_wrong_split_multicast_fail(self):
+ net_p_dict = self._prepare_net_profile_data('overlay')
+ data = {'network_profile': {
+ 'sub_type': 'native_vxlan',
+ 'multicast_ip_range': '224.1.1.1.224.1.1.3'}}
+ net_p_req = self.new_create_request('network_profiles', data,
+ net_p_dict)
+ res = net_p_req.get_response(self.ext_api)
+ self.assertEqual(res.status_int, 400)
+
+ def test_create_overlay_network_profile_invalid_minip_multicast_fail(self):
+ net_p_dict = self._prepare_net_profile_data('overlay')
+ data = {'network_profile': {
+ 'sub_type': 'native_vxlan',
+ 'multicast_ip_range': '10.0.0.1-224.1.1.3'}}
+ net_p_req = self.new_create_request('network_profiles', data,
+ net_p_dict)
+ res = net_p_req.get_response(self.ext_api)
+ self.assertEqual(res.status_int, 400)
+
+ def test_create_overlay_network_profile_invalid_maxip_multicast_fail(self):
+ net_p_dict = self._prepare_net_profile_data('overlay')
+ data = {'network_profile': {
+ 'sub_type': 'native_vxlan',
+ 'multicast_ip_range': '224.1.1.1-20.0.0.1'}}
+ net_p_req = self.new_create_request('network_profiles', data,
+ net_p_dict)
+ res = net_p_req.get_response(self.ext_api)
+ self.assertEqual(res.status_int, 400)
+
+ def test_create_overlay_network_profile_correct_multicast_pass(self):
+ data = self._prepare_net_profile_data('overlay')
+ net_p_req = self.new_create_request('network_profiles', data)
+ res = net_p_req.get_response(self.ext_api)
+ self.assertEqual(res.status_int, 201)
+
class TestN1kvBasicGet(test_plugin.TestBasicGet,
N1kvPluginTestCase):