'default': True,
'convert_to': convert_to_boolean,
'is_visible': True},
+ 'ipv6_ra_mode': {'allow_post': True, 'allow_put': True,
+ 'default': ATTR_NOT_SPECIFIED,
+ 'validate': {'type:values': constants.IPV6_MODES},
+ 'is_visible': True},
+ 'ipv6_address_mode': {'allow_post': True, 'allow_put': True,
+ 'default': ATTR_NOT_SPECIFIED,
+ 'validate': {'type:values':
+ constants.IPV6_MODES},
+ 'is_visible': True},
SHARED: {'allow_post': False,
'allow_put': False,
'default': False,
# Neighbor Solicitation (135),
# Neighbor Advertisement (136)
ICMPV6_ALLOWED_TYPES = [130, 131, 132, 134, 135, 136]
+
+DHCPV6_STATEFUL = 'dhcpv6-stateful'
+DHCPV6_STATELESS = 'dhcpv6-stateless'
+IPV6_SLAAC = 'slaac'
+IPV6_MODES = [DHCPV6_STATEFUL, DHCPV6_STATELESS, IPV6_SLAAC]
tenant_ids.pop() != original.tenant_id):
raise n_exc.InvalidSharedSetting(network=original.name)
+ def _validate_ipv6_attributes(self, subnet, cur_subnet):
+ ra_mode_set = attributes.is_attr_set(subnet.get('ipv6_ra_mode'))
+ address_mode_set = attributes.is_attr_set(
+ subnet.get('ipv6_address_mode'))
+ if cur_subnet:
+ ra_mode = (subnet['ipv6_ra_mode'] if ra_mode_set
+ else cur_subnet['ipv6_ra_mode'])
+ addr_mode = (subnet['ipv6_address_mode'] if address_mode_set
+ else cur_subnet['ipv6_address_mode'])
+ if ra_mode_set or address_mode_set:
+ # Check that updated subnet ipv6 attributes do not conflict
+ self._validate_ipv6_combination(ra_mode, addr_mode)
+ self._validate_ipv6_update_dhcp(subnet, cur_subnet)
+ else:
+ self._validate_ipv6_dhcp(ra_mode_set, address_mode_set,
+ subnet['enable_dhcp'])
+ if ra_mode_set and address_mode_set:
+ self._validate_ipv6_combination(subnet['ipv6_ra_mode'],
+ subnet['ipv6_address_mode'])
+
+ def _validate_ipv6_combination(self, ra_mode, address_mode):
+ if ra_mode != address_mode:
+ msg = _("ipv6_ra_mode set to '%(ra_mode)s' with ipv6_address_mode "
+ "set to '%(addr_mode)s' is not valid. "
+ "If both attributes are set, they must be the same value"
+ ) % {'ra_mode': ra_mode, 'addr_mode': address_mode}
+ raise n_exc.InvalidInput(error_message=msg)
+
+ def _validate_ipv6_dhcp(self, ra_mode_set, address_mode_set, enable_dhcp):
+ if (ra_mode_set or address_mode_set) and not enable_dhcp:
+ msg = _("ipv6_ra_mode or ipv6_address_mode cannot be set when "
+ "enable_dhcp is set to False.")
+ raise n_exc.InvalidInput(error_message=msg)
+
+ def _validate_ipv6_update_dhcp(self, subnet, cur_subnet):
+ if ('enable_dhcp' in subnet and not subnet['enable_dhcp']):
+ msg = _("Cannot disable enable_dhcp with "
+ "ipv6 attributes set")
+
+ ra_mode_set = attributes.is_attr_set(subnet.get('ipv6_ra_mode'))
+ address_mode_set = attributes.is_attr_set(
+ subnet.get('ipv6_address_mode'))
+
+ if ra_mode_set or address_mode_set:
+ raise n_exc.InvalidInput(error_message=msg)
+
+ old_ra_mode_set = attributes.is_attr_set(
+ cur_subnet.get('ipv6_ra_mode'))
+ old_address_mode_set = attributes.is_attr_set(
+ cur_subnet.get('ipv6_address_mode'))
+
+ if old_ra_mode_set or old_address_mode_set:
+ raise n_exc.InvalidInput(error_message=msg)
+
def _make_network_dict(self, network, fields=None,
process_extensions=True):
res = {'id': network['id'],
for pool in subnet['allocation_pools']],
'gateway_ip': subnet['gateway_ip'],
'enable_dhcp': subnet['enable_dhcp'],
+ 'ipv6_ra_mode': subnet['ipv6_ra_mode'],
+ 'ipv6_address_mode': subnet['ipv6_address_mode'],
'dns_nameservers': [dns['address']
for dns in subnet['dns_nameservers']],
'host_routes': [{'destination': route['destination'],
for rt in s['host_routes']:
self._validate_host_route(rt, ip_ver)
+ if ip_ver == 6:
+ self._validate_ipv6_attributes(s, cur_subnet)
+
def _validate_gw_out_of_pools(self, gateway_ip, pools):
for allocation_pool in pools:
pool_range = netaddr.IPRange(
'enable_dhcp': s['enable_dhcp'],
'gateway_ip': s['gateway_ip'],
'shared': network.shared}
+ if s['ip_version'] == 6 and s['enable_dhcp']:
+ if attributes.is_attr_set(s['ipv6_ra_mode']):
+ args['ipv6_ra_mode'] = s['ipv6_ra_mode']
+ if attributes.is_attr_set(s['ipv6_address_mode']):
+ args['ipv6_address_mode'] = s['ipv6_address_mode']
subnet = models_v2.Subnet(**args)
context.session.add(subnet)
--- /dev/null
+# Copyright 2014 OpenStack Foundation
+#
+# 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.
+#
+# @author Sean M. Collins (Comcast)
+
+"""Add IPv6 Subnet properties
+
+Revision ID: 2447ad0e9585
+Revises: 33dd0a9fa487
+Create Date: 2013-10-23 16:36:44.188904
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '2447ad0e9585'
+down_revision = '33dd0a9fa487'
+
+# Change to ['*'] if this migration applies to all plugins
+
+migration_for_plugins = [
+ '*'
+]
+
+from alembic import op
+import sqlalchemy as sa
+
+
+from neutron.db import migration
+
+
+def upgrade(active_plugins=None, options=None):
+ if not migration.should_run(active_plugins, migration_for_plugins):
+ return
+
+ # Workaround for Alemic bug #89
+ # https://bitbucket.org/zzzeek/alembic/issue/89
+ context = op.get_context()
+ if context.bind.dialect.name == 'postgresql':
+ op.execute("CREATE TYPE ipv6_modes AS ENUM ('%s', '%s', '%s')"
+ % ('slaac', 'dhcpv6-stateful', 'dhcpv6-stateless'))
+ op.add_column('subnets',
+ sa.Column('ipv6_ra_mode',
+ sa.Enum('slaac',
+ 'dhcpv6-stateful',
+ 'dhcpv6-stateless',
+ name='ipv6_modes'),
+ nullable=True)
+ )
+ op.add_column('subnets',
+ sa.Column('ipv6_address_mode',
+ sa.Enum('slaac',
+ 'dhcpv6-stateful',
+ 'dhcpv6-stateless',
+ name='ipv6_modes'),
+ nullable=True)
+ )
+
+
+def downgrade(active_plugins=None, options=None):
+ if not migration.should_run(active_plugins, migration_for_plugins):
+ return
+
+ op.drop_column('subnets', 'ipv6_ra_mode')
+ op.drop_column('subnets', 'ipv6_address_mode')
+ context = op.get_context()
+ if context.bind.dialect.name == 'postgresql':
+ op.execute('DROP TYPE ipv6_modes')
import sqlalchemy as sa
from sqlalchemy import orm
+from neutron.common import constants
from neutron.db import model_base
from neutron.openstack.common import uuidutils
backref='subnet',
cascade='all, delete, delete-orphan')
shared = sa.Column(sa.Boolean)
+ ipv6_ra_mode = sa.Column(sa.Enum(constants.IPV6_SLAAC,
+ constants.DHCPV6_STATEFUL,
+ constants.DHCPV6_STATELESS,
+ name='ipv6_modes'), nullable=True)
+ ipv6_address_mode = sa.Column(sa.Enum(constants.IPV6_SLAAC,
+ constants.DHCPV6_STATEFUL,
+ constants.DHCPV6_STATELESS,
+ name='ipv6_modes'), nullable=True)
class Network(model_base.BASEV2, HasId, HasTenant):
def test_update_subnet_inconsistent_ipv6_hostroute_np_v4(self):
self.skipTest("NVSD Plugin does not support IPV6.")
+ def test_create_subnet_ipv6_attributes(self):
+ self.skipTest("NVSD Plugin does not support IPV6.")
+
+ def test_create_subnet_ipv6_single_attribute_set(self):
+ self.skipTest("NVSD Plugin does not support IPV6.")
+
+ def test_update_subnet_ipv6_attributes(self):
+ self.skipTest("NVSD Plugin does not support IPV6.")
+
+ def test_update_subnet_ipv6_inconsistent_enable_dhcp(self):
+ self.skipTest("NVSD Plugin does not support IPV6.")
+
+ def test_update_subnet_ipv6_inconsistent_ra_attribute(self):
+ self.skipTest("NVSD Plugin does not support IPV6.")
+
+ def test_update_subnet_ipv6_inconsistent_address_attribute(self):
+ self.skipTest("NVSD Plugin does not support IPV6.")
+
class TestOneConvergencePluginPortsV2(test_plugin.TestPortsV2,
test_bindings.PortBindingsTestCase,
for arg in ('ip_version', 'tenant_id',
'enable_dhcp', 'allocation_pools',
'dns_nameservers', 'host_routes',
- 'shared'):
+ 'shared', 'ipv6_ra_mode', 'ipv6_address_mode'):
# Arg must be present and not null (but can be false)
if arg in kwargs and kwargs[arg] is not None:
data['subnet'][arg] = kwargs[arg]
def _make_subnet(self, fmt, network, gateway, cidr,
allocation_pools=None, ip_version=4, enable_dhcp=True,
- dns_nameservers=None, host_routes=None, shared=None):
+ dns_nameservers=None, host_routes=None, shared=None,
+ ipv6_ra_mode=None, ipv6_address_mode=None):
res = self._create_subnet(fmt,
net_id=network['network']['id'],
cidr=cidr,
enable_dhcp=enable_dhcp,
dns_nameservers=dns_nameservers,
host_routes=host_routes,
- shared=shared)
+ shared=shared,
+ ipv6_ra_mode=ipv6_ra_mode,
+ ipv6_address_mode=ipv6_address_mode)
# Things can go wrong - raise HTTP exc with res code only
# so it can be caught by unit tests
if res.status_int >= webob.exc.HTTPClientError.code:
dns_nameservers=None,
host_routes=None,
shared=None,
- do_delete=True):
+ do_delete=True,
+ ipv6_ra_mode=None,
+ ipv6_address_mode=None):
with optional_ctx(network, self.network) as network_to_use:
subnet = self._make_subnet(fmt or self.fmt,
network_to_use,
enable_dhcp,
dns_nameservers,
host_routes,
- shared=shared)
+ shared=shared,
+ ipv6_ra_mode=ipv6_ra_mode,
+ ipv6_address_mode=ipv6_address_mode)
try:
yield subnet
finally:
res = subnet_req.get_response(self.api)
self.assertEqual(res.status_int, webob.exc.HTTPClientError.code)
+ def test_create_subnet_ipv6_attributes(self):
+ gateway_ip = 'fe80::1'
+ cidr = 'fe80::/80'
+
+ for mode in constants.IPV6_MODES:
+ self._test_create_subnet(gateway_ip=gateway_ip,
+ cidr=cidr, ip_version=6,
+ ipv6_ra_mode=mode,
+ ipv6_address_mode=mode)
+
+ def test_create_subnet_ipv6_attributes_no_dhcp_enabled(self):
+ gateway_ip = 'fe80::1'
+ cidr = 'fe80::/80'
+ with testlib_api.ExpectedException(
+ webob.exc.HTTPClientError) as ctx_manager:
+ for mode in constants.IPV6_MODES:
+ self._test_create_subnet(gateway_ip=gateway_ip,
+ cidr=cidr, ip_version=6,
+ enable_dhcp=False,
+ ipv6_ra_mode=mode,
+ ipv6_address_mode=mode)
+ self.assertEqual(ctx_manager.exception.code,
+ webob.exc.HTTPClientError.code)
+
+ def test_create_subnet_invalid_ipv6_ra_mode(self):
+ gateway_ip = 'fe80::1'
+ cidr = 'fe80::/80'
+ with testlib_api.ExpectedException(
+ webob.exc.HTTPClientError) as ctx_manager:
+ self._test_create_subnet(gateway_ip=gateway_ip,
+ cidr=cidr, ip_version=6,
+ ipv6_ra_mode='foo',
+ ipv6_address_mode='slaac')
+ self.assertEqual(ctx_manager.exception.code,
+ webob.exc.HTTPClientError.code)
+
+ def test_create_subnet_invalid_ipv6_address_mode(self):
+ gateway_ip = 'fe80::1'
+ cidr = 'fe80::/80'
+ with testlib_api.ExpectedException(
+ webob.exc.HTTPClientError) as ctx_manager:
+ self._test_create_subnet(gateway_ip=gateway_ip,
+ cidr=cidr, ip_version=6,
+ ipv6_ra_mode='slaac',
+ ipv6_address_mode='baz')
+ self.assertEqual(ctx_manager.exception.code,
+ webob.exc.HTTPClientError.code)
+
+ def test_create_subnet_invalid_ipv6_combination(self):
+ gateway_ip = 'fe80::1'
+ cidr = 'fe80::/80'
+ with testlib_api.ExpectedException(
+ webob.exc.HTTPClientError) as ctx_manager:
+ self._test_create_subnet(gateway_ip=gateway_ip,
+ cidr=cidr, ip_version=6,
+ ipv6_ra_mode='stateful',
+ ipv6_address_mode='stateless')
+ self.assertEqual(ctx_manager.exception.code,
+ webob.exc.HTTPClientError.code)
+
+ def test_create_subnet_ipv6_single_attribute_set(self):
+ gateway_ip = 'fe80::1'
+ cidr = 'fe80::/80'
+ for mode in constants.IPV6_MODES:
+ self._test_create_subnet(gateway_ip=gateway_ip,
+ cidr=cidr, ip_version=6,
+ ipv6_ra_mode=None,
+ ipv6_address_mode=mode)
+ self._test_create_subnet(gateway_ip=gateway_ip,
+ cidr=cidr, ip_version=6,
+ ipv6_ra_mode=mode,
+ ipv6_address_mode=None)
+
def test_update_subnet_no_gateway(self):
with self.subnet() as subnet:
data = {'subnet': {'gateway_ip': '11.0.0.1'}}
self.assertEqual(res.status_int,
webob.exc.HTTPConflict.code)
+ def test_update_subnet_ipv6_attributes(self):
+ with self.subnet(ip_version=6, cidr='fe80::/80',
+ ipv6_ra_mode=constants.IPV6_SLAAC,
+ ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
+ data = {'subnet': {'ipv6_ra_mode': constants.DHCPV6_STATEFUL,
+ 'ipv6_address_mode': constants.DHCPV6_STATEFUL}}
+ req = self.new_update_request('subnets', data,
+ subnet['subnet']['id'])
+ res = self.deserialize(self.fmt, req.get_response(self.api))
+ self.assertEqual(res['subnet']['ipv6_ra_mode'],
+ data['subnet']['ipv6_ra_mode'])
+ self.assertEqual(res['subnet']['ipv6_address_mode'],
+ data['subnet']['ipv6_address_mode'])
+
+ def test_update_subnet_ipv6_inconsistent_ra_attribute(self):
+ with self.subnet(ip_version=6, cidr='fe80::/80',
+ ipv6_ra_mode=constants.IPV6_SLAAC,
+ ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
+ data = {'subnet': {'ipv6_ra_mode': constants.DHCPV6_STATEFUL}}
+ req = self.new_update_request('subnets', data,
+ subnet['subnet']['id'])
+ res = req.get_response(self.api)
+ self.assertEqual(res.status_int,
+ webob.exc.HTTPClientError.code)
+
+ def test_update_subnet_ipv6_inconsistent_address_attribute(self):
+ with self.subnet(ip_version=6, cidr='fe80::/80',
+ ipv6_ra_mode=constants.IPV6_SLAAC,
+ ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
+ data = {'subnet': {'ipv6_address_mode': constants.DHCPV6_STATEFUL}}
+ req = self.new_update_request('subnets', data,
+ subnet['subnet']['id'])
+ res = req.get_response(self.api)
+ self.assertEqual(res.status_int,
+ webob.exc.HTTPClientError.code)
+
+ def test_update_subnet_ipv6_inconsistent_enable_dhcp(self):
+ with self.subnet(ip_version=6, cidr='fe80::/80',
+ ipv6_ra_mode=constants.IPV6_SLAAC,
+ ipv6_address_mode=constants.IPV6_SLAAC) as subnet:
+ data = {'subnet': {'enable_dhcp': False}}
+ req = self.new_update_request('subnets', data,
+ subnet['subnet']['id'])
+ res = req.get_response(self.api)
+ self.assertEqual(res.status_int,
+ webob.exc.HTTPClientError.code)
+
def test_show_subnet(self):
with self.network() as network:
with self.subnet(network=network) as subnet:
Code Review / openstack-build / neutron-build.git / commitdiff