Merge "OVO for Ipam(Allocation,Subnet,Pool)"

neutron/common/constants.py

@@ -142,3 +142,9 @@ PRIVATE_CIDR_RANGE = '169.254.0.0/16'
 DVR_FIP_LL_CIDR = '169.254.64.0/18'
 L3_HA_NET_CIDR = '169.254.192.0/18'
 METADATA_CIDR = '169.254.169.254/32'
+
+# The only defined IpamAllocation status at this stage is 'ALLOCATED'.
+# More states will be available in the future - e.g.: RECYCLABLE
+IPAM_ALLOCATION_STATUS_ALLOCATED = 'ALLOCATED'
+
+VALID_IPAM_ALLOCATION_STATUSES = (IPAM_ALLOCATION_STATUS_ALLOCATED,)

neutron/db/migration/alembic_migrations/versions/newton/contract/3b935b28e7a0_migrate_to_pluggable_ipam.py

@@ -23,6 +23,8 @@ from alembic import op
 from oslo_utils import uuidutils
 import sqlalchemy as sa
 
+from neutron.common import constants as const
+
 # A simple models for tables with only the fields needed for the migration.
 neutron_subnet = sa.Table('subnets', sa.MetaData(),
                           sa.Column('id', sa.String(length=36),
@@ -125,7 +127,7 @@ def upgrade():
     for ip_alloc in session.query(ip_allocation):
         ipam_allocation_values.append(dict(
             ip_address=ip_alloc.ip_address,
-            status='ALLOCATED',
+            status=const.IPAM_ALLOCATION_STATUS_ALLOCATED,
             ipam_subnet_id=map_neutron_id_to_ipam[ip_alloc.subnet_id]))
     op.bulk_insert(ipam_allocation, ipam_allocation_values)
     session.commit()

neutron/ipam/drivers/neutrondb_ipam/db_api.py

@@ -15,7 +15,8 @@
 
 from oslo_utils import uuidutils
 
-from neutron.ipam.drivers.neutrondb_ipam import db_models
+from neutron.common import constants as const
+from neutron.objects import ipam as ipam_objs
 
 # Database operations for Neutron's DB-backed IPAM driver
 
@@ -24,8 +25,10 @@ class IpamSubnetManager(object):
 
     @classmethod
     def load_by_neutron_subnet_id(cls, context, neutron_subnet_id):
-        return context.session.query(db_models.IpamSubnet).filter_by(
-            neutron_subnet_id=neutron_subnet_id).first()
+
+        objs = ipam_objs.IpamSubnet.get_objects(
+                context, neutron_subnet_id=neutron_subnet_id)
+        return objs.pop() if objs else None
 
     def __init__(self, ipam_subnet_id, neutron_subnet_id):
         self._ipam_subnet_id = ipam_subnet_id
@@ -46,10 +49,9 @@ class IpamSubnetManager(object):
         """
         if not self._ipam_subnet_id:
             self._ipam_subnet_id = uuidutils.generate_uuid()
-        ipam_subnet = db_models.IpamSubnet(
-            id=self._ipam_subnet_id,
-            neutron_subnet_id=self._neutron_subnet_id)
-        context.session.add(ipam_subnet)
+        ipam_objs.IpamSubnet(
+            context, id=self._ipam_subnet_id,
+            neutron_subnet_id=self._neutron_subnet_id).create()
         return self._ipam_subnet_id
 
     @classmethod
@@ -62,8 +64,8 @@ class IpamSubnetManager(object):
         :param context: neutron api request context
         :param neutron_subnet_id: neutron subnet id associated with ipam subnet
         """
-        return context.session.query(db_models.IpamSubnet).filter_by(
-            neutron_subnet_id=neutron_subnet_id).delete()
+        return ipam_objs.IpamSubnet.delete_objects(context,
+             neutron_subnet_id=neutron_subnet_id)
 
     def create_pool(self, context, pool_start, pool_end):
         """Create an allocation pool for the subnet.
@@ -75,62 +77,54 @@ class IpamSubnetManager(object):
         :param pool_end: string expressing the end of the pool
         :return: the newly created pool object.
         """
-        ip_pool = db_models.IpamAllocationPool(
-            ipam_subnet_id=self._ipam_subnet_id,
-            first_ip=pool_start,
+        ip_pool_obj = ipam_objs.IpamAllocationPool(
+            context, ipam_subnet_id=self._ipam_subnet_id, first_ip=pool_start,
             last_ip=pool_end)
-        context.session.add(ip_pool)
-        return ip_pool
+        ip_pool_obj.create()
+        return ip_pool_obj
 
     def delete_allocation_pools(self, context):
         """Remove all allocation pools for the current subnet.
 
         :param context: neutron api request context
         """
-        context.session.query(db_models.IpamAllocationPool).filter_by(
-            ipam_subnet_id=self._ipam_subnet_id).delete()
+        ipam_objs.IpamAllocationPool.delete_objects(
+            context, ipam_subnet_id=self._ipam_subnet_id)
 
     def list_pools(self, context):
         """Return pools for the current subnet."""
-        return context.session.query(
-            db_models.IpamAllocationPool).filter_by(
-            ipam_subnet_id=self._ipam_subnet_id)
+        return ipam_objs.IpamAllocationPool.get_objects(
+            context, ipam_subnet_id=self._ipam_subnet_id)
 
     def check_unique_allocation(self, context, ip_address):
         """Validate that the IP address on the subnet is not in use."""
-        iprequest = context.session.query(db_models.IpamAllocation).filter_by(
-            ipam_subnet_id=self._ipam_subnet_id, status='ALLOCATED',
-            ip_address=ip_address).first()
-        if iprequest:
-            return False
-        return True
+        return not ipam_objs.IpamAllocation.objects_exist(
+                     context, ipam_subnet_id=self._ipam_subnet_id,
+                     status=const.IPAM_ALLOCATION_STATUS_ALLOCATED,
+                     ip_address=ip_address)
 
-    def list_allocations(self, context, status='ALLOCATED'):
+    def list_allocations(self, context,
+                         status=const.IPAM_ALLOCATION_STATUS_ALLOCATED):
         """Return current allocations for the subnet.
 
         :param context: neutron api request context
         :param status: IP allocation status
-        :returns: a list of IP allocation as instance of
-            neutron.ipam.drivers.neutrondb_ipam.db_models.IpamAllocation
+        :returns: a list of IpamAllocation OVO objects
         """
-        return context.session.query(
-            db_models.IpamAllocation).filter_by(
-            ipam_subnet_id=self._ipam_subnet_id,
-            status=status)
+        return ipam_objs.IpamAllocation.get_objects(
+            context, ipam_subnet_id=self._ipam_subnet_id, status=status)
 
     def create_allocation(self, context, ip_address,
-                          status='ALLOCATED'):
+                          status=const.IPAM_ALLOCATION_STATUS_ALLOCATED):
         """Create an IP allocation entry.
 
         :param context: neutron api request context
         :param ip_address: the IP address to allocate
         :param status: IP allocation status
         """
-        ip_request = db_models.IpamAllocation(
-            ip_address=ip_address,
-            status=status,
-            ipam_subnet_id=self._ipam_subnet_id)
-        context.session.add(ip_request)
+        ipam_objs.IpamAllocation(
+            context, ip_address=ip_address, status=status,
+            ipam_subnet_id=self._ipam_subnet_id).create()
 
     def delete_allocation(self, context, ip_address):
         """Remove an IP allocation for this subnet.
@@ -138,8 +132,9 @@ class IpamSubnetManager(object):
         :param context: neutron api request context
         :param ip_address: IP address for which the allocation entry should
             be removed.
+        :returns: number of deleted allocation entries.
         """
-        return context.session.query(db_models.IpamAllocation).filter_by(
-            ip_address=ip_address,
-            ipam_subnet_id=self._ipam_subnet_id).delete(
-                synchronize_session=False)
+        return ipam_objs.IpamAllocation.delete_objects(
+            context,
+            ipam_subnet_id=self._ipam_subnet_id,
+            ip_address=ip_address)

neutron/ipam/drivers/neutrondb_ipam/driver.py

@@ -155,7 +155,7 @@ class NeutronDbSubnet(ipam_base.Subnet):
         """Generate an IP address from the set of available addresses."""
         ip_allocations = netaddr.IPSet()
         for ipallocation in self.subnet_manager.list_allocations(context):
-            ip_allocations.add(netaddr.IPAddress(ipallocation.ip_address))
+            ip_allocations.add(ipallocation.ip_address)
 
         for ip_pool in self.subnet_manager.list_pools(context):
             ip_set = netaddr.IPSet()

neutron/objects/common_types.py

@@ -163,6 +163,11 @@ class FlowDirectionEnumField(obj_fields.AutoTypedField):
     AUTO_TYPE = obj_fields.Enum(valid_values=constants.VALID_DIRECTIONS)
 
 
+class IpamAllocationStatusEnumField(obj_fields.AutoTypedField):
+    AUTO_TYPE = obj_fields.Enum(
+        valid_values=constants.VALID_IPAM_ALLOCATION_STATUSES)
+
+
 class EtherTypeEnumField(obj_fields.AutoTypedField):
     AUTO_TYPE = obj_fields.Enum(valid_values=constants.VALID_ETHERTYPES)
 

neutron/objects/ipam.py

@@ -0,0 +1,104 @@
+# Copyright (c) 2016 Intel Corporation.
+# All Rights Reserved.
+#
+#    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.
+
+import netaddr
+from oslo_versionedobjects import base as obj_base
+from oslo_versionedobjects import fields as obj_fields
+
+from neutron.ipam.drivers.neutrondb_ipam import db_models
+from neutron.objects import base
+from neutron.objects import common_types
+
+
+@obj_base.VersionedObjectRegistry.register
+class IpamAllocationPool(base.NeutronDbObject):
+    # Version 1.0: Initial version
+    VERSION = '1.0'
+
+    db_model = db_models.IpamAllocationPool
+
+    foreign_keys = {'IpamSubnet': {'ipam_subnet_id': 'id'}}
+
+    fields = {
+        'id': common_types.UUIDField(),
+        'ipam_subnet_id': common_types.UUIDField(),
+        'first_ip': obj_fields.IPAddressField(),
+        'last_ip': obj_fields.IPAddressField(),
+    }
+
+    @classmethod
+    def modify_fields_from_db(cls, db_obj):
+        result = super(IpamAllocationPool, cls).modify_fields_from_db(db_obj)
+        if 'first_ip' in result:
+            result['first_ip'] = netaddr.IPAddress(result['first_ip'])
+        if 'last_ip' in result:
+            result['last_ip'] = netaddr.IPAddress(result['last_ip'])
+        return result
+
+    @classmethod
+    def modify_fields_to_db(cls, fields):
+        result = super(IpamAllocationPool, cls).modify_fields_to_db(fields)
+        if 'first_ip' in result:
+            result['first_ip'] = cls.filter_to_str(result['first_ip'])
+        if 'last_ip' in result:
+            result['last_ip'] = cls.filter_to_str(result['last_ip'])
+        return result
+
+
+@obj_base.VersionedObjectRegistry.register
+class IpamAllocation(base.NeutronDbObject):
+    # Version 1.0: Initial version
+    VERSION = '1.0'
+
+    db_model = db_models.IpamAllocation
+
+    primary_keys = ['ip_address', 'ipam_subnet_id']
+
+    fields = {
+        'ip_address': obj_fields.IPAddressField(),
+        'status': common_types.IpamAllocationStatusEnumField(nullable=True),
+        'ipam_subnet_id': common_types.UUIDField()
+    }
+
+    @classmethod
+    def modify_fields_from_db(cls, db_obj):
+        result = super(IpamAllocation, cls).modify_fields_from_db(db_obj)
+        if 'ip_address' in result:
+            result['ip_address'] = netaddr.IPAddress(result['ip_address'])
+        return result
+
+    @classmethod
+    def modify_fields_to_db(cls, fields):
+        result = super(IpamAllocation, cls).modify_fields_to_db(fields)
+        if 'ip_address' in result:
+            result['ip_address'] = cls.filter_to_str(result['ip_address'])
+        return result
+
+
+@obj_base.VersionedObjectRegistry.register
+class IpamSubnet(base.NeutronDbObject):
+    # Version 1.0: Initial version
+    VERSION = '1.0'
+
+    db_model = db_models.IpamSubnet
+
+    fields = {
+        'id': common_types.UUIDField(),
+        'neutron_subnet_id': common_types.UUIDField(nullable=True),
+        'allocation_pools': obj_fields.ListOfObjectsField(
+            'IpamAllocationPool')
+    }
+
+    synthetic_fields = ['allocation_pools']

neutron/tests/functional/db/test_ipam.py

@@ -16,6 +16,7 @@
 from neutron_lib import constants
 from neutron_lib import exceptions as n_exc
 from oslo_config import cfg
+from oslo_utils import uuidutils
 import testtools
 
 from neutron import context
@@ -46,10 +47,10 @@ class IpamTestCase(testlib_api.SqlTestCase):
                                    tenant_id=None,
                                    is_admin=True,
                                    overwrite=False)
-        self.tenant_id = 'test_tenant'
-        self.network_id = 'test_net_id'
-        self.subnet_id = 'test_sub_id'
-        self.port_id = 'test_p_id'
+        self.tenant_id = uuidutils.generate_uuid()
+        self.network_id = uuidutils.generate_uuid()
+        self.subnet_id = uuidutils.generate_uuid()
+        self.port_id = uuidutils.generate_uuid()
         self._create_network()
         self._create_subnet()
 
@@ -128,7 +129,7 @@ class IpamTestCase(testlib_api.SqlTestCase):
     def test_allocate_ip_exausted_pool(self):
         # available from .2 up to .6 -> 5
         for i in range(1, 6):
-            self._create_port(self.port_id + str(i))
+            self._create_port(uuidutils.generate_uuid())
 
         ip_alloc_pool_expected = [{'first_ip': '10.10.10.2',
                                    'last_ip': '10.10.10.6',

neutron/tests/tools.py

@@ -291,6 +291,10 @@ def get_random_ether_type():
     return random.choice(n_const.VALID_ETHERTYPES)
 
 
+def get_random_ipam_status():
+    return random.choice(n_const.VALID_IPAM_ALLOCATION_STATUSES)
+
+
 def get_random_ip_protocol():
     return random.choice(list(constants.IP_PROTOCOL_MAP.keys()))
 

neutron/tests/unit/ipam/drivers/neutrondb_ipam/test_db_api.py

@@ -17,15 +17,18 @@ from oslo_utils import uuidutils
 
 from neutron import context
 from neutron.ipam.drivers.neutrondb_ipam import db_api
-from neutron.ipam.drivers.neutrondb_ipam import db_models
+from neutron.objects import ipam as ipam_obj
 from neutron.tests.unit import testlib_api
 
+CORE_PLUGIN = 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2'
+
 
 class TestIpamSubnetManager(testlib_api.SqlTestCase):
     """Test case for SubnetManager DB helper class"""
 
     def setUp(self):
         super(TestIpamSubnetManager, self).setUp()
+        self.setup_coreplugin(core_plugin=CORE_PLUGIN)
         self.ctx = context.get_admin_context()
         self.neutron_subnet_id = uuidutils.generate_uuid()
         self.ipam_subnet_id = uuidutils.generate_uuid()
@@ -39,17 +42,17 @@ class TestIpamSubnetManager(testlib_api.SqlTestCase):
 
     def test_create(self):
         self.assertEqual(self.ipam_subnet_id, self.subnet_manager_id)
-        subnets = self.ctx.session.query(db_models.IpamSubnet).filter_by(
-            id=self.ipam_subnet_id).all()
-        self.assertEqual(1, len(subnets))
+        subnet_count = ipam_obj.IpamSubnet.count(
+            self.ctx, id=self.ipam_subnet_id)
+        self.assertEqual(1, subnet_count)
 
     def test_remove(self):
         count = db_api.IpamSubnetManager.delete(self.ctx,
                                                 self.neutron_subnet_id)
         self.assertEqual(1, count)
-        subnets = self.ctx.session.query(db_models.IpamSubnet).filter_by(
-            id=self.ipam_subnet_id).all()
-        self.assertEqual(0, len(subnets))
+        subnet_exists = ipam_obj.IpamSubnet.objects_exist(
+            self.ctx, id=self.ipam_subnet_id)
+        self.assertFalse(subnet_exists)
 
     def test_remove_non_existent_subnet(self):
         count = db_api.IpamSubnetManager.delete(self.ctx,
@@ -58,16 +61,17 @@ class TestIpamSubnetManager(testlib_api.SqlTestCase):
 
     def _validate_ips(self, pools, db_pool):
         self.assertTrue(
-            any(pool == (db_pool.first_ip, db_pool.last_ip) for pool in pools))
+            any(pool == (str(db_pool.first_ip), str(db_pool.last_ip))
+                for pool in pools))
 
     def test_create_pool(self):
         self.subnet_manager.create_pool(self.ctx,
                                         self.single_pool[0],
                                         self.single_pool[1])
 
-        ipam_pool = self.ctx.session.query(db_models.IpamAllocationPool).\
-            filter_by(ipam_subnet_id=self.ipam_subnet_id).first()
-        self._validate_ips([self.single_pool], ipam_pool)
+        ipam_pools = ipam_obj.IpamAllocationPool.get_objects(
+            self.ctx, ipam_subnet_id=self.ipam_subnet_id)
+        self._validate_ips([self.single_pool], ipam_pools[0])
 
     def test_check_unique_allocation(self):
         self.assertTrue(self.subnet_manager.check_unique_allocation(
@@ -83,18 +87,18 @@ class TestIpamSubnetManager(testlib_api.SqlTestCase):
         ips = ['1.2.3.4', '1.2.3.6', '1.2.3.7']
         for ip in ips:
             self.subnet_manager.create_allocation(self.ctx, ip)
-        allocs = self.subnet_manager.list_allocations(self.ctx).all()
+        allocs = self.subnet_manager.list_allocations(self.ctx)
         self.assertEqual(len(ips), len(allocs))
         for allocation in allocs:
-            self.assertIn(allocation.ip_address, ips)
+            self.assertIn(str(allocation.ip_address), ips)
 
     def _test_create_allocation(self):
         self.subnet_manager.create_allocation(self.ctx,
                                               self.subnet_ip)
-        alloc = self.ctx.session.query(db_models.IpamAllocation).filter_by(
-            ipam_subnet_id=self.ipam_subnet_id).all()
+        alloc = ipam_obj.IpamAllocation.get_objects(
+            self.ctx, ipam_subnet_id=self.ipam_subnet_id)
         self.assertEqual(1, len(alloc))
-        self.assertEqual(self.subnet_ip, alloc[0].ip_address)
+        self.assertEqual(self.subnet_ip, str(alloc[0].ip_address))
         return alloc
 
     def test_create_allocation(self):
@@ -105,6 +109,6 @@ class TestIpamSubnetManager(testlib_api.SqlTestCase):
         self.subnet_manager.delete_allocation(self.ctx,
                                               allocs[0].ip_address)
 
-        allocs = self.ctx.session.query(db_models.IpamAllocation).filter_by(
-            ipam_subnet_id=self.ipam_subnet_id).all()
-        self.assertEqual(0, len(allocs))
+        alloc_exists = ipam_obj.IpamAllocation.objects_exist(
+            self.ctx, ipam_subnet_id=self.ipam_subnet_id)
+        self.assertFalse(alloc_exists)

neutron/tests/unit/ipam/drivers/neutrondb_ipam/test_driver.py

@@ -18,13 +18,14 @@ import netaddr
 from neutron_lib import constants
 from neutron_lib import exceptions as n_exc
 from neutron_lib.plugins import directory
+from oslo_utils import uuidutils
 
 from neutron.common import constants as n_const
 from neutron import context
-from neutron.ipam.drivers.neutrondb_ipam import db_models
 from neutron.ipam.drivers.neutrondb_ipam import driver
 from neutron.ipam import exceptions as ipam_exc
 from neutron.ipam import requests as ipam_req
+from neutron.objects import ipam as ipam_obj
 from neutron.tests.unit.db import test_db_base_plugin_v2 as test_db_plugin
 from neutron.tests.unit import testlib_api
 
@@ -206,9 +207,10 @@ class TestNeutronDbIpamPool(testlib_api.SqlTestCase,
 
     def test_get_details_for_invalid_subnet_id_fails(self):
         cidr = '10.0.0.0/24'
+        non_existent_id = uuidutils.generate_uuid()
         subnet_req = ipam_req.SpecificSubnetRequest(
             self._tenant_id,
-            'non-existent-id',
+            non_existent_id,
             cidr)
         self.ipam_pool.allocate_subnet(subnet_req)
         # Neutron subnet does not exist, so get_subnet should fail
@@ -386,8 +388,10 @@ class TestNeutronDbIpamSubnet(testlib_api.SqlTestCase,
         # This test should pass because ipam subnet is no longer
         # have foreign key relationship with neutron subnet.
         # Creating ipam subnet before neutron subnet is a valid case.
+        tenant_id = uuidutils.generate_uuid()
+        subnet_id = uuidutils.generate_uuid()
         subnet_req = ipam_req.SpecificSubnetRequest(
-            'tenant_id', 'meh', '192.168.0.0/24')
+            tenant_id, subnet_id, '192.168.0.0/24')
         self.ipam_pool.allocate_subnet(subnet_req)
 
     def test_update_allocation_pools_with_no_pool_change(self):
@@ -405,14 +409,16 @@ class TestNeutronDbIpamSubnet(testlib_api.SqlTestCase,
         self.assertFalse(ipam_subnet.create_allocation_pools.called)
 
     def _test__no_pool_changes(self, new_pools):
-        id = 'some-id'
+        id = uuidutils.generate_uuid()
         ipam_subnet = driver.NeutronDbSubnet(id, self.ctx)
-        pools = [db_models.IpamAllocationPool(ipam_subnet_id=id,
-                                              first_ip='192.168.10.20',
-                                              last_ip='192.168.10.41'),
-                 db_models.IpamAllocationPool(ipam_subnet_id=id,
-                                              first_ip='192.168.10.50',
-                                              last_ip='192.168.10.60')]
+        pools = [ipam_obj.IpamAllocationPool(self.ctx,
+                                             ipam_subnet_id=id,
+                                             first_ip='192.168.10.20',
+                                             last_ip='192.168.10.41'),
+                 ipam_obj.IpamAllocationPool(self.ctx,
+                                             ipam_subnet_id=id,
+                                             first_ip='192.168.10.50',
+                                             last_ip='192.168.10.60')]
 
         ipam_subnet.subnet_manager.list_pools = mock.Mock(return_value=pools)
         return ipam_subnet._no_pool_changes(self.ctx, new_pools)

neutron/tests/unit/objects/test_base.py

@@ -431,6 +431,7 @@ FIELD_TYPE_VALUE_GENERATOR_MAP = {
     common_types.DscpMarkField: get_random_dscp_mark,
     common_types.EtherTypeEnumField: tools.get_random_ether_type,
     common_types.FlowDirectionEnumField: tools.get_random_flow_direction,
+    common_types.IpamAllocationStatusEnumField: tools.get_random_ipam_status,
     common_types.IPNetworkField: tools.get_random_ip_network,
     common_types.IPNetworkPrefixLenField: tools.get_random_prefixlen,
     common_types.IPV6ModeEnumField: tools.get_random_ipv6_mode,

neutron/tests/unit/objects/test_ipam.py

@@ -0,0 +1,74 @@
+#    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.
+
+from neutron.objects import ipam
+from neutron.tests.unit.objects import test_base as obj_test_base
+from neutron.tests.unit import testlib_api
+
+
+class IpamSubnetObjectIfaceTestCase(obj_test_base.BaseObjectIfaceTestCase):
+
+    _test_class = ipam.IpamSubnet
+
+
+class IpamSubnetDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
+                                 testlib_api.SqlTestCase):
+
+    _test_class = ipam.IpamSubnet
+
+    def setUp(self):
+        super(IpamSubnetDbObjectTestCase, self).setUp()
+        self._create_test_network()
+        self._create_test_subnet(self._network)
+        self.update_obj_fields({'neutron_subnet_id': self._subnet['id']})
+
+
+class IpamAllocationPoolObjectIfaceTestCase(
+    obj_test_base.BaseObjectIfaceTestCase):
+
+    _test_class = ipam.IpamAllocationPool
+
+
+class IpamAllocationPoolDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
+                                         testlib_api.SqlTestCase):
+    _test_class = ipam.IpamAllocationPool
+
+    def setUp(self):
+        super(IpamAllocationPoolDbObjectTestCase, self).setUp()
+        self._create_test_ipam_subnet()
+        self.update_obj_fields({'ipam_subnet_id': self._ipam_subnet['id']})
+
+    def _create_test_ipam_subnet(self):
+        attrs = self.get_random_object_fields(obj_cls=ipam.IpamSubnet)
+        self._ipam_subnet = ipam.IpamSubnet(self.context, **attrs)
+        self._ipam_subnet.create()
+
+
+class IpamAllocationObjectIfaceTestCase(obj_test_base.BaseObjectIfaceTestCase):
+
+    _test_class = ipam.IpamAllocation
+
+
+class IpamAllocationDbObjectTestCase(obj_test_base.BaseDbObjectTestCase,
+                                    testlib_api.SqlTestCase):
+
+    _test_class = ipam.IpamAllocation
+
+    def setUp(self):
+        super(IpamAllocationDbObjectTestCase, self).setUp()
+        self._create_test_ipam_subnet()
+        self.update_obj_fields({'ipam_subnet_id': self._ipam_subnet['id']})
+
+    def _create_test_ipam_subnet(self):
+        attrs = self.get_random_object_fields(obj_cls=ipam.IpamSubnet)
+        self._ipam_subnet = ipam.IpamSubnet(self.context, **attrs)
+        self._ipam_subnet.create()

neutron/tests/unit/objects/test_objects.py

@@ -45,6 +45,9 @@ object_data = {
     'GreEndpoint': '1.0-040f026996b5952e2ae4ccd40ac61ca6',
     'IPAllocation': '1.0-47251b4c6d45c3b5feb0297fe5c461f2',
     'IPAllocationPool': '1.0-371016a6480ed0b4299319cb46d9215d',
+    'IpamAllocation': '1.0-ace65431abd0a7be84cc4a5f32d034a3',
+    'IpamAllocationPool': '1.0-c4fa1460ed1b176022ede7af7d1510d5',
+    'IpamSubnet': '1.0-713de401682a70f34891e13af645fa08',
     'Network': '1.0-f2f6308f79731a767b92b26b0f4f3849',
     'NetworkDNSDomain': '1.0-420db7910294608534c1e2e30d6d8319',
     'NetworkPortSecurity': '1.0-b30802391a87945ee9c07582b4ff95e3',