New alembic migration to support namespaces in postgresql

We are now using drop_index to delete old constraint (name + project_id)
in namespace support migration. It works only for MySQL, for PostgreSQL
we should use drop_constraint to delete old constraint.
drop_index does not affect PostgreSQL and drop_constraint does not affect
MySQL, so we should use both for namespace support for both databases.
Added old constraint (name + project_id) deletion for
PostgreSQL with new migration.

Change-Id: I137162dd2b643b24f02799c6a13e3b0c0a8d3fb5
Closes-bug: #1838635

mistral/db/sqlalchemy/migration/alembic_migrations/versions/035_namespace_support_postgresql.py

@@ -0,0 +1,43 @@
+# Copyright 2019 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.
+
+"""Namespace support postgresql
+
+Revision ID: 035
+Revises: 034
+Create Date: 2019-08-01 15:48:34.115639
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '035'
+down_revision = '034'
+
+from alembic import op
+from sqlalchemy.engine import reflection
+
+
+def upgrade():
+
+    inspect = reflection.Inspector.from_engine(op.get_bind())
+
+    unique_constraints = [
+        unique_constraint['name'] for unique_constraint in
+        inspect.get_unique_constraints('workflow_definitions_v2')
+    ]
+
+    if 'workflow_definitions_v2_name_project_id_key' in unique_constraints:
+        op.drop_constraint('workflow_definitions_v2_name_project_id_key',
+                           table_name='workflow_definitions_v2')

mistral/tests/unit/db/v2/test_sqlalchemy_db_api.py

@@ -482,6 +482,28 @@ WF_DEFINITIONS = [
         'created_at': datetime.datetime(2016, 12, 1, 15, 1, 0),
         'namespace': 'mynamespace'
     },
+    {
+        'name': 'my_wf1_with_namespace',
+        'definition': 'empty',
+        'spec': {},
+        'tags': ['mc'],
+        'scope': 'public',
+        'project_id': '1233',
+        'trust_id': '1234',
+        'created_at': datetime.datetime(2016, 12, 1, 15, 0, 0),
+        'namespace': 'abc'
+    },
+    {
+        'name': 'my_wf1_with_namespace',
+        'definition': 'empty',
+        'spec': {},
+        'tags': ['mc'],
+        'scope': 'public',
+        'project_id': '1233',
+        'trust_id': '1234',
+        'created_at': datetime.datetime(2016, 12, 1, 15, 0, 0),
+        'namespace': 'def'
+    },
 ]
 
 
@@ -703,6 +725,38 @@ class WorkflowDefinitionTest(SQLAlchemyTest):
             WF_DEFINITIONS[2]
         )
 
+    def test_create_same_workflow_definition_in_different_namespace(self):
+        name = WF_DEFINITIONS[3]['name']
+        namespace1 = WF_DEFINITIONS[3]['namespace']
+        namespace2 = WF_DEFINITIONS[4]['namespace']
+
+        self.assertIsNone(db_api.load_workflow_definition(name, namespace1))
+        self.assertIsNone(db_api.load_workflow_definition(name, namespace2))
+
+        created1 = db_api.create_workflow_definition(
+            WF_DEFINITIONS[3]
+        )
+
+        created2 = db_api.create_workflow_definition(
+            WF_DEFINITIONS[4]
+        )
+
+        self.assertIsNotNone(created1)
+        self.assertIsNotNone(created2)
+        self.assertIsNotNone(created1.name)
+        self.assertIsNotNone(created2.name)
+        self.assertIsNotNone(created1.namespace)
+        self.assertIsNotNone(created2.namespace)
+
+        fetched1 = db_api.get_workflow_definition(created1.name,
+                                                  created1.namespace)
+
+        fetched2 = db_api.get_workflow_definition(created2.name,
+                                                  created2.namespace)
+
+        self.assertEqual(created1, fetched1)
+        self.assertEqual(created2, fetched2)
+
     def test_update_workflow_definition(self):
         created = db_api.create_workflow_definition(WF_DEFINITIONS[0])