Merge "NSX|V3+P: Enable ipv6 in allowed address pairs"

This commit is contained in:
Zuul 2019-03-24 13:49:13 +00:00 committed by Gerrit Code Review
commit e73abc9d35
2 changed files with 17 additions and 4 deletions

View File

@ -249,10 +249,11 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
self._validate_address_scope_for_router_interface(
context.elevated(), router_db.id, gw_network_id, subnet['id'])
def _validate_ipv4_address_pairs(self, address_pairs):
def _validate_address_pairs(self, address_pairs):
for pair in address_pairs:
ip = pair.get('ip_address')
if not utils.is_ipv4_ip_address(ip):
if len(ip.split('/')) > 1:
LOG.error("cidr is not supported in allowed address pairs")
raise nsx_exc.InvalidIPAddress(ip_address=ip)
def _create_port_address_pairs(self, context, port_data):
@ -264,7 +265,7 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
if not port_security:
raise addr_exc.AddressPairAndPortSecurityRequired()
else:
self._validate_ipv4_address_pairs(address_pairs)
self._validate_address_pairs(address_pairs)
self._process_create_allowed_address_pairs(context, port_data,
address_pairs)
else:
@ -348,7 +349,7 @@ class NsxPluginV3Base(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
raise addr_exc.AddressPairAndPortSecurityRequired()
if delete_addr_pairs or has_addr_pairs:
self._validate_ipv4_address_pairs(
self._validate_address_pairs(
updated_port[addr_apidef.ADDRESS_PAIRS])
# delete address pairs and read them in
self._delete_allowed_address_pairs(context, id)

View File

@ -55,6 +55,18 @@ class TestAllowedAddressPairsNSXv3(test_v3_plugin.NsxV3PluginTestCaseMixin,
'ip_address': '10.0.0.1/24'}]
self._create_port_with_address_pairs(address_pairs, 400)
def test_create_port_allowed_address_pairs_v6(self):
with self.network() as net:
address_pairs = [{'ip_address': '1001::12'}]
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)
address_pairs[0]['mac_address'] = port['port']['mac_address']
self.assertEqual(port['port'][addr_apidef.ADDRESS_PAIRS],
address_pairs)
self._delete('ports', port['port']['id'])
def test_update_add_bad_address_pairs_with_cidr(self):
with self.network() as net:
res = self._create_port(self.fmt, net['network']['id'])