text: # Copyright 2012 Red Hat, Inc
  # 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.
  
  """Functional test cases for glance-manage"""
  
  import os
  import sys
  
  from oslo_config import cfg
  from oslo_db import options as db_options
  
  from glance.common import utils
  from glance.db import migration as db_migration
  from glance.db.sqlalchemy import alembic_migrations
  from glance.db.sqlalchemy.alembic_migrations import data_migrations
  from glance.db.sqlalchemy import api as db_api
  from glance.tests import functional
  from glance.tests.utils import depends_on_exe
  from glance.tests.utils import execute
  from glance.tests.utils import skip_if_disabled
  
  CONF = cfg.CONF
  
  
  class TestGlanceManage(functional.FunctionalTest):
      """Functional tests for glance-manage"""
  
      def setUp(self):
          super(TestGlanceManage, self).setUp()
          conf_dir = os.path.join(self.test_dir, 'etc')
          utils.safe_mkdirs(conf_dir)
          self.conf_filepath = os.path.join(conf_dir, 'glance-manage.conf')
          self.db_filepath = os.path.join(self.test_dir, 'tests.sqlite')
          self.connection = ('sql_connection = sqlite:///%s' %
                             self.db_filepath)
          db_options.set_defaults(CONF, connection='sqlite:///%s' %
                                                   self.db_filepath)
  
      def _db_command(self, db_method):
          with open(self.conf_filepath, 'w') as conf_file:
              conf_file.write('[DEFAULT]\n')
              conf_file.write(self.connection)
              conf_file.flush()
  
          cmd = ('%s -m glance.cmd.manage --config-file %s db %s' %
                 (sys.executable, self.conf_filepath, db_method))
          return execute(cmd, raise_error=True)
  
      def _check_db(self, expected_exitcode):
          with open(self.conf_filepath, 'w') as conf_file:
              conf_file.write('[DEFAULT]\n')
              conf_file.write(self.connection)
              conf_file.flush()
  
          cmd = ('%s -m glance.cmd.manage --config-file %s db check' %
                 (sys.executable, self.conf_filepath))
          exitcode, out, err = execute(cmd, raise_error=True,
                                       expected_exitcode=expected_exitcode)
          return exitcode, out
  
      def _assert_table_exists(self, db_table):
          cmd = ("sqlite3 {0} \"SELECT name FROM sqlite_master WHERE "
                 "type='table' AND name='{1}'\"").format(self.db_filepath,
                                                         db_table)
          exitcode, out, err = execute(cmd, raise_error=True)
          msg = "Expected table {0} was not found in the schema".format(db_table)
          self.assertEqual(out.rstrip().decode("utf-8"), db_table, msg)
  
      @depends_on_exe('sqlite3')
      @skip_if_disabled
      def test_db_creation(self):
          """Test schema creation by db_sync on a fresh DB"""
          self._db_command(db_method='sync')
  
          for table in ['images', 'image_tags', 'image_locations',
                        'image_members', 'image_properties']:
              self._assert_table_exists(table)
  
      @depends_on_exe('sqlite3')
      @skip_if_disabled
      def test_sync(self):
          """Test DB sync which internally calls EMC"""
          self._db_command(db_method='sync')
          contract_head = alembic_migrations.get_alembic_branch_head(
              db_migration.CONTRACT_BRANCH)
  
          cmd = ("sqlite3 {0} \"SELECT version_num FROM alembic_version\""
                 ).format(self.db_filepath)
          exitcode, out, err = execute(cmd, raise_error=True)
          self.assertEqual(contract_head, out.rstrip().decode("utf-8"))
  
      @depends_on_exe('sqlite3')
      @skip_if_disabled
      def test_check(self):
          exitcode, out = self._check_db(3)
          self.assertEqual(3, exitcode)
  
          self._db_command(db_method='expand')
          if data_migrations.has_pending_migrations(db_api.get_engine()):
              exitcode, out = self._check_db(4)
              self.assertEqual(4, exitcode)
  
          self._db_command(db_method='migrate')
          exitcode, out = self._check_db(5)
          self.assertEqual(5, exitcode)
  
          self._db_command(db_method='contract')
          exitcode, out = self._check_db(0)
          self.assertEqual(0, exitcode)
  
      @depends_on_exe('sqlite3')
      @skip_if_disabled
      def test_expand(self):
          """Test DB expand"""
          self._db_command(db_method='expand')
          expand_head = alembic_migrations.get_alembic_branch_head(
              db_migration.EXPAND_BRANCH)
  
          cmd = ("sqlite3 {0} \"SELECT version_num FROM alembic_version\""
                 ).format(self.db_filepath)
          exitcode, out, err = execute(cmd, raise_error=True)
          self.assertEqual(expand_head, out.rstrip().decode("utf-8"))
          exitcode, out, err = self._db_command(db_method='expand')
          self.assertIn('Database expansion is up to date. '
                        'No expansion needed.', str(out))
  
      @depends_on_exe('sqlite3')
      @skip_if_disabled
      def test_migrate(self):
          """Test DB migrate"""
          self._db_command(db_method='expand')
          if data_migrations.has_pending_migrations(db_api.get_engine()):
              self._db_command(db_method='migrate')
          expand_head = alembic_migrations.get_alembic_branch_head(
              db_migration.EXPAND_BRANCH)
  
          cmd = ("sqlite3 {0} \"SELECT version_num FROM alembic_version\""
                 ).format(self.db_filepath)
          exitcode, out, err = execute(cmd, raise_error=True)
          self.assertEqual(expand_head, out.rstrip().decode("utf-8"))
          self.assertEqual(False, data_migrations.has_pending_migrations(
              db_api.get_engine()))
          if data_migrations.has_pending_migrations(db_api.get_engine()):
              exitcode, out, err = self._db_command(db_method='migrate')
              self.assertIn('Database migration is up to date. No migration '
                            'needed.', str(out))
  
      @depends_on_exe('sqlite3')
      @skip_if_disabled
      def test_contract(self):
          """Test DB contract"""
          self._db_command(db_method='expand')
          if data_migrations.has_pending_migrations(db_api.get_engine()):
              self._db_command(db_method='migrate')
          self._db_command(db_method='contract')
          contract_head = alembic_migrations.get_alembic_branch_head(
              db_migration.CONTRACT_BRANCH)
  
          cmd = ("sqlite3 {0} \"SELECT version_num FROM alembic_version\""
                 ).format(self.db_filepath)
          exitcode, out, err = execute(cmd, raise_error=True)
          self.assertEqual(contract_head, out.rstrip().decode("utf-8"))
          exitcode, out, err = self._db_command(db_method='contract')
          self.assertIn('Database is up to date. No migrations needed.',
                        str(out))
  