Add a blocker migration for missing consumer records

There are several TODO throughout the code about ensuring
consumer records calls that can be removed once there is
blocker migration that will fail if there are missing
consumer records.

This patch adds that blocker migration. Subsequent patches
will remove the redundant code.

Change-Id: I6029c5095ed1e6ff7c46d480454db1382073bd57

placement/db/sqlalchemy/alembic/versions/b5c396305c25_block_on_null_consumer.py

@@ -0,0 +1,50 @@
+#    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.
+
+"""Block on null consumer
+
+Revision ID: b5c396305c25
+Revises: 611cd6dffd7b
+Create Date: 2019-06-11 16:30:04.114287
+
+"""
+from alembic import context
+import sqlalchemy as sa
+from sqlalchemy import func as sqlfunc
+
+
+# revision identifiers, used by Alembic.
+revision = 'b5c396305c25'
+down_revision = '611cd6dffd7b'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    connection = context.get_bind()
+    metadata = sa.MetaData(bind=connection)
+    consumers = sa.Table('consumers', metadata, autoload=True)
+    allocations = sa.Table('allocations', metadata, autoload=True)
+    alloc_to_consumer = sa.outerjoin(
+        allocations, consumers,
+        allocations.c.consumer_id == consumers.c.uuid)
+    cols = [
+        sqlfunc.count(),
+    ]
+    sel = sa.select(cols)
+    sel = sel.select_from(alloc_to_consumer)
+    sel = sel.where(consumers.c.id.is_(None))
+
+    if connection.scalar(sel):
+        raise Exception('There is at least one allocation record which is '
+                        'missing a consumer record. Run the "placement-manage '
+                        'db online_data_migrations" command.')

placement/tests/functional/db/test_migrations.py

@@ -193,6 +193,39 @@ class MigrationCheckersMixin(object):
         # Re-run the upgrade and it should be OK.
         self.migration_api.upgrade('611cd6dffd7b')
 
+    def test_block_on_missing_consumer(self):
+        """Upgrades the schema to b4ed3a175331 (initial), injects an allocation
+        without a corresponding consumer record and then tries to upgrade to
+        head which should fail on the b5c396305c25 blocker migration.
+        """
+        # Upgrade to populate the schema.
+        self.migration_api.upgrade('b4ed3a175331')
+        # Now insert a resource provider to build off
+        rps = db_utils.get_table(self.engine, 'resource_providers')
+        rp_id = rps.insert(values={
+            'name': 'fake-rp-name', 'uuid': uuids.rp_uuid,
+            'root_provider_id': 1
+        }).execute().inserted_primary_key[0]
+        # Now insert an allocation
+        allocations = db_utils.get_table(self.engine, 'allocations')
+        allocations.insert(values={
+            'resource_provider_id': rp_id, 'resource_class_id': 1,
+            'used': 5, 'consumer_id': uuids.consumer1
+        }).execute().inserted_primary_key[0]
+        # Now run the blocker migration and it should raise an error.
+        ex = self.assertRaises(  # noqa H202
+            Exception, self.migration_api.upgrade, 'b5c396305c25')
+        # Make sure it's the error we expect.
+        self.assertIn('There is at least one allocation record which is '
+                      'missing a consumer record.',
+                      six.text_type(ex))
+        # Add a (faked) consumer record and try again
+        consumers = db_utils.get_table(self.engine, 'consumers')
+        consumers.insert(values={
+            'uuid': uuids.consumer1, 'project_id': 1, 'user_id': 1
+        }).execute().inserted_primary_key[0]
+        self.migration_api.upgrade('b5c396305c25')
+
 
 class PlacementOpportunisticFixture(object):
     def get_enginefacade(self):