django/tests/introspection/tests.py

from django.db import DatabaseError, connection
from django.db.models import Index
from django.test import TransactionTestCase, skipUnlessDBFeature

from .models import (
    Article, ArticleReporter, CheckConstraintModel, City, Comment, Country,
    District, Reporter, UniqueConstraintConditionModel,
)


class IntrospectionTests(TransactionTestCase):

    available_apps = ['introspection']

    def test_table_names(self):
        tl = connection.introspection.table_names()
        self.assertEqual(tl, sorted(tl))
        self.assertIn(Reporter._meta.db_table, tl, "'%s' isn't in table_list()." % Reporter._meta.db_table)
        self.assertIn(Article._meta.db_table, tl, "'%s' isn't in table_list()." % Article._meta.db_table)

    def test_django_table_names(self):
        with connection.cursor() as cursor:
            cursor.execute('CREATE TABLE django_ixn_test_table (id INTEGER);')
            tl = connection.introspection.django_table_names()
            cursor.execute("DROP TABLE django_ixn_test_table;")
            self.assertNotIn('django_ixn_test_table', tl,
                             "django_table_names() returned a non-Django table")

    def test_django_table_names_retval_type(self):
        # Table name is a list #15216
        tl = connection.introspection.django_table_names(only_existing=True)
        self.assertIs(type(tl), list)
        tl = connection.introspection.django_table_names(only_existing=False)
        self.assertIs(type(tl), list)

    def test_table_names_with_views(self):
        with connection.cursor() as cursor:
            try:
                cursor.execute(
                    'CREATE VIEW introspection_article_view AS SELECT headline '
                    'from introspection_article;')
            except DatabaseError as e:
                if 'insufficient privileges' in str(e):
                    self.fail("The test user has no CREATE VIEW privileges")
                else:
                    raise
        try:
            self.assertIn('introspection_article_view', connection.introspection.table_names(include_views=True))
            self.assertNotIn('introspection_article_view', connection.introspection.table_names())
        finally:
            with connection.cursor() as cursor:
                cursor.execute('DROP VIEW introspection_article_view')

    def test_unmanaged_through_model(self):
        tables = connection.introspection.django_table_names()
        self.assertNotIn(ArticleReporter._meta.db_table, tables)

    def test_installed_models(self):
        tables = [Article._meta.db_table, Reporter._meta.db_table]
        models = connection.introspection.installed_models(tables)
        self.assertEqual(models, {Article, Reporter})

    def test_sequence_list(self):
        sequences = connection.introspection.sequence_list()
        reporter_seqs = [seq for seq in sequences if seq['table'] == Reporter._meta.db_table]
        self.assertEqual(len(reporter_seqs), 1, 'Reporter sequence not found in sequence_list()')
        self.assertEqual(reporter_seqs[0]['column'], 'id')

    def test_get_table_description_names(self):
        with connection.cursor() as cursor:
            desc = connection.introspection.get_table_description(cursor, Reporter._meta.db_table)
        self.assertEqual([r[0] for r in desc],
                         [f.column for f in Reporter._meta.fields])

    def test_get_table_description_types(self):
        with connection.cursor() as cursor:
            desc = connection.introspection.get_table_description(cursor, Reporter._meta.db_table)
        self.assertEqual(
            [connection.introspection.get_field_type(r[1], r) for r in desc],
            [
                connection.features.introspected_field_types[field] for field in (
                    'AutoField', 'CharField', 'CharField', 'CharField',
                    'BigIntegerField', 'BinaryField', 'SmallIntegerField',
                    'DurationField',
                )
            ],
        )

    def test_get_table_description_col_lengths(self):
        with connection.cursor() as cursor:
            desc = connection.introspection.get_table_description(cursor, Reporter._meta.db_table)
        self.assertEqual(
            [r[3] for r in desc if connection.introspection.get_field_type(r[1], r) == 'CharField'],
            [30, 30, 254]
        )

    def test_get_table_description_nullable(self):
        with connection.cursor() as cursor:
            desc = connection.introspection.get_table_description(cursor, Reporter._meta.db_table)
        nullable_by_backend = connection.features.interprets_empty_strings_as_nulls
        self.assertEqual(
            [r[6] for r in desc],
            [False, nullable_by_backend, nullable_by_backend, nullable_by_backend, True, True, False, False]
        )

    def test_bigautofield(self):
        with connection.cursor() as cursor:
            desc = connection.introspection.get_table_description(cursor, City._meta.db_table)
        self.assertIn(
            connection.features.introspected_field_types['BigAutoField'],
            [connection.introspection.get_field_type(r[1], r) for r in desc],
        )

    def test_smallautofield(self):
        with connection.cursor() as cursor:
            desc = connection.introspection.get_table_description(cursor, Country._meta.db_table)
        self.assertIn(
            connection.features.introspected_field_types['SmallAutoField'],
            [connection.introspection.get_field_type(r[1], r) for r in desc],
        )

    # Regression test for #9991 - 'real' types in postgres
    @skipUnlessDBFeature('has_real_datatype')
    def test_postgresql_real_type(self):
        with connection.cursor() as cursor:
            cursor.execute("CREATE TABLE django_ixn_real_test_table (number REAL);")
            desc = connection.introspection.get_table_description(cursor, 'django_ixn_real_test_table')
            cursor.execute('DROP TABLE django_ixn_real_test_table;')
        self.assertEqual(connection.introspection.get_field_type(desc[0][1], desc[0]), 'FloatField')

    @skipUnlessDBFeature('can_introspect_foreign_keys')
    def test_get_relations(self):
        with connection.cursor() as cursor:
            relations = connection.introspection.get_relations(cursor, Article._meta.db_table)

        # That's {field_name: (field_name_other_table, other_table)}
        expected_relations = {
            'reporter_id': ('id', Reporter._meta.db_table),
            'response_to_id': ('id', Article._meta.db_table),
        }
        self.assertEqual(relations, expected_relations)

        # Removing a field shouldn't disturb get_relations (#17785)
        body = Article._meta.get_field('body')
        with connection.schema_editor() as editor:
            editor.remove_field(Article, body)
        with connection.cursor() as cursor:
            relations = connection.introspection.get_relations(cursor, Article._meta.db_table)
        with connection.schema_editor() as editor:
            editor.add_field(Article, body)
        self.assertEqual(relations, expected_relations)

    def test_get_primary_key_column(self):
        with connection.cursor() as cursor:
            primary_key_column = connection.introspection.get_primary_key_column(cursor, Article._meta.db_table)
            pk_fk_column = connection.introspection.get_primary_key_column(cursor, District._meta.db_table)
        self.assertEqual(primary_key_column, 'id')
        self.assertEqual(pk_fk_column, 'city_id')

    def test_get_constraints_index_types(self):
        with connection.cursor() as cursor:
            constraints = connection.introspection.get_constraints(cursor, Article._meta.db_table)
        index = {}
        index2 = {}
        for val in constraints.values():
            if val['columns'] == ['headline', 'pub_date']:
                index = val
            if val['columns'] == ['headline', 'response_to_id', 'pub_date', 'reporter_id']:
                index2 = val
        self.assertEqual(index['type'], Index.suffix)
        self.assertEqual(index2['type'], Index.suffix)

    @skipUnlessDBFeature('supports_index_column_ordering')
    def test_get_constraints_indexes_orders(self):
        """
        Indexes have the 'orders' key with a list of 'ASC'/'DESC' values.
        """
        with connection.cursor() as cursor:
            constraints = connection.introspection.get_constraints(cursor, Article._meta.db_table)
        indexes_verified = 0
        expected_columns = [
            ['headline', 'pub_date'],
            ['headline', 'response_to_id', 'pub_date', 'reporter_id'],
        ]
        if connection.features.indexes_foreign_keys:
            expected_columns += [
                ['reporter_id'],
                ['response_to_id'],
            ]
        for val in constraints.values():
            if val['index'] and not (val['primary_key'] or val['unique']):
                self.assertIn(val['columns'], expected_columns)
                self.assertEqual(val['orders'], ['ASC'] * len(val['columns']))
                indexes_verified += 1
        self.assertEqual(indexes_verified, len(expected_columns))

    @skipUnlessDBFeature('supports_index_column_ordering', 'supports_partial_indexes')
    def test_get_constraints_unique_indexes_orders(self):
        with connection.cursor() as cursor:
            constraints = connection.introspection.get_constraints(
                cursor,
                UniqueConstraintConditionModel._meta.db_table,
            )
        self.assertIn('cond_name_without_color_uniq', constraints)
        constraint = constraints['cond_name_without_color_uniq']
        self.assertIs(constraint['unique'], True)
        self.assertEqual(constraint['columns'], ['name'])
        self.assertEqual(constraint['orders'], ['ASC'])

    def test_get_constraints(self):
        def assertDetails(details, cols, primary_key=False, unique=False, index=False, check=False, foreign_key=None):
            # Different backends have different values for same constraints:
            #               PRIMARY KEY     UNIQUE CONSTRAINT    UNIQUE INDEX
            # MySQL      pk=1 uniq=1 idx=1  pk=0 uniq=1 idx=1  pk=0 uniq=1 idx=1
            # PostgreSQL pk=1 uniq=1 idx=0  pk=0 uniq=1 idx=0  pk=0 uniq=1 idx=1
            # SQLite     pk=1 uniq=0 idx=0  pk=0 uniq=1 idx=0  pk=0 uniq=1 idx=1
            if details['primary_key']:
                details['unique'] = True
            if details['unique']:
                details['index'] = False
            self.assertEqual(details['columns'], cols)
            self.assertEqual(details['primary_key'], primary_key)
            self.assertEqual(details['unique'], unique)
            self.assertEqual(details['index'], index)
            self.assertEqual(details['check'], check)
            self.assertEqual(details['foreign_key'], foreign_key)

        # Test custom constraints
        custom_constraints = {
            'article_email_pub_date_uniq',
            'email_pub_date_idx',
        }
        with connection.cursor() as cursor:
            constraints = connection.introspection.get_constraints(cursor, Comment._meta.db_table)
            if (
                connection.features.supports_column_check_constraints and
                connection.features.can_introspect_check_constraints
            ):
                constraints.update(
                    connection.introspection.get_constraints(cursor, CheckConstraintModel._meta.db_table)
                )
                custom_constraints.add('up_votes_gte_0_check')
                assertDetails(constraints['up_votes_gte_0_check'], ['up_votes'], check=True)
        assertDetails(constraints['article_email_pub_date_uniq'], ['article_id', 'email', 'pub_date'], unique=True)
        assertDetails(constraints['email_pub_date_idx'], ['email', 'pub_date'], index=True)
        # Test field constraints
        field_constraints = set()
        for name, details in constraints.items():
            if name in custom_constraints:
                continue
            elif details['columns'] == ['up_votes'] and details['check']:
                assertDetails(details, ['up_votes'], check=True)
                field_constraints.add(name)
            elif details['columns'] == ['voting_number'] and details['check']:
                assertDetails(details, ['voting_number'], check=True)
                field_constraints.add(name)
            elif details['columns'] == ['ref'] and details['unique']:
                assertDetails(details, ['ref'], unique=True)
                field_constraints.add(name)
            elif details['columns'] == ['voting_number'] and details['unique']:
                assertDetails(details, ['voting_number'], unique=True)
                field_constraints.add(name)
            elif details['columns'] == ['article_id'] and details['index']:
                assertDetails(details, ['article_id'], index=True)
                field_constraints.add(name)
            elif details['columns'] == ['id'] and details['primary_key']:
                assertDetails(details, ['id'], primary_key=True, unique=True)
                field_constraints.add(name)
            elif details['columns'] == ['article_id'] and details['foreign_key']:
                assertDetails(details, ['article_id'], foreign_key=('introspection_article', 'id'))
                field_constraints.add(name)
            elif details['check']:
                # Some databases (e.g. Oracle) include additional check
                # constraints.
                field_constraints.add(name)
        # All constraints are accounted for.
        self.assertEqual(constraints.keys() ^ (custom_constraints | field_constraints), set())