table name is a list 15216 regression test for 9991 real types in postgres that s fieldname fieldnameothertable othertable removing a field shouldn t disturb getrelations 17785 indexes have the orders key with a list of asc desc values different backends have different values for same constraints primary key unique constraint unique index mysql pk1 uniq1 idx1 pk0 uniq1 idx1 pk0 uniq1 idx1 postgresql pk1 uniq1 idx0 pk0 uniq1 idx0 pk0 uniq1 idx1 sqlite pk1 uniq0 idx0 pk0 uniq1 idx0 pk0 uniq1 idx1 test custom constraints test field constraints some databases e g oracle include additional check constraints all constraints are accounted for table name is a list 15216 regression test for 9991 real types in postgres that s field_name field_name_other_table other_table removing a field shouldn t disturb get_relations 17785 indexes have the orders key with a list of asc desc values 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 test custom constraints test field constraints some databases e g oracle include additional check constraints all constraints are accounted for
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,
    DbCommentModel,
    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):
        
        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[2]
                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],
        )

    @skipUnlessDBFeature("supports_comments")
    def test_db_comments(self):
        with connection.cursor() as cursor:
            desc = connection.introspection.get_table_description(
                cursor, DbCommentModel._meta.db_table
            )
            table_list = connection.introspection.get_table_list(cursor)
        self.assertEqual(
            ["'Name' column comment"],
            [field.comment for field in desc if field.name == "name"],
        )
        self.assertEqual(
            ["Custom table comment"],
            [
                table.comment
                for table in table_list
                if table.name == "introspection_dbcommentmodel"
            ],
        )

    
    @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
            )

        
        expected_relations = {
            "reporter_id": ("id", Reporter._meta.db_table),
            "response_to_id": ("id", Article._meta.db_table),
        }
        self.assertEqual(relations, expected_relations)

        
        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):
        
        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,
        ):
            
            
            
            
            
            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)

        
        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
        )
        
        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"]:
                
                
                field_constraints.add(name)
        
        self.assertEqual(
            constraints.keys() ^ (custom_constraints | field_constraints), set()
        )