Functional test IPAM DB operation

Add a functional test to check that the IPAM logic is correct and
that the data stored in the DB are sane.

Partial-bug: #1331564
Change-Id: I7776a50824184663abd37e210d08a63eb5519cea

diff --git a/neutron/tests/functional/db/test_ipam.py b/neutron/tests/functional/db/test_ipam.py
new file mode 100644 (file)
index 0000000..95bb927
--- /dev/null
+++ b/neutron/tests/functional/db/test_ipam.py
@@ -0,0 +1,222 @@
+# Copyright 2015 SUSE Linux Products GmbH
+# 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.
+
+from oslo.config import cfg
+from oslo.db.sqlalchemy import session
+from oslo.db.sqlalchemy import test_base
+import testtools
+
+from neutron.api.v2 import attributes
+from neutron.common import constants
+from neutron.common import exceptions as n_exc
+from neutron import context
+from neutron.db import db_base_plugin_v2 as base_plugin
+from neutron.db import model_base
+from neutron.db import models_v2
+from neutron.tests import base
+
+
+def get_admin_test_context(db_url):
+    """
+    get_admin_test_context is used to provide a test context. A new session is
+    created using the db url specified
+    """
+    ctx = context.Context(user_id=None,
+                          tenant_id=None,
+                          is_admin=True,
+                          read_deleted="no",
+                          load_admin_roles=True,
+                          overwrite=False)
+    facade = session.EngineFacade(db_url, mysql_sql_mode='STRICT_ALL_TABLES')
+    ctx._session = facade.get_session(autocommit=False, expire_on_commit=True)
+    return ctx
+
+
+class IpamTestCase(object):
+    """
+    Base class for tests that aim to test ip allocation.
+    """
+
+    def configure_test(self):
+        model_base.BASEV2.metadata.create_all(self.engine)
+        cfg.CONF.set_override('notify_nova_on_port_status_changes', False)
+        self.plugin = base_plugin.NeutronDbPluginV2()
+        self.cxt = get_admin_test_context(self.engine.url)
+        self.addCleanup(self.cxt._session.close)
+        self.tenant_id = 'test_tenant'
+        self.network_id = 'test_net_id'
+        self.subnet_id = 'test_sub_id'
+        self.port_id = 'test_p_id'
+        self._create_network()
+        self._create_subnet()
+
+    def result_set_to_dicts(self, resultset, keys):
+        dicts = []
+        for item in resultset:
+            item_dict = dict((x, item[x]) for x in keys)
+            dicts.append(item_dict)
+        return dicts
+
+    def assert_ip_alloc_matches(self, expected):
+        result_set = self.cxt.session.query(models_v2.IPAllocation).all()
+        keys = ['port_id', 'ip_address', 'subnet_id', 'network_id']
+        actual = self.result_set_to_dicts(result_set, keys)
+        self.assertEqual(expected, actual)
+
+    def assert_ip_avail_range_matches(self, expected):
+        result_set = self.cxt.session.query(
+            models_v2.IPAvailabilityRange).all()
+        keys = ['first_ip', 'last_ip']
+        actual = self.result_set_to_dicts(result_set, keys)
+        self.assertEqual(expected, actual)
+
+    def assert_ip_alloc_pool_matches(self, expected):
+        result_set = self.cxt.session.query(models_v2.IPAllocationPool).all()
+        keys = ['first_ip', 'last_ip', 'subnet_id']
+        actual = self.result_set_to_dicts(result_set, keys)
+        self.assertEqual(expected, actual)
+
+    def _create_network(self):
+        network = {'tenant_id': self.tenant_id,
+                   'id': self.network_id,
+                   'name': 'test-net',
+                   'admin_state_up': True,
+                   'shared': False,
+                   'status': constants.NET_STATUS_ACTIVE}
+        return self.plugin.create_network(self.cxt, {'network': network})
+
+    def _create_subnet(self):
+        subnet = {'tenant_id': self.tenant_id,
+                  'id': self.subnet_id,
+                  'name': 'test_sub',
+                  'network_id': self.network_id,
+                  'ip_version': 4,
+                  'cidr': '10.10.10.0/29',
+                  'enable_dhcp': False,
+                  'gateway_ip': '10.10.10.1',
+                  'shared': False,
+                  'allocation_pools': attributes.ATTR_NOT_SPECIFIED,
+                  'dns_nameservers': attributes.ATTR_NOT_SPECIFIED,
+                  'host_routes': attributes.ATTR_NOT_SPECIFIED}
+        return self.plugin.create_subnet(self.cxt, {'subnet': subnet})
+
+    def _create_port(self, port_id, fixed_ips=None):
+        port_fixed_ips = (fixed_ips if fixed_ips else
+                          attributes.ATTR_NOT_SPECIFIED)
+        port = {'tenant_id': self.tenant_id,
+                'name': 'test_port',
+                'id': port_id,
+                'network_id': self.network_id,
+                'mac_address': attributes.ATTR_NOT_SPECIFIED,
+                'admin_state_up': True,
+                'status': constants.PORT_STATUS_ACTIVE,
+                'device_id': 'test_dev_id',
+                'device_owner': 'compute',
+                'fixed_ips': port_fixed_ips}
+        self.plugin.create_port(self.cxt, {'port': port})
+
+    def test_allocate_fixed_ip(self):
+        fixed_ip = [{'ip_address': "10.10.10.3", 'subnet_id': self.subnet_id}]
+        self._create_port(self.port_id, fixed_ip)
+
+        ip_alloc_expected = [{'port_id': self.port_id,
+                              'ip_address': fixed_ip[0].get('ip_address'),
+                              'subnet_id': self.subnet_id,
+                              'network_id': self.network_id}]
+        ip_avail_ranges_expected = [{'first_ip': '10.10.10.2',
+                                     'last_ip': '10.10.10.2'},
+                                    {'first_ip': '10.10.10.4',
+                                     'last_ip': '10.10.10.6'}]
+        ip_alloc_pool_expected = [{'first_ip': '10.10.10.2',
+                                   'last_ip': '10.10.10.6',
+                                   'subnet_id': self.subnet_id}]
+        self.assert_ip_alloc_matches(ip_alloc_expected)
+        self.assert_ip_alloc_pool_matches(ip_alloc_pool_expected)
+        self.assert_ip_avail_range_matches(
+            ip_avail_ranges_expected)
+
+    def test_allocate_first_available_ip(self):
+        self._create_port(self.port_id)
+        ip_alloc_expected = [{'port_id': self.port_id,
+                              'ip_address': '10.10.10.2',
+                              'subnet_id': self.subnet_id,
+                              'network_id': self.network_id}]
+        ip_avail_ranges_expected = [{'first_ip': '10.10.10.3',
+                                     'last_ip': '10.10.10.6'}]
+        ip_alloc_pool_expected = [{'first_ip': '10.10.10.2',
+                                   'last_ip': '10.10.10.6',
+                                   'subnet_id': self.subnet_id}]
+        self.assert_ip_alloc_matches(ip_alloc_expected)
+        self.assert_ip_alloc_pool_matches(ip_alloc_pool_expected)
+        self.assert_ip_avail_range_matches(
+            ip_avail_ranges_expected)
+
+    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))
+
+        ip_avail_ranges_expected = []
+        ip_alloc_pool_expected = [{'first_ip': '10.10.10.2',
+                                   'last_ip': '10.10.10.6',
+                                   'subnet_id': self.subnet_id}]
+        self.assert_ip_alloc_pool_matches(ip_alloc_pool_expected)
+        self.assert_ip_avail_range_matches(
+            ip_avail_ranges_expected)
+        # Create another port
+        with testtools.ExpectedException(n_exc.IpAddressGenerationFailure):
+            self._create_port(self.port_id)
+
+    def test_rebuild_availability_range(self):
+        for i in range(1, 6):
+            self._create_port(self.port_id + str(i))
+
+        ip_avail_ranges_expected = []
+        ip_alloc_pool_expected = [{'first_ip': '10.10.10.2',
+                                   'last_ip': '10.10.10.6',
+                                   'subnet_id': self.subnet_id}]
+        self.assert_ip_alloc_pool_matches(ip_alloc_pool_expected)
+        self.assert_ip_avail_range_matches(
+            ip_avail_ranges_expected)
+        # Delete some ports, this will free the first two IPs
+        for i in range(1, 3):
+            self.plugin.delete_port(self.cxt, self.port_id + str(i))
+        # Create another port, this will trigger the rebuilding of the
+        # availability ranges
+        self._create_port(self.port_id)
+        ip_avail_ranges_expected = [{'first_ip': '10.10.10.3',
+                                     'last_ip': '10.10.10.3'}]
+
+        ip_alloc = self.cxt.session.query(models_v2.IPAllocation).all()
+        self.assertEqual(4, len(ip_alloc))
+        self.assert_ip_alloc_pool_matches(ip_alloc_pool_expected)
+        self.assert_ip_avail_range_matches(
+            ip_avail_ranges_expected)
+
+
+class TestIpamMySql(test_base.MySQLOpportunisticTestCase, base.BaseTestCase,
+                    IpamTestCase):
+
+    def setUp(self):
+        super(TestIpamMySql, self).setUp()
+        self.configure_test()
+
+
+class TestIpamPsql(test_base.PostgreSQLOpportunisticTestCase,
+                   base.BaseTestCase, IpamTestCase):
+
+    def setUp(self):
+        super(TestIpamPsql, self).setUp()
+        self.configure_test()