from django.test.runner import DiscoverRunner
from django.db import connections
from django.test import utils


class UnManagedModelTestRunner(DiscoverRunner):
    def setup_databases(self, *args, **kwargs):
        return setup_databases(
            self.verbosity,
            self.interactive,
            time_keeper=self.time_keeper,
            keepdb=self.keepdb,
            debug_sql=self.debug_sql,
            parallel=self.parallel,
            **kwargs,
        )


def get_unmanaged_models():
    from django.apps import apps

    get_models = apps.get_models
    return [m for m in get_models() if not m._meta.managed]


"""
Copy from django.test.utils.setup_databases.

Manually create unmanaged models
"""


def setup_databases(
    verbosity,
    interactive,
    *,
    time_keeper=None,
    keepdb=False,
    debug_sql=False,
    parallel=0,
    aliases=None,
    serialized_aliases=None,
    **kwargs,
):
    """Create the test databases."""
    if time_keeper is None:
        time_keeper = utils.NullTimeKeeper()

    test_databases, mirrored_aliases = utils.get_unique_databases_and_mirrors(aliases)

    old_names = []

    for db_name, aliases in test_databases.values():
        first_alias = None
        for alias in aliases:
            connection = connections[alias]
            old_names.append((connection, db_name, first_alias is None))

            # Actually create the database for the first connection
            if first_alias is None:
                first_alias = alias
                with time_keeper.timed("  Creating '%s'" % alias):
                    serialize_alias = (
                        serialized_aliases is None or alias in serialized_aliases
                    )
                    test_database_name = connection.creation.create_test_db(
                        verbosity=verbosity,
                        autoclobber=not interactive,
                        keepdb=keepdb,
                        serialize=serialize_alias,
                    )
                    # Create unmanaged models
                    unmanaged_models = get_unmanaged_models()
                    with connection.schema_editor() as schema_editor:
                        with connection.cursor() as cursor:
                            for model in unmanaged_models:
                                cursor.execute(
                                    """
                                    SELECT TABLE_NAME 
                                    FROM information_schema.tables 
                                    WHERE TABLE_SCHEMA = %s
                                    AND TABLE_NAME = %s;
                                    """,
                                    (test_database_name, model._meta.db_table),
                                )
                                if cursor.fetchone() is None:
                                    schema_editor.create_model(model)

                if parallel > 1:
                    for index in range(parallel):
                        with time_keeper.timed("  Cloning '%s'" % alias):
                            connection.creation.clone_test_db(
                                suffix=str(index + 1),
                                verbosity=verbosity,
                                keepdb=keepdb,
                            )
            # Configure all other connections as mirrors of the first one
            else:
                connections[alias].creation.set_as_test_mirror(
                    connections[first_alias].settings_dict
                )

    # Configure the test mirrors.
    for alias, mirror_alias in mirrored_aliases.items():
        connections[alias].creation.set_as_test_mirror(
            connections[mirror_alias].settings_dict
        )

    if debug_sql:
        for alias in connections:
            connections[alias].force_debug_cursor = True

    return old_names