<filename>tests/backends/aiida_sqlalchemy/migrations/test_schemas.py
# -*- coding: utf-8 -*-
###########################################################################
# Copyright (c), The AiiDA team. All rights reserved.                     #
# This file is part of the AiiDA code.                                    #
#                                                                         #
# The code is hosted on GitHub at https://github.com/aiidateam/aiida-core #
# For further information on the license, see the LICENSE.txt file        #
# For further information please visit http://www.aiida.net               #
###########################################################################
"""Check that the schema that results from a migration is the same generated by the models.

This is important since migrations are frequently written by hand or extended manually,
and we have to ensure that the final result is what is conceived in the SQLA models.
"""
import os

from alembic import command
from alembic.config import Config
import pytest
from sqlalchemy.engine import create_engine
from sqlalchemydiff import compare

from aiida.backends.sqlalchemy import manager
from aiida.backends.sqlalchemy.models.base import Base

from ..test_utils import destroy_database, new_database


class TestMigrationSchemaVsModelsSchema:
    """Class to perform test."""

    @pytest.fixture(autouse=True)
    def init_db(self, clear_database_before_test, backend):  # pylint: disable=unused-argument
        """Initialise the databases"""
        # pylint: disable=attribute-defined-outside-init
        from sqlalchemydiff.util import get_temporary_uri

        from aiida.backends.sqlalchemy.migrations import versions

        # The path to the folder that contains the migration configuration (the
        # actual configuration - not the testing)
        migr_method_dir_path = os.path.dirname(os.path.realpath(manager.__file__))
        # Set the alembic script directory location
        alembic_dpath = os.path.join(migr_method_dir_path, manager.ALEMBIC_REL_PATH)  # pylint: disable=no-member

        # Constructing the versions directory
        versions_dpath = os.path.join(os.path.dirname(versions.__file__))

        # Setting dynamically the the path to the alembic configuration
        # (this is where the env.py file can be found)
        self.alembic_cfg_left = Config()
        self.alembic_cfg_left.set_main_option('script_location', alembic_dpath)
        # Setting dynamically the versions directory. These are the
        # migration scripts to pass from one version to the other. The
        # default ones are overridden with test-specific migrations.
        self.alembic_cfg_left.set_main_option('version_locations', versions_dpath)

        # The correction URL to the SQLA database of the current
        # AiiDA connection
        curr_db_url = backend.get_session().bind.url

        # Create new urls for the two new databases
        self.db_url_left = get_temporary_uri(str(curr_db_url))
        self.db_url_right = get_temporary_uri(str(curr_db_url))

        # Put the correct database url to the database used by alembic
        self.alembic_cfg_left.set_main_option('sqlalchemy.url', self.db_url_left)

        # Database creation
        new_database(self.db_url_left)
        new_database(self.db_url_right)

        yield

        destroy_database(self.db_url_left)
        destroy_database(self.db_url_right)

    def test_model_and_migration_schemas_are_the_same(self):  # pylint: disable=invalid-name
        """Compare two databases.

        Compares the database obtained with all migrations against the
        one we get out of the models.  It produces a text file with the
        results to help debug differences.
        """
        with create_engine(self.db_url_left).begin() as connection:
            self.alembic_cfg_left.attributes['connection'] = connection  # pylint: disable=unsupported-assignment-operation
            command.upgrade(self.alembic_cfg_left, 'head')

        engine_right = create_engine(self.db_url_right)
        Base.metadata.create_all(engine_right)
        engine_right.dispose()

        result = compare(self.db_url_left, self.db_url_right, set(['alembic_version']))

        assert result.is_match, (
            "The migration database doesn't match to the one "  # pylint: disable=protected-access
            f'created by the models.\nDifferences: {result._dump_data(result.errors)}'
        )
