Merge "Adding unique constraint for port_id"

neutron_fwaas/db/firewall/v2/firewall_db_v2.py

@@ -125,6 +125,7 @@ class FirewallGroupPortAssociation(model_base.BASEV2):
                                   primary_key=True)
     port_id = sa.Column(sa.String(db_constants.UUID_FIELD_SIZE),
                         sa.ForeignKey('ports.id', ondelete="CASCADE"),
+                        unique=True,
                         primary_key=True)
 
 
@@ -818,12 +819,19 @@ class Firewall_db_mixin_v2(fw_ext.Firewallv2PluginBase, base_db.CommonDbMixin):
         port_id_list = fwg['ports']
         if not port_id_list:
             return
-        with context.session.begin(subtransactions=True):
-            for port_id in port_id_list:
-                fwg_port_db = FirewallGroupPortAssociation(
-                    firewall_group_id=fwg_db['id'],
-                    port_id=port_id)
-                context.session.add(fwg_port_db)
+
+        exc_ports = []
+        for port_id in port_id_list:
+            try:
+                with context.session.begin(subtransactions=True):
+                    fwg_port_db = FirewallGroupPortAssociation(
+                        firewall_group_id=fwg_db['id'],
+                        port_id=port_id)
+                    context.session.add(fwg_port_db)
+            except db_exc.DBDuplicateEntry:
+                exc_ports.append(port_id)
+        if exc_ports:
+            raise f_exc.FirewallGroupPortInUse(port_ids=exc_ports)
 
     def _get_ports_in_firewall_group(self, context, firewall_group_id):
         """Get the Ports associated with the  firewall group."""

neutron_fwaas/db/migration/alembic_migrations/versions/EXPAND_HEAD

@@ -1 +1 @@
-876782258a43
+f24e0d5e5bff

neutron_fwaas/db/migration/alembic_migrations/versions/queens/expand/f24e0d5e5bff_uniq_firewallgroupportassociation0port.py

@@ -0,0 +1,71 @@
+# Copyright 2017 Fujitsu Limited
+#
+#    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.
+#
+
+"""uniq_firewallgroupportassociation0port
+
+Revision ID: f24e0d5e5bff
+Revises: 876782258a43
+Create Date: 2017-11-08 15:55:40.990272
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = 'f24e0d5e5bff'
+down_revision = '876782258a43'
+
+from alembic import op
+from neutron_lib import exceptions
+import sqlalchemy as sa
+
+from neutron._i18n import _
+
+
+fwg_port_association = sa.Table(
+    'firewall_group_port_associations_v2', sa.MetaData(),
+    sa.Column('firewall_group_id', sa.String(36)),
+    sa.Column('port_id', sa.String(36)))
+
+
+class DuplicatePortRecordinFirewallGroupPortAssociation(exceptions.Conflict):
+    message = _("Duplicate port(s) %(port_id)s records exist in"
+                "firewall_group_port_associations_v2 table. Database cannot"
+                "be upgraded. Please remove all duplicated records before"
+                "upgrading the database.")
+
+
+def upgrade():
+    op.create_unique_constraint(
+        'uniq_firewallgroupportassociation0port_id',
+        'firewall_group_port_associations_v2',
+        ['port_id'])
+
+
+def check_sanity(connection):
+    duplicated_port_ids = (
+        get_duplicate_port_records_in_fwg_port_association(connection))
+    if duplicated_port_ids:
+        raise DuplicatePortRecordinFirewallGroupPortAssociation(
+            port_id=",".join(duplicated_port_ids))
+
+
+def get_duplicate_port_records_in_fwg_port_association(connection):
+    insp = sa.engine.reflection.Inspector.from_engine(connection)
+    if 'firewall_group_port_associations_v2' not in insp.get_table_names():
+        return []
+    session = sa.orm.Session(bind=connection.connect())
+    query = (session.query(fwg_port_association.c.port_id)
+             .group_by(fwg_port_association.c.port_id)
+             .having(sa.func.count() > 1)).all()
+    return [q[0] for q in query]

neutron_fwaas/tests/functional/db/test_migrations.py

@@ -10,11 +10,13 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from alembic import script as alembic_script
 from neutron.db.migration.alembic_migrations import external
 from neutron.db.migration import cli as migration
 from neutron.tests.functional.db import test_migrations
 from neutron.tests.unit import testlib_api
 from oslo_config import cfg
+import sqlalchemy
 
 from neutron_fwaas.db.models import head
 
@@ -59,3 +61,39 @@ class TestModelsMigrationsPostgresql(testlib_api.PostgreSQLTestCaseMixin,
                                      _TestModelsMigrationsFWaaS,
                                      testlib_api.SqlTestCaseLight):
     pass
+
+
+class TestSanityCheck(testlib_api.SqlTestCaseLight):
+    BUILD_SCHEMA = False
+
+    def setUp(self):
+        super(TestSanityCheck, self).setUp()
+
+        for conf in migration.get_alembic_configs():
+            self.alembic_config = conf
+            self.alembic_config.neutron_config = cfg.CONF
+
+    def _drop_table(self, table):
+        with self.engine.begin() as conn:
+            table.drop(conn)
+
+    def test_check_sanity_f24e0d5e5bff(self):
+        current_revision = "f24e0d5e5bff"
+        fwg_port_association = sqlalchemy.Table(
+            'firewall_group_port_associations_v2', sqlalchemy.MetaData(),
+            sqlalchemy.Column('firewall_group_id', sqlalchemy.String(36)),
+            sqlalchemy.Column('port_id', sqlalchemy.String(36)))
+
+        with self.engine.connect() as conn:
+            fwg_port_association.create(conn)
+            self.addCleanup(self._drop_table, fwg_port_association)
+            conn.execute(fwg_port_association.insert(), [
+                {'firewall_group_id': '1234', 'port_id': '12345'},
+                {'firewall_group_id': '12343', 'port_id': '12345'}
+            ])
+            script_dir = alembic_script.ScriptDirectory.from_config(
+                self.alembic_config)
+            script = script_dir.get_revision(current_revision).module
+            self.assertRaises(
+                script.DuplicatePortRecordinFirewallGroupPortAssociation,
+                script.check_sanity, conn)