openstack
/
neutron
Code Issues Proposed changes

Merge "Raise proper exception in case duplicate ipv6 address is allocated"

This commit is contained in:
Jenkins 2014-08-25 18:26:11 +00:00 committed by Gerrit Code Review
parent 8fde2e37f0 eb3be39521
commit b52e6c3afb
3 changed files with 23 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
neutron/db/db_base_plugin_v2.py
View File

@ -522,6 +522,12 @@ class NeutronDbPluginV2(neutron_plugin_base_v2.NeutronPluginBaseV2,
prefix = subnet['cidr']
ip_address = ipv6_utils.get_ipv6_addr_by_EUI64(
prefix, mac)
if not self._check_unique_ip(
context, p['network_id'],
subnet['id'], ip_address.format()):
raise n_exc.IpAddressInUse(
net_id=p['network_id'],
ip_address=ip_address.format())
ips.append({'ip_address': ip_address.format(),
'subnet_id': subnet['id']})
v6.remove(subnet)

3
neutron/tests/unit/oneconvergence/test_nvsd_plugin.py
View File

@ -124,6 +124,9 @@ class TestOneConvergencePluginPortsV2(test_plugin.TestPortsV2,
def test_ip_allocation_for_ipv6_subnet_slaac_adddress_mode(self):
self.skipTest("NVSD Plugin does not support IPV6.")
def test_generated_duplicate_ip_ipv6(self):
self.skipTest("NVSD Plugin does not support IPV6.")
class TestOneConvergenceBasicGet(test_plugin.TestBasicGet,
OneConvergencePluginV2TestCase):

14
neutron/tests/unit/test_db_plugin.py
View File

@ -1302,6 +1302,20 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
res = self._create_port(self.fmt, net_id=net_id, **kwargs)
self.assertEqual(res.status_int, webob.exc.HTTPConflict.code)
def test_generated_duplicate_ip_ipv6(self):
with self.subnet(ip_version=6,
cidr="2014::/64",
ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
with self.port(subnet=subnet,
fixed_ips=[{'subnet_id': subnet['subnet']['id'],
'ip_address':
"2014::1322:33ff:fe44:5566"}]) as port:
# Check configuring of duplicate IP
kwargs = {"mac_address": "11:22:33:44:55:66"}
net_id = port['port']['network_id']
res = self._create_port(self.fmt, net_id=net_id, **kwargs)
self.assertEqual(res.status_int, webob.exc.HTTPConflict.code)
def test_requested_subnet_delete(self):
with self.subnet() as subnet:
with self.port(subnet=subnet) as port: