openstack
/
neutron
Code Issues Proposed changes

Switch to use _get_subnet_object() in ipam_backend_mixin module 

This patch switches to use _get_subnet_object() instead of
_get_subnet() method in ipam_backend_mixin module to use Subnet
OVO instead of db model class directly.

Change-Id: Ibdc91ba20b560dc42319565b5f4aeadba0dfcb29
Partially-Implements: blueprint adopt-oslo-versioned-objects-for-db
This commit is contained in:
Sławek Kapłoński 2017-12-11 15:54:31 +01:00
parent 259eedaaba
commit 8654749082
4 changed files with 56 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
neutron/db/db_base_plugin_v2.py
View File

@ -835,9 +835,7 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
# value since _validate_subnet() expects subnet spec has 'ip_version'
# and 'allocation_pools' fields.
s['ip_version'] = subnet_obj.ip_version
# TODO(slaweq): convert cidr to str will not be necessary when we
# will switch to use subnet OVO in ipam module
s['cidr'] = str(subnet_obj.cidr)
s['cidr'] = subnet_obj.cidr
s['id'] = subnet_obj.id
s['project_id'] = subnet_obj.project_id
s['tenant_id'] = subnet_obj.project_id
@ -849,10 +847,10 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
if new_cidr and ipv6_utils.is_ipv6_pd_enabled(s):
# This is an ipv6 prefix delegation-enabled subnet being given an
# updated cidr by the process_prefix_update RPC
s['cidr'] = new_cidr
net = netaddr.IPNetwork(s['cidr'], s['ip_version'])
s['cidr'] = netaddr.IPNetwork(new_cidr, s['ip_version'])
# Update gateway_ip and allocation pools based on new cidr
s['gateway_ip'] = utils.get_first_host_ip(net, s['ip_version'])
s['gateway_ip'] = utils.get_first_host_ip(
s['cidr'], s['ip_version'])
s['allocation_pools'] = self._update_allocation_pools(s)
range_pools = None

20
neutron/db/ipam_backend_mixin.py
View File

@ -209,9 +209,10 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
changes['service_types'] = (
self._update_subnet_service_types(context, subnet_id, s))
subnet = self._get_subnet(context, subnet_id)
subnet.update(s)
return subnet, changes
subnet_obj = self._get_subnet_object(context, subnet_id)
subnet_obj.update_fields(s)
subnet_obj.update()
return subnet_obj, changes
def _validate_subnet_cidr(self, context, network, new_subnet_cidr):
"""Validate the CIDR for a subnet.
@ -238,7 +239,7 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
subnet_list = self._get_subnets(context)
for subnet in subnet_list:
if ((netaddr.IPSet([subnet.cidr]) & new_subnet_ipset) and
subnet.cidr != const.PROVISIONAL_IPV6_PD_PREFIX):
str(subnet.cidr) != const.PROVISIONAL_IPV6_PD_PREFIX):
# don't give out details of the overlapping subnet
err_msg = ("Requested subnet with cidr: %(cidr)s for "
"network: %(network_id)s overlaps with another "
@ -347,11 +348,12 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
return subnet
subnet = get_matching_subnet()
if not subnet:
subnet = self._get_subnet(context, fixed['subnet_id'])
subnet_obj = self._get_subnet_object(context,
fixed['subnet_id'])
msg = (_("Failed to create port on network %(network_id)s"
", because fixed_ips included invalid subnet "
"%(subnet_id)s") %
{'network_id': subnet['network_id'],
{'network_id': subnet_obj.network_id,
'subnet_id': fixed['subnet_id']})
raise exc.InvalidInput(error_message=msg)
# Ensure that the IP is valid on the subnet
@ -400,9 +402,9 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
if device_owner in const.ROUTER_INTERFACE_OWNERS_SNAT:
return True
subnet = self._get_subnet(context, subnet_id)
return not (ipv6_utils.is_auto_address_subnet(subnet) and
not ipv6_utils.is_ipv6_pd_enabled(subnet))
subnet_obj = self._get_subnet_object(context, subnet_id)
return not (ipv6_utils.is_auto_address_subnet(subnet_obj) and
not ipv6_utils.is_ipv6_pd_enabled(subnet_obj))
def _get_changed_ips_for_port(self, context, original_ips,
new_ips, device_owner):

5
neutron/tests/unit/api/rpc/handlers/test_l3_rpc.py
View File

@ -13,6 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import netaddr
from neutron_lib import constants
from neutron_lib import context
from neutron_lib.plugins import directory
@ -58,10 +59,10 @@ class TestL3RpcCallback(testlib_api.SqlTestCase):
def test_process_prefix_update(self):
subnet = self._prepare_ipv6_pd_subnet()
data = {subnet['id']: '2001:db8::/64'}
data = {subnet['id']: netaddr.IPNetwork('2001:db8::/64')}
allocation_pools = [{'start': '2001:db8::2',
'end': '2001:db8::ffff:ffff:ffff:ffff'}]
res = self.callbacks.process_prefix_update(self.ctx, subnets=data)
updated_subnet = res[0]
self.assertEqual(updated_subnet['cidr'], data[subnet['id']])
self.assertEqual(str(data[subnet['id']]), updated_subnet['cidr'])
self.assertEqual(updated_subnet['allocation_pools'], allocation_pools)

60
neutron/tests/unit/db/test_ipam_backend_mixin.py
View File

@ -17,11 +17,13 @@ import mock
import netaddr
from neutron_lib.api.definitions import portbindings
from neutron_lib import constants
from oslo_utils import uuidutils
import webob.exc
from neutron.db import db_base_plugin_v2
from neutron.db import ipam_backend_mixin
from neutron.db import portbindings_db
from neutron.objects import subnet as subnet_obj
from neutron.tests import base
from neutron.tests.unit.db import test_db_base_plugin_v2
@ -50,26 +52,37 @@ class TestIpamBackendMixin(base.BaseTestCase):
return results
def _mock_slaac_subnet_on(self):
slaac_subnet = {'ipv6_address_mode': constants.IPV6_SLAAC,
'ipv6_ra_mode': constants.IPV6_SLAAC}
self.mixin._get_subnet = mock.Mock(return_value=slaac_subnet)
slaac_subnet_obj = subnet_obj.Subnet(
self.ctx,
ipv6_address_mode=constants.IPV6_SLAAC,
ipv6_ra_mode=constants.IPV6_SLAAC)
self.mixin._get_subnet_object = mock.Mock(
return_value=slaac_subnet_obj)
def _mock_slaac_subnet_off(self):
non_slaac_subnet = {'ipv6_address_mode': None,
'ipv6_ra_mode': None}
self.mixin._get_subnet = mock.Mock(return_value=non_slaac_subnet)
non_slaac_subnet_obj = subnet_obj.Subnet(
self.ctx,
ipv6_address_mode=None,
ipv6_ra_mode=None)
self.mixin._get_subnet_object = mock.Mock(
return_value=non_slaac_subnet_obj)
def _mock_slaac_for_subnet_ids(self, subnet_ids):
"""Mock incoming subnets as autoaddressed."""
def _get_subnet(context, subnet_id):
def _get_subnet_object(context, subnet_id):
if subnet_id in subnet_ids:
return {'ipv6_address_mode': constants.IPV6_SLAAC,
'ipv6_ra_mode': constants.IPV6_SLAAC}
return subnet_obj.Subnet(
self.ctx,
ipv6_address_mode=constants.IPV6_SLAAC,
ipv6_ra_mode=constants.IPV6_SLAAC)
else:
return {'ipv6_address_mode': None,
'ipv6_ra_mode': None}
return subnet_obj.Subnet(
self.ctx,
ipv6_address_mode=None,
ipv6_ra_mode=None)
self.mixin._get_subnet = mock.Mock(side_effect=_get_subnet)
self.mixin._get_subnet_object = mock.Mock(
side_effect=_get_subnet_object)
def _test_get_changed_ips_for_port(self, expected, original_ips,
new_ips, owner):
@ -136,10 +149,13 @@ class TestIpamBackendMixin(base.BaseTestCase):
original_ips = self._prepare_ips(original)
# mock to test auto address part
pd_subnet = {'subnetpool_id': constants.IPV6_PD_POOL_ID,
'ipv6_address_mode': constants.IPV6_SLAAC,
'ipv6_ra_mode': constants.IPV6_SLAAC}
self.mixin._get_subnet = mock.Mock(return_value=pd_subnet)
pd_subnet_obj = subnet_obj.Subnet(
self.ctx,
id=uuidutils.generate_uuid(),
subnetpool_id=constants.IPV6_PD_POOL_ID,
ipv6_address_mode=constants.IPV6_SLAAC,
ipv6_ra_mode=constants.IPV6_SLAAC)
self.mixin._get_subnet_object = mock.Mock(return_value=pd_subnet_obj)
# make a copy of original_ips
# since it is changed by _get_changed_ips_for_port
@ -252,36 +268,36 @@ class TestIpamBackendMixin(base.BaseTestCase):
def test__is_ip_required_by_subnet_for_router_port(self):
# Owner -> router:
# _get_subnet should not be called,
# _get_subnet_object should not be called,
# expected True
self._mock_slaac_subnet_off()
result = self.mixin._is_ip_required_by_subnet(self.ctx, 'id',
self.owner_router)
self.assertTrue(result)
self.assertFalse(self.mixin._get_subnet.called)
self.assertFalse(self.mixin._get_subnet_object.called)
def test__is_ip_required_by_subnet_for_non_router_port(self):
# Owner -> not router:
# _get_subnet should be called,
# _get_subnet_object should be called,
# expected True, because subnet is not slaac
self._mock_slaac_subnet_off()
result = self.mixin._is_ip_required_by_subnet(self.ctx, 'id',
self.owner_non_router)
self.assertTrue(result)
self.assertTrue(self.mixin._get_subnet.called)
self.assertTrue(self.mixin._get_subnet_object.called)
def test__is_ip_required_by_subnet_for_non_router_port_and_slaac(self):
# Owner -> not router:
# _get_subnet should be called,
# _get_subnet_object should be called,
# expected False, because subnet is slaac
self._mock_slaac_subnet_on()
result = self.mixin._is_ip_required_by_subnet(self.ctx, 'id',
self.owner_non_router)
self.assertFalse(result)
self.assertTrue(self.mixin._get_subnet.called)
self.assertTrue(self.mixin._get_subnet_object.called)
class TestPlugin(db_base_plugin_v2.NeutronDbPluginV2,