Browse Source

Revert "NSX|V3+P: Address pair ip cannot duplicate the port fixed ip"

This reverts commit 7d4e6ee4cd.

Change-Id: I3f0b88e12f1e8d87bbdecc0f045db2ba002ede03
changes/95/741095/1
Adit Sarfaty 3 weeks ago
parent
commit
ace1638888
3 changed files with 3 additions and 90 deletions
  1. +3
    -25
      vmware_nsx/plugins/common_v3/plugin.py
  2. +0
    -24
      vmware_nsx/tests/unit/extensions/test_addresspairs.py
  3. +0
    -41
      vmware_nsx/tests/unit/nsx_v3/test_plugin.py

+ 3
- 25
vmware_nsx/plugins/common_v3/plugin.py View File

@@ -326,14 +326,7 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
"""Can be implemented by each plugin"""
return False

def _validate_address_pairs(self, address_pairs, fixed_ips=None):
port_ips = []
pairs_ips = []
if fixed_ips:
# Make sure there are no duplications
for fixed_ip in fixed_ips:
port_ips.append(fixed_ip['ip_address'])

def _validate_address_pairs(self, address_pairs):
for pair in address_pairs:
ip = pair.get('ip_address')
if ':' in ip:
@@ -348,14 +341,6 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
err_msg = (_("Allowed address pairs Cidr %s cannot "
"have host bits set") % ip)
raise n_exc.InvalidInput(error_message=err_msg)
# verify no overlaps in ipv6 addresses
current_set = netaddr.IPSet(port_ips + pairs_ips)
if netaddr.IPSet([ip]) & current_set:
err_msg = (_("Allowed address pairs %s cannot overlap "
"with port ips or other address pairs") % ip)
raise n_exc.InvalidInput(error_message=err_msg)

pairs_ips.append(ip)
else:
# IPv4 address pair
if len(ip.split('/')) > 1 and ip.split('/')[1] != '32':
@@ -373,10 +358,6 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
LOG.error("Cidr %s is not supported in allowed "
"address pairs", ip)
raise nsx_exc.InvalidIPAddress(ip_address=ip)
if ip in port_ips:
err_msg = (_("Port cannot have duplicate values %s as part of "
"port manual bindings") % ip)
raise n_exc.InvalidInput(error_message=err_msg)
if ip in ['127.0.0.0', '0.0.0.0', '::']:
LOG.error("IP %s is not supported in allowed address "
"pairs", ip)
@@ -410,8 +391,7 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
if not port_security:
raise addr_exc.AddressPairAndPortSecurityRequired()
else:
self._validate_address_pairs(
address_pairs, fixed_ips=port_data.get('fixed_ips'))
self._validate_address_pairs(address_pairs)
self._validate_number_of_address_pairs(port_data)
self._process_create_allowed_address_pairs(context, port_data,
address_pairs)
@@ -493,9 +473,7 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,

if delete_addr_pairs or has_addr_pairs:
self._validate_address_pairs(
updated_port[addr_apidef.ADDRESS_PAIRS],
fixed_ips=(updated_port.get('fixed_ips') or
port_data.get('fixed_ips')))
updated_port[addr_apidef.ADDRESS_PAIRS])
# delete address pairs and read them in
self._delete_allowed_address_pairs(context, id)
self._process_create_allowed_address_pairs(


+ 0
- 24
vmware_nsx/tests/unit/extensions/test_addresspairs.py View File

@@ -40,9 +40,6 @@ class TestAllowedAddressPairsNSXv2(test_v3_plugin.NsxV3PluginTestCaseMixin,
def test_create_port_security_false_allowed_address_pairs(self):
self.skipTest('TBD')

def test_create_overlap_with_fixed_ip(self):
self.skipTest('Not supported')


class TestAllowedAddressPairsNSXp(test_p_plugin.NsxPPluginTestCaseMixin,
ext_pairs.TestAllowedAddressPairs):
@@ -90,24 +87,6 @@ class TestAllowedAddressPairsNSXp(test_p_plugin.NsxPPluginTestCaseMixin,
port = self.deserialize(self.fmt, res)
self.assertIn('NeutronError', port)

# overlapping ips
address_pairs = [{'ip_address': '1001::/64'},
{'ip_address': '1001::/128'}]
res = self._create_port(self.fmt, net['network']['id'],
arg_list=(addr_apidef.ADDRESS_PAIRS,),
allowed_address_pairs=address_pairs)
port = self.deserialize(self.fmt, res)
self.assertIn('NeutronError', port)

# identical ips
address_pairs = [{'ip_address': '1001::'},
{'ip_address': '1001::/128'}]
res = self._create_port(self.fmt, net['network']['id'],
arg_list=(addr_apidef.ADDRESS_PAIRS,),
allowed_address_pairs=address_pairs)
port = self.deserialize(self.fmt, res)
self.assertIn('NeutronError', port)

def test_update_add_bad_address_pairs_with_cidr(self):
with self.network() as net:
res = self._create_port(self.fmt, net['network']['id'])
@@ -201,9 +180,6 @@ class TestAllowedAddressPairsNSXv3(test_v3_plugin.NsxV3PluginTestCaseMixin,
def test_create_port_security_false_allowed_address_pairs(self):
self.skipTest('TBD')

def test_create_overlap_with_fixed_ip(self):
self.skipTest('Not supported')


class TestAllowedAddressPairsNSXv(test_nsx_v_plugin.NsxVPluginV2TestCase,
ext_pairs.TestAllowedAddressPairs):


+ 0
- 41
vmware_nsx/tests/unit/nsx_v3/test_plugin.py View File

@@ -54,7 +54,6 @@ from oslo_utils import uuidutils
from webob import exc

from vmware_nsx.api_client import exception as api_exc
from vmware_nsx.common import exceptions as nsx_exc
from vmware_nsx.common import utils
from vmware_nsx.db import db as nsx_db
from vmware_nsx.plugins.nsx_v3 import plugin as nsx_plugin
@@ -1130,46 +1129,6 @@ class TestPortsV2(common_v3.NsxV3SubnetMixin,
self.assertRaises(n_exc.InvalidInput,
self.plugin.create_port, self.ctx, data)

def test_fail_create_allowed_address_pairs_dup(self):
with self.network() as network, self.subnet(
network=network, cidr="1.1.1.0/24",
enable_dhcp=True) as s1:
data = {
'port': {
'network_id': network['network']['id'],
'tenant_id': self._tenant_id,
'name': 'pair_port',
'admin_state_up': True,
'device_id': 'fake_device',
'device_owner': 'fake_owner',
'fixed_ips': [{'subnet_id': s1['subnet']['id'],
'ip_address': '1.1.1.30'}]
}
}
data['port']['allowed_address_pairs'] = [
{'ip_address': '1.1.1.30'}]
self.assertRaises(n_exc.InvalidInput,
self.plugin.create_port, self.ctx, data)

def test_fail_create_allowed_address_pairs_illegal_ip(self):
with self.network() as network, self.subnet(
network=network, enable_dhcp=True) as s1:
data = {
'port': {
'network_id': network['network']['id'],
'tenant_id': self._tenant_id,
'name': 'pair_port',
'admin_state_up': True,
'device_id': 'fake_device',
'device_owner': 'fake_owner',
'fixed_ips': [{'subnet_id': s1['subnet']['id']}]
}
}
data['port']['allowed_address_pairs'] = [
{'ip_address': '127.0.0.0'}]
self.assertRaises(nsx_exc.InvalidIPAddress,
self.plugin.create_port, self.ctx, data)

def test_fail_update_lb_port_with_fixed_ip(self):
with self.network() as network:
data = {'port': {


Loading…
Cancel
Save