JezK

Edit File: test_reflection.py

import operator
import re

import sqlalchemy as sa
from .. import config
from .. import engines
from .. import eq_
from .. import expect_warnings
from .. import fixtures
from .. import is_
from ..provision import get_temp_table_name
from ..provision import temp_table_keyword_args
from ..schema import Column
from ..schema import Table
from ... import event
from ... import ForeignKey
from ... import func
from ... import Identity
from ... import inspect
from ... import Integer
from ... import MetaData
from ... import String
from ... import testing
from ... import types as sql_types
from ...schema import DDL
from ...schema import Index
from ...sql.elements import quoted_name
from ...sql.schema import BLANK_SCHEMA
from ...testing import is_false
from ...testing import is_true

metadata, users = None, None

class OneConnectionTablesTest(fixtures.TablesTest):
    @classmethod
    def setup_bind(cls):
        # TODO: when temp tables are subject to server reset,
        # this will also have to disable that server reset from
        # happening
        if config.requirements.independent_connections.enabled:
            from sqlalchemy import pool

return engines.testing_engine(
                options=dict(poolclass=pool.StaticPool, scope="class"),
            )
        else:
            return config.db

class HasTableTest(OneConnectionTablesTest):
    __backend__ = True

@classmethod
    def define_tables(cls, metadata):
        Table(
            "test_table",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("data", String(50)),
        )
        if testing.requires.schemas.enabled:
            Table(
                "test_table_s",
                metadata,
                Column("id", Integer, primary_key=True),
                Column("data", String(50)),
                schema=config.test_schema,
            )

if testing.requires.view_reflection:
            cls.define_views(metadata)
        if testing.requires.has_temp_table.enabled:
            cls.define_temp_tables(metadata)

@classmethod
    def define_views(cls, metadata):
        query = "CREATE VIEW vv AS SELECT id, data FROM test_table"

event.listen(metadata, "after_create", DDL(query))
        event.listen(metadata, "before_drop", DDL("DROP VIEW vv"))

if testing.requires.schemas.enabled:
            query = (
                "CREATE VIEW %s.vv AS SELECT id, data FROM %s.test_table_s"
                % (
                    config.test_schema,
                    config.test_schema,
                )
            )
            event.listen(metadata, "after_create", DDL(query))
            event.listen(
                metadata,
                "before_drop",
                DDL("DROP VIEW %s.vv" % (config.test_schema)),
            )

@classmethod
    def temp_table_name(cls):
        return get_temp_table_name(
            config, config.db, "user_tmp_%s" % (config.ident,)
        )

@classmethod
    def define_temp_tables(cls, metadata):
        kw = temp_table_keyword_args(config, config.db)
        table_name = cls.temp_table_name()
        user_tmp = Table(
            table_name,
            metadata,
            Column("id", sa.INT, primary_key=True),
            Column("name", sa.VARCHAR(50)),
            **kw
        )
        if (
            testing.requires.view_reflection.enabled
            and testing.requires.temporary_views.enabled
        ):
            event.listen(
                user_tmp,
                "after_create",
                DDL(
                    "create temporary view user_tmp_v as "
                    "select * from user_tmp_%s" % config.ident
                ),
            )
            event.listen(user_tmp, "before_drop", DDL("drop view user_tmp_v"))

def test_has_table(self):
        with config.db.begin() as conn:
            is_true(config.db.dialect.has_table(conn, "test_table"))
            is_false(config.db.dialect.has_table(conn, "test_table_s"))
            is_false(config.db.dialect.has_table(conn, "nonexistent_table"))

@testing.requires.schemas
    def test_has_table_schema(self):
        with config.db.begin() as conn:
            is_false(
                config.db.dialect.has_table(
                    conn, "test_table", schema=config.test_schema
                )
            )
            is_true(
                config.db.dialect.has_table(
                    conn, "test_table_s", schema=config.test_schema
                )
            )
            is_false(
                config.db.dialect.has_table(
                    conn, "nonexistent_table", schema=config.test_schema
                )
            )

@testing.fails_on(
        "oracle",
        "per #8700 this remains at its previous behavior of not "
        "working within 1.4.",
    )
    @testing.requires.views
    def test_has_table_view(self, connection):
        insp = inspect(connection)
        is_true(insp.has_table("vv"))

@testing.requires.has_temp_table
    def test_has_table_temp_table(self, connection):
        insp = inspect(connection)
        temp_table_name = self.temp_table_name()
        is_true(insp.has_table(temp_table_name))

@testing.requires.has_temp_table
    @testing.requires.view_reflection
    @testing.requires.temporary_views
    def test_has_table_temp_view(self, connection):
        insp = inspect(connection)
        is_true(insp.has_table("user_tmp_v"))

@testing.fails_on(
        "oracle",
        "per #8700 this remains at its previous behavior of not "
        "working within 1.4",
    )
    @testing.requires.views
    @testing.requires.schemas
    def test_has_table_view_schema(self, connection):
        insp = inspect(connection)
        is_true(insp.has_table("vv", config.test_schema))

class HasIndexTest(fixtures.TablesTest):
    __backend__ = True

@classmethod
    def define_tables(cls, metadata):
        tt = Table(
            "test_table",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("data", String(50)),
        )
        Index("my_idx", tt.c.data)

if testing.requires.schemas.enabled:
            tt = Table(
                "test_table",
                metadata,
                Column("id", Integer, primary_key=True),
                Column("data", String(50)),
                schema=config.test_schema,
            )
            Index("my_idx_s", tt.c.data)

def test_has_index(self):
        with config.db.begin() as conn:
            assert config.db.dialect.has_index(conn, "test_table", "my_idx")
            assert not config.db.dialect.has_index(
                conn, "test_table", "my_idx_s"
            )
            assert not config.db.dialect.has_index(
                conn, "nonexistent_table", "my_idx"
            )
            assert not config.db.dialect.has_index(
                conn, "test_table", "nonexistent_idx"
            )

@testing.requires.schemas
    def test_has_index_schema(self):
        with config.db.begin() as conn:
            assert config.db.dialect.has_index(
                conn, "test_table", "my_idx_s", schema=config.test_schema
            )
            assert not config.db.dialect.has_index(
                conn, "test_table", "my_idx", schema=config.test_schema
            )
            assert not config.db.dialect.has_index(
                conn,
                "nonexistent_table",
                "my_idx_s",
                schema=config.test_schema,
            )
            assert not config.db.dialect.has_index(
                conn,
                "test_table",
                "nonexistent_idx_s",
                schema=config.test_schema,
            )

class QuotedNameArgumentTest(fixtures.TablesTest):
    run_create_tables = "once"
    __backend__ = True

@classmethod
    def define_tables(cls, metadata):
        Table(
            "quote ' one",
            metadata,
            Column("id", Integer),
            Column("name", String(50)),
            Column("data", String(50)),
            Column("related_id", Integer),
            sa.PrimaryKeyConstraint("id", name="pk quote ' one"),
            sa.Index("ix quote ' one", "name"),
            sa.UniqueConstraint(
                "data",
                name="uq quote' one",
            ),
            sa.ForeignKeyConstraint(
                ["id"], ["related.id"], name="fk quote ' one"
            ),
            sa.CheckConstraint("name != 'foo'", name="ck quote ' one"),
            comment=r"""quote ' one comment""",
            test_needs_fk=True,
        )

if testing.requires.symbol_names_w_double_quote.enabled:
            Table(
                'quote " two',
                metadata,
                Column("id", Integer),
                Column("name", String(50)),
                Column("data", String(50)),
                Column("related_id", Integer),
                sa.PrimaryKeyConstraint("id", name='pk quote " two'),
                sa.Index('ix quote " two', "name"),
                sa.UniqueConstraint(
                    "data",
                    name='uq quote" two',
                ),
                sa.ForeignKeyConstraint(
                    ["id"], ["related.id"], name='fk quote " two'
                ),
                sa.CheckConstraint("name != 'foo'", name='ck quote " two '),
                comment=r"""quote " two comment""",
                test_needs_fk=True,
            )

Table(
            "related",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("related", Integer),
            test_needs_fk=True,
        )

if testing.requires.view_column_reflection.enabled:

if testing.requires.symbol_names_w_double_quote.enabled:
                names = [
                    "quote ' one",
                    'quote " two',
                ]
            else:
                names = [
                    "quote ' one",
                ]
            for name in names:
                query = "CREATE VIEW %s AS SELECT * FROM %s" % (
                    config.db.dialect.identifier_preparer.quote(
                        "view %s" % name
                    ),
                    config.db.dialect.identifier_preparer.quote(name),
                )

event.listen(metadata, "after_create", DDL(query))
                event.listen(
                    metadata,
                    "before_drop",
                    DDL(
                        "DROP VIEW %s"
                        % config.db.dialect.identifier_preparer.quote(
                            "view %s" % name
                        )
                    ),
                )

def quote_fixtures(fn):
        return testing.combinations(
            ("quote ' one",),
            ('quote " two', testing.requires.symbol_names_w_double_quote),
        )(fn)

@quote_fixtures
    def test_get_table_options(self, name):
        insp = inspect(config.db)

insp.get_table_options(name)

@quote_fixtures
    @testing.requires.view_column_reflection
    def test_get_view_definition(self, name):
        insp = inspect(config.db)
        assert insp.get_view_definition("view %s" % name)

@quote_fixtures
    def test_get_columns(self, name):
        insp = inspect(config.db)
        assert insp.get_columns(name)

@quote_fixtures
    def test_get_pk_constraint(self, name):
        insp = inspect(config.db)
        assert insp.get_pk_constraint(name)

@quote_fixtures
    def test_get_foreign_keys(self, name):
        insp = inspect(config.db)
        assert insp.get_foreign_keys(name)

@quote_fixtures
    def test_get_indexes(self, name):
        insp = inspect(config.db)
        assert insp.get_indexes(name)

@quote_fixtures
    @testing.requires.unique_constraint_reflection
    def test_get_unique_constraints(self, name):
        insp = inspect(config.db)
        assert insp.get_unique_constraints(name)

@quote_fixtures
    @testing.requires.comment_reflection
    def test_get_table_comment(self, name):
        insp = inspect(config.db)
        assert insp.get_table_comment(name)

@quote_fixtures
    @testing.requires.check_constraint_reflection
    def test_get_check_constraints(self, name):
        insp = inspect(config.db)
        assert insp.get_check_constraints(name)

class ComponentReflectionTest(OneConnectionTablesTest):
    run_inserts = run_deletes = None

__backend__ = True

@classmethod
    def define_tables(cls, metadata):
        cls.define_reflected_tables(metadata, None)
        if testing.requires.schemas.enabled:
            cls.define_reflected_tables(metadata, testing.config.test_schema)

@classmethod
    def define_reflected_tables(cls, metadata, schema):
        if schema:
            schema_prefix = schema + "."
        else:
            schema_prefix = ""

if testing.requires.self_referential_foreign_keys.enabled:
            users = Table(
                "users",
                metadata,
                Column("user_id", sa.INT, primary_key=True),
                Column("test1", sa.CHAR(5), nullable=False),
                Column("test2", sa.Float(5), nullable=False),
                Column(
                    "parent_user_id",
                    sa.Integer,
                    sa.ForeignKey(
                        "%susers.user_id" % schema_prefix, name="user_id_fk"
                    ),
                ),
                schema=schema,
                test_needs_fk=True,
            )
        else:
            users = Table(
                "users",
                metadata,
                Column("user_id", sa.INT, primary_key=True),
                Column("test1", sa.CHAR(5), nullable=False),
                Column("test2", sa.Float(5), nullable=False),
                schema=schema,
                test_needs_fk=True,
            )

Table(
            "dingalings",
            metadata,
            Column("dingaling_id", sa.Integer, primary_key=True),
            Column(
                "address_id",
                sa.Integer,
                sa.ForeignKey("%semail_addresses.address_id" % schema_prefix),
            ),
            Column("data", sa.String(30)),
            schema=schema,
            test_needs_fk=True,
        )
        Table(
            "email_addresses",
            metadata,
            Column("address_id", sa.Integer),
            Column(
                "remote_user_id", sa.Integer, sa.ForeignKey(users.c.user_id)
            ),
            Column("email_address", sa.String(20)),
            sa.PrimaryKeyConstraint("address_id", name="email_ad_pk"),
            schema=schema,
            test_needs_fk=True,
        )
        Table(
            "comment_test",
            metadata,
            Column("id", sa.Integer, primary_key=True, comment="id comment"),
            Column("data", sa.String(20), comment="data % comment"),
            Column(
                "d2",
                sa.String(20),
                comment=r"""Comment types type speedily ' " \ '' Fun!""",
            ),
            schema=schema,
            comment=r"""the test % ' " \ table comment""",
        )

if testing.requires.cross_schema_fk_reflection.enabled:
            if schema is None:
                Table(
                    "local_table",
                    metadata,
                    Column("id", sa.Integer, primary_key=True),
                    Column("data", sa.String(20)),
                    Column(
                        "remote_id",
                        ForeignKey(
                            "%s.remote_table_2.id" % testing.config.test_schema
                        ),
                    ),
                    test_needs_fk=True,
                    schema=config.db.dialect.default_schema_name,
                )
            else:
                Table(
                    "remote_table",
                    metadata,
                    Column("id", sa.Integer, primary_key=True),
                    Column(
                        "local_id",
                        ForeignKey(
                            "%s.local_table.id"
                            % config.db.dialect.default_schema_name
                        ),
                    ),
                    Column("data", sa.String(20)),
                    schema=schema,
                    test_needs_fk=True,
                )
                Table(
                    "remote_table_2",
                    metadata,
                    Column("id", sa.Integer, primary_key=True),
                    Column("data", sa.String(20)),
                    schema=schema,
                    test_needs_fk=True,
                )

if testing.requires.index_reflection.enabled:
            cls.define_index(metadata, users)

if not schema:
                # test_needs_fk is at the moment to force MySQL InnoDB
                noncol_idx_test_nopk = Table(
                    "noncol_idx_test_nopk",
                    metadata,
                    Column("q", sa.String(5)),
                    test_needs_fk=True,
                )

noncol_idx_test_pk = Table(
                    "noncol_idx_test_pk",
                    metadata,
                    Column("id", sa.Integer, primary_key=True),
                    Column("q", sa.String(5)),
                    test_needs_fk=True,
                )

if testing.requires.indexes_with_ascdesc.enabled:
                    Index("noncol_idx_nopk", noncol_idx_test_nopk.c.q.desc())
                    Index("noncol_idx_pk", noncol_idx_test_pk.c.q.desc())

if testing.requires.view_column_reflection.enabled:
            cls.define_views(metadata, schema)
        if not schema and testing.requires.temp_table_reflection.enabled:
            cls.define_temp_tables(metadata)

@classmethod
    def define_temp_tables(cls, metadata):
        kw = temp_table_keyword_args(config, config.db)
        table_name = get_temp_table_name(
            config, config.db, "user_tmp_%s" % config.ident
        )
        user_tmp = Table(
            table_name,
            metadata,
            Column("id", sa.INT, primary_key=True),
            Column("name", sa.VARCHAR(50)),
            Column("foo", sa.INT),
            # disambiguate temp table unique constraint names.  this is
            # pretty arbitrary for a generic dialect however we are doing
            # it to suit SQL Server which will produce name conflicts for
            # unique constraints created against temp tables in different
            # databases.
            # https://www.arbinada.com/en/node/1645
            sa.UniqueConstraint("name", name="user_tmp_uq_%s" % config.ident),
            sa.Index("user_tmp_ix", "foo"),
            **kw
        )
        if (
            testing.requires.view_reflection.enabled
            and testing.requires.temporary_views.enabled
        ):
            event.listen(
                user_tmp,
                "after_create",
                DDL(
                    "create temporary view user_tmp_v as "
                    "select * from user_tmp_%s" % config.ident
                ),
            )
            event.listen(user_tmp, "before_drop", DDL("drop view user_tmp_v"))

@classmethod
    def define_index(cls, metadata, users):
        Index("users_t_idx", users.c.test1, users.c.test2)
        Index("users_all_idx", users.c.user_id, users.c.test2, users.c.test1)

@classmethod
    def define_views(cls, metadata, schema):
        for table_name in ("users", "email_addresses"):
            fullname = table_name
            if schema:
                fullname = "%s.%s" % (schema, table_name)
            view_name = fullname + "_v"
            query = "CREATE VIEW %s AS SELECT * FROM %s" % (
                view_name,
                fullname,
            )

event.listen(metadata, "after_create", DDL(query))
            event.listen(
                metadata, "before_drop", DDL("DROP VIEW %s" % view_name)
            )

@testing.requires.schema_reflection
    def test_get_schema_names(self):
        insp = inspect(self.bind)

self.assert_(testing.config.test_schema in insp.get_schema_names())

@testing.requires.schema_reflection
    def test_get_schema_names_w_translate_map(self, connection):
        """test #7300"""

connection = connection.execution_options(
            schema_translate_map={
                "foo": "bar",
                BLANK_SCHEMA: testing.config.test_schema,
            }
        )
        insp = inspect(connection)

self.assert_(testing.config.test_schema in insp.get_schema_names())

@testing.requires.schema_reflection
    def test_dialect_initialize(self):
        engine = engines.testing_engine()
        inspect(engine)
        assert hasattr(engine.dialect, "default_schema_name")

@testing.requires.schema_reflection
    def test_get_default_schema_name(self):
        insp = inspect(self.bind)
        eq_(insp.default_schema_name, self.bind.dialect.default_schema_name)

@testing.requires.foreign_key_constraint_reflection
    @testing.combinations(
        (None, True, False, False),
        (None, True, False, True, testing.requires.schemas),
        ("foreign_key", True, False, False),
        (None, False, True, False),
        (None, False, True, True, testing.requires.schemas),
        (None, True, True, False),
        (None, True, True, True, testing.requires.schemas),
        argnames="order_by,include_plain,include_views,use_schema",
    )
    def test_get_table_names(
        self, connection, order_by, include_plain, include_views, use_schema
    ):

if use_schema:
            schema = config.test_schema
        else:
            schema = None

_ignore_tables = [
            "comment_test",
            "noncol_idx_test_pk",
            "noncol_idx_test_nopk",
            "local_table",
            "remote_table",
            "remote_table_2",
        ]

insp = inspect(connection)

if include_views:
            table_names = insp.get_view_names(schema)
            table_names.sort()
            answer = ["email_addresses_v", "users_v"]
            eq_(sorted(table_names), answer)

if include_plain:
            if order_by:
                tables = [
                    rec[0]
                    for rec in insp.get_sorted_table_and_fkc_names(schema)
                    if rec[0]
                ]
            else:
                tables = insp.get_table_names(schema)
            table_names = [t for t in tables if t not in _ignore_tables]

if order_by == "foreign_key":
                answer = ["users", "email_addresses", "dingalings"]
                eq_(table_names, answer)
            else:
                answer = ["dingalings", "email_addresses", "users"]
                eq_(sorted(table_names), answer)

@testing.requires.temp_table_names
    def test_get_temp_table_names(self):
        insp = inspect(self.bind)
        temp_table_names = insp.get_temp_table_names()
        eq_(sorted(temp_table_names), ["user_tmp_%s" % config.ident])

@testing.requires.view_reflection
    @testing.requires.temp_table_names
    @testing.requires.temporary_views
    def test_get_temp_view_names(self):
        insp = inspect(self.bind)
        temp_table_names = insp.get_temp_view_names()
        eq_(sorted(temp_table_names), ["user_tmp_v"])

@testing.requires.comment_reflection
    def test_get_comments(self):
        self._test_get_comments()

@testing.requires.comment_reflection
    @testing.requires.schemas
    def test_get_comments_with_schema(self):
        self._test_get_comments(testing.config.test_schema)

def _test_get_comments(self, schema=None):
        insp = inspect(self.bind)

eq_(
            insp.get_table_comment("comment_test", schema=schema),
            {"text": r"""the test % ' " \ table comment"""},
        )

eq_(insp.get_table_comment("users", schema=schema), {"text": None})

eq_(
            [
                {"name": rec["name"], "comment": rec["comment"]}
                for rec in insp.get_columns("comment_test", schema=schema)
            ],
            [
                {"comment": "id comment", "name": "id"},
                {"comment": "data % comment", "name": "data"},
                {
                    "comment": (
                        r"""Comment types type speedily ' " \ '' Fun!"""
                    ),
                    "name": "d2",
                },
            ],
        )

@testing.combinations(
        (False, False),
        (False, True, testing.requires.schemas),
        (True, False, testing.requires.view_reflection),
        (
            True,
            True,
            testing.requires.schemas + testing.requires.view_reflection,
        ),
        argnames="use_views,use_schema",
    )
    def test_get_columns(self, connection, use_views, use_schema):

if use_schema:
            schema = config.test_schema
        else:
            schema = None

users, addresses = (self.tables.users, self.tables.email_addresses)
        if use_views:
            table_names = ["users_v", "email_addresses_v"]
        else:
            table_names = ["users", "email_addresses"]

insp = inspect(connection)
        for table_name, table in zip(table_names, (users, addresses)):
            schema_name = schema
            cols = insp.get_columns(table_name, schema=schema_name)
            self.assert_(len(cols) > 0, len(cols))

# should be in order

for i, col in enumerate(table.columns):
                eq_(col.name, cols[i]["name"])
                ctype = cols[i]["type"].__class__
                ctype_def = col.type
                if isinstance(ctype_def, sa.types.TypeEngine):
                    ctype_def = ctype_def.__class__

# Oracle returns Date for DateTime.

if testing.against("oracle") and ctype_def in (
                    sql_types.Date,
                    sql_types.DateTime,
                ):
                    ctype_def = sql_types.Date

# assert that the desired type and return type share
                # a base within one of the generic types.

self.assert_(
                    len(
                        set(ctype.__mro__)
                        .intersection(ctype_def.__mro__)
                        .intersection(
                            [
                                sql_types.Integer,
                                sql_types.Numeric,
                                sql_types.DateTime,
                                sql_types.Date,
                                sql_types.Time,
                                sql_types.String,
                                sql_types._Binary,
                            ]
                        )
                    )
                    > 0,
                    "%s(%s), %s(%s)"
                    % (col.name, col.type, cols[i]["name"], ctype),
                )

if not col.primary_key:
                    assert cols[i]["default"] is None

@testing.requires.temp_table_reflection
    def test_get_temp_table_columns(self):
        table_name = get_temp_table_name(
            config, self.bind, "user_tmp_%s" % config.ident
        )
        user_tmp = self.tables[table_name]
        insp = inspect(self.bind)
        cols = insp.get_columns(table_name)
        self.assert_(len(cols) > 0, len(cols))

for i, col in enumerate(user_tmp.columns):
            eq_(col.name, cols[i]["name"])

@testing.requires.temp_table_reflection
    @testing.requires.view_column_reflection
    @testing.requires.temporary_views
    def test_get_temp_view_columns(self):
        insp = inspect(self.bind)
        cols = insp.get_columns("user_tmp_v")
        eq_([col["name"] for col in cols], ["id", "name", "foo"])

@testing.combinations(
        (False,), (True, testing.requires.schemas), argnames="use_schema"
    )
    @testing.requires.primary_key_constraint_reflection
    def test_get_pk_constraint(self, connection, use_schema):
        if use_schema:
            schema = testing.config.test_schema
        else:
            schema = None

users, addresses = self.tables.users, self.tables.email_addresses
        insp = inspect(connection)

users_cons = insp.get_pk_constraint(users.name, schema=schema)
        users_pkeys = users_cons["constrained_columns"]
        eq_(users_pkeys, ["user_id"])

addr_cons = insp.get_pk_constraint(addresses.name, schema=schema)
        addr_pkeys = addr_cons["constrained_columns"]
        eq_(addr_pkeys, ["address_id"])

with testing.requires.reflects_pk_names.fail_if():
            eq_(addr_cons["name"], "email_ad_pk")

@testing.combinations(
        (False,), (True, testing.requires.schemas), argnames="use_schema"
    )
    @testing.requires.foreign_key_constraint_reflection
    def test_get_foreign_keys(self, connection, use_schema):
        if use_schema:
            schema = config.test_schema
        else:
            schema = None

users, addresses = (self.tables.users, self.tables.email_addresses)
        insp = inspect(connection)
        expected_schema = schema
        # users

if testing.requires.self_referential_foreign_keys.enabled:
            users_fkeys = insp.get_foreign_keys(users.name, schema=schema)
            fkey1 = users_fkeys[0]

with testing.requires.named_constraints.fail_if():
                eq_(fkey1["name"], "user_id_fk")

eq_(fkey1["referred_schema"], expected_schema)
            eq_(fkey1["referred_table"], users.name)
            eq_(fkey1["referred_columns"], ["user_id"])
            if testing.requires.self_referential_foreign_keys.enabled:
                eq_(fkey1["constrained_columns"], ["parent_user_id"])

# addresses
        addr_fkeys = insp.get_foreign_keys(addresses.name, schema=schema)
        fkey1 = addr_fkeys[0]

with testing.requires.implicitly_named_constraints.fail_if():
            self.assert_(fkey1["name"] is not None)

eq_(fkey1["referred_schema"], expected_schema)
        eq_(fkey1["referred_table"], users.name)
        eq_(fkey1["referred_columns"], ["user_id"])
        eq_(fkey1["constrained_columns"], ["remote_user_id"])

@testing.requires.cross_schema_fk_reflection
    @testing.requires.schemas
    def test_get_inter_schema_foreign_keys(self):
        local_table, remote_table, remote_table_2 = self.tables(
            "%s.local_table" % self.bind.dialect.default_schema_name,
            "%s.remote_table" % testing.config.test_schema,
            "%s.remote_table_2" % testing.config.test_schema,
        )

insp = inspect(self.bind)

local_fkeys = insp.get_foreign_keys(local_table.name)
        eq_(len(local_fkeys), 1)

fkey1 = local_fkeys[0]
        eq_(fkey1["referred_schema"], testing.config.test_schema)
        eq_(fkey1["referred_table"], remote_table_2.name)
        eq_(fkey1["referred_columns"], ["id"])
        eq_(fkey1["constrained_columns"], ["remote_id"])

remote_fkeys = insp.get_foreign_keys(
            remote_table.name, schema=testing.config.test_schema
        )
        eq_(len(remote_fkeys), 1)

fkey2 = remote_fkeys[0]

assert fkey2["referred_schema"] in (
            None,
            self.bind.dialect.default_schema_name,
        )
        eq_(fkey2["referred_table"], local_table.name)
        eq_(fkey2["referred_columns"], ["id"])
        eq_(fkey2["constrained_columns"], ["local_id"])

def _assert_insp_indexes(self, indexes, expected_indexes):
        index_names = [d["name"] for d in indexes]
        for e_index in expected_indexes:
            assert e_index["name"] in index_names
            index = indexes[index_names.index(e_index["name"])]
            for key in e_index:
                eq_(e_index[key], index[key])

@testing.combinations(
        (False,), (True, testing.requires.schemas), argnames="use_schema"
    )
    def test_get_indexes(self, connection, use_schema):

if use_schema:
            schema = config.test_schema
        else:
            schema = None

# The database may decide to create indexes for foreign keys, etc.
        # so there may be more indexes than expected.
        insp = inspect(self.bind)
        indexes = insp.get_indexes("users", schema=schema)
        expected_indexes = [
            {
                "unique": False,
                "column_names": ["test1", "test2"],
                "name": "users_t_idx",
            },
            {
                "unique": False,
                "column_names": ["user_id", "test2", "test1"],
                "name": "users_all_idx",
            },
        ]
        self._assert_insp_indexes(indexes, expected_indexes)

@testing.combinations(
        ("noncol_idx_test_nopk", "noncol_idx_nopk"),
        ("noncol_idx_test_pk", "noncol_idx_pk"),
        argnames="tname,ixname",
    )
    @testing.requires.index_reflection
    @testing.requires.indexes_with_ascdesc
    def test_get_noncol_index(self, connection, tname, ixname):
        insp = inspect(connection)
        indexes = insp.get_indexes(tname)

# reflecting an index that has "x DESC" in it as the column.
        # the DB may or may not give us "x", but make sure we get the index
        # back, it has a name, it's connected to the table.
        expected_indexes = [{"unique": False, "name": ixname}]
        self._assert_insp_indexes(indexes, expected_indexes)

t = Table(tname, MetaData(), autoload_with=connection)
        eq_(len(t.indexes), 1)
        is_(list(t.indexes)[0].table, t)
        eq_(list(t.indexes)[0].name, ixname)

@testing.requires.temp_table_reflection
    @testing.requires.unique_constraint_reflection
    def test_get_temp_table_unique_constraints(self):
        insp = inspect(self.bind)
        reflected = insp.get_unique_constraints("user_tmp_%s" % config.ident)
        for refl in reflected:
            # Different dialects handle duplicate index and constraints
            # differently, so ignore this flag
            refl.pop("duplicates_index", None)
        eq_(
            reflected,
            [
                {
                    "column_names": ["name"],
                    "name": "user_tmp_uq_%s" % config.ident,
                }
            ],
        )

@testing.requires.temp_table_reflect_indexes
    def test_get_temp_table_indexes(self):
        insp = inspect(self.bind)
        table_name = get_temp_table_name(
            config, config.db, "user_tmp_%s" % config.ident
        )
        indexes = insp.get_indexes(table_name)
        for ind in indexes:
            ind.pop("dialect_options", None)
        expected = [
            {"unique": False, "column_names": ["foo"], "name": "user_tmp_ix"}
        ]
        if testing.requires.index_reflects_included_columns.enabled:
            expected[0]["include_columns"] = []
        eq_(
            [idx for idx in indexes if idx["name"] == "user_tmp_ix"],
            expected,
        )

@testing.combinations(
        (True, testing.requires.schemas), (False,), argnames="use_schema"
    )
    @testing.requires.unique_constraint_reflection
    def test_get_unique_constraints(self, metadata, connection, use_schema):
        # SQLite dialect needs to parse the names of the constraints
        # separately from what it gets from PRAGMA index_list(), and
        # then matches them up.  so same set of column_names in two
        # constraints will confuse it.    Perhaps we should no longer
        # bother with index_list() here since we have the whole
        # CREATE TABLE?

if use_schema:
            schema = config.test_schema
        else:
            schema = None
        uniques = sorted(
            [
                {"name": "unique_a", "column_names": ["a"]},
                {"name": "unique_a_b_c", "column_names": ["a", "b", "c"]},
                {"name": "unique_c_a_b", "column_names": ["c", "a", "b"]},
                {"name": "unique_asc_key", "column_names": ["asc", "key"]},
                {"name": "i.have.dots", "column_names": ["b"]},
                {"name": "i have spaces", "column_names": ["c"]},
            ],
            key=operator.itemgetter("name"),
        )
        table = Table(
            "testtbl",
            metadata,
            Column("a", sa.String(20)),
            Column("b", sa.String(30)),
            Column("c", sa.Integer),
            # reserved identifiers
            Column("asc", sa.String(30)),
            Column("key", sa.String(30)),
            schema=schema,
        )
        for uc in uniques:
            table.append_constraint(
                sa.UniqueConstraint(*uc["column_names"], name=uc["name"])
            )
        table.create(connection)

inspector = inspect(connection)
        reflected = sorted(
            inspector.get_unique_constraints("testtbl", schema=schema),
            key=operator.itemgetter("name"),
        )

names_that_duplicate_index = set()

eq_(len(uniques), len(reflected))

for orig, refl in zip(uniques, reflected):
            # Different dialects handle duplicate index and constraints
            # differently, so ignore this flag
            dupe = refl.pop("duplicates_index", None)
            if dupe:
                names_that_duplicate_index.add(dupe)
            eq_(orig, refl)

reflected_metadata = MetaData()
        reflected = Table(
            "testtbl",
            reflected_metadata,
            autoload_with=connection,
            schema=schema,
        )

# test "deduplicates for index" logic.   MySQL and Oracle
        # "unique constraints" are actually unique indexes (with possible
        # exception of a unique that is a dupe of another one in the case
        # of Oracle).  make sure # they aren't duplicated.
        idx_names = set([idx.name for idx in reflected.indexes])
        uq_names = set(
            [
                uq.name
                for uq in reflected.constraints
                if isinstance(uq, sa.UniqueConstraint)
            ]
        ).difference(["unique_c_a_b"])

assert not idx_names.intersection(uq_names)
        if names_that_duplicate_index:
            eq_(names_that_duplicate_index, idx_names)
            eq_(uq_names, set())

@testing.requires.view_reflection
    @testing.combinations(
        (False,), (True, testing.requires.schemas), argnames="use_schema"
    )
    def test_get_view_definition(self, connection, use_schema):
        if use_schema:
            schema = config.test_schema
        else:
            schema = None
        view_name1 = "users_v"
        view_name2 = "email_addresses_v"
        insp = inspect(connection)
        v1 = insp.get_view_definition(view_name1, schema=schema)
        self.assert_(v1)
        v2 = insp.get_view_definition(view_name2, schema=schema)
        self.assert_(v2)

# why is this here if it's PG specific ?
    @testing.combinations(
        ("users", False),
        ("users", True, testing.requires.schemas),
        argnames="table_name,use_schema",
    )
    @testing.only_on("postgresql", "PG specific feature")
    def test_get_table_oid(self, connection, table_name, use_schema):
        if use_schema:
            schema = config.test_schema
        else:
            schema = None
        insp = inspect(connection)
        oid = insp.get_table_oid(table_name, schema)
        self.assert_(isinstance(oid, int))

@testing.requires.table_reflection
    def test_autoincrement_col(self):
        """test that 'autoincrement' is reflected according to sqla's policy.

Don't mark this test as unsupported for any backend !

(technically it fails with MySQL InnoDB since "id" comes before "id2")

A backend is better off not returning "autoincrement" at all,
        instead of potentially returning "False" for an auto-incrementing
        primary key column.

"""

insp = inspect(self.bind)

for tname, cname in [
            ("users", "user_id"),
            ("email_addresses", "address_id"),
            ("dingalings", "dingaling_id"),
        ]:
            cols = insp.get_columns(tname)
            id_ = {c["name"]: c for c in cols}[cname]
            assert id_.get("autoincrement", True)

class TableNoColumnsTest(fixtures.TestBase):
    __requires__ = ("reflect_tables_no_columns",)
    __backend__ = True

@testing.fixture
    def table_no_columns(self, connection, metadata):
        Table("empty", metadata)
        metadata.create_all(connection)

@testing.fixture
    def view_no_columns(self, connection, metadata):
        Table("empty", metadata)
        metadata.create_all(connection)

Table("empty", metadata)
        event.listen(
            metadata,
            "after_create",
            DDL("CREATE VIEW empty_v AS SELECT * FROM empty"),
        )

# for transactional DDL the transaction is rolled back before this
        # drop statement is invoked
        event.listen(
            metadata, "before_drop", DDL("DROP VIEW IF EXISTS empty_v")
        )
        metadata.create_all(connection)

@testing.requires.reflect_tables_no_columns
    def test_reflect_table_no_columns(self, connection, table_no_columns):
        t2 = Table("empty", MetaData(), autoload_with=connection)
        eq_(list(t2.c), [])

@testing.requires.reflect_tables_no_columns
    def test_get_columns_table_no_columns(self, connection, table_no_columns):
        eq_(inspect(connection).get_columns("empty"), [])

@testing.requires.reflect_tables_no_columns
    def test_reflect_incl_table_no_columns(self, connection, table_no_columns):
        m = MetaData()
        m.reflect(connection)
        assert set(m.tables).intersection(["empty"])

@testing.requires.views
    @testing.requires.reflect_tables_no_columns
    def test_reflect_view_no_columns(self, connection, view_no_columns):
        t2 = Table("empty_v", MetaData(), autoload_with=connection)
        eq_(list(t2.c), [])

@testing.requires.views
    @testing.requires.reflect_tables_no_columns
    def test_get_columns_view_no_columns(self, connection, view_no_columns):
        eq_(inspect(connection).get_columns("empty_v"), [])

class ComponentReflectionTestExtra(fixtures.TestBase):

__backend__ = True

@testing.combinations(
        (True, testing.requires.schemas), (False,), argnames="use_schema"
    )
    @testing.requires.check_constraint_reflection
    def test_get_check_constraints(self, metadata, connection, use_schema):
        if use_schema:
            schema = config.test_schema
        else:
            schema = None

Table(
            "sa_cc",
            metadata,
            Column("a", Integer()),
            sa.CheckConstraint("a > 1 AND a < 5", name="cc1"),
            sa.CheckConstraint(
                "a = 1 OR (a > 2 AND a < 5)", name="UsesCasing"
            ),
            schema=schema,
        )

metadata.create_all(connection)

inspector = inspect(connection)
        reflected = sorted(
            inspector.get_check_constraints("sa_cc", schema=schema),
            key=operator.itemgetter("name"),
        )

# trying to minimize effect of quoting, parenthesis, etc.
        # may need to add more to this as new dialects get CHECK
        # constraint reflection support
        def normalize(sqltext):
            return " ".join(
                re.findall(r"and|\d|=|a|or|<|>", sqltext.lower(), re.I)
            )

reflected = [
            {"name": item["name"], "sqltext": normalize(item["sqltext"])}
            for item in reflected
        ]
        eq_(
            reflected,
            [
                {"name": "UsesCasing", "sqltext": "a = 1 or a > 2 and a < 5"},
                {"name": "cc1", "sqltext": "a > 1 and a < 5"},
            ],
        )

@testing.requires.indexes_with_expressions
    def test_reflect_expression_based_indexes(self, metadata, connection):
        t = Table(
            "t",
            metadata,
            Column("x", String(30)),
            Column("y", String(30)),
        )

Index("t_idx", func.lower(t.c.x), func.lower(t.c.y))

Index("t_idx_2", t.c.x)

metadata.create_all(connection)

insp = inspect(connection)

expected = [
            {
                "name": "t_idx_2",
                "column_names": ["x"],
                "unique": False,
                "dialect_options": {},
            }
        ]

if testing.requires.index_reflects_included_columns.enabled:
            expected[0]["include_columns"] = []
            expected[0]["dialect_options"] = {
                "%s_include" % connection.engine.name: []
            }

with expect_warnings(
            "Skipped unsupported reflection of expression-based index t_idx"
        ):
            eq_(insp.get_indexes("t"), expected)

@testing.requires.index_reflects_included_columns
    def test_reflect_covering_index(self, metadata, connection):
        t = Table(
            "t",
            metadata,
            Column("x", String(30)),
            Column("y", String(30)),
        )
        idx = Index("t_idx", t.c.x)
        idx.dialect_options[connection.engine.name]["include"] = ["y"]

metadata.create_all(connection)

insp = inspect(connection)

eq_(
            insp.get_indexes("t"),
            [
                {
                    "name": "t_idx",
                    "column_names": ["x"],
                    "include_columns": ["y"],
                    "unique": False,
                    "dialect_options": {
                        "%s_include" % connection.engine.name: ["y"]
                    },
                }
            ],
        )

t2 = Table("t", MetaData(), autoload_with=connection)
        eq_(
            list(t2.indexes)[0].dialect_options[connection.engine.name][
                "include"
            ],
            ["y"],
        )

def _type_round_trip(self, connection, metadata, *types):
        t = Table(
            "t",
            metadata,
            *[Column("t%d" % i, type_) for i, type_ in enumerate(types)]
        )
        t.create(connection)

return [c["type"] for c in inspect(connection).get_columns("t")]

@testing.requires.table_reflection
    def test_numeric_reflection(self, connection, metadata):
        for typ in self._type_round_trip(
            connection, metadata, sql_types.Numeric(18, 5)
        ):
            assert isinstance(typ, sql_types.Numeric)
            eq_(typ.precision, 18)
            eq_(typ.scale, 5)

@testing.requires.table_reflection
    def test_varchar_reflection(self, connection, metadata):
        typ = self._type_round_trip(
            connection, metadata, sql_types.String(52)
        )[0]
        assert isinstance(typ, sql_types.String)
        eq_(typ.length, 52)

@testing.requires.table_reflection
    def test_nullable_reflection(self, connection, metadata):
        t = Table(
            "t",
            metadata,
            Column("a", Integer, nullable=True),
            Column("b", Integer, nullable=False),
        )
        t.create(connection)
        eq_(
            dict(
                (col["name"], col["nullable"])
                for col in inspect(connection).get_columns("t")
            ),
            {"a": True, "b": False},
        )

@testing.combinations(
        (
            None,
            "CASCADE",
            None,
            testing.requires.foreign_key_constraint_option_reflection_ondelete,
        ),
        (
            None,
            None,
            "SET NULL",
            testing.requires.foreign_key_constraint_option_reflection_onupdate,
        ),
        (
            {},
            None,
            "NO ACTION",
            testing.requires.foreign_key_constraint_option_reflection_onupdate,
        ),
        (
            {},
            "NO ACTION",
            None,
            testing.requires.fk_constraint_option_reflection_ondelete_noaction,
        ),
        (
            None,
            None,
            "RESTRICT",
            testing.requires.fk_constraint_option_reflection_onupdate_restrict,
        ),
        (
            None,
            "RESTRICT",
            None,
            testing.requires.fk_constraint_option_reflection_ondelete_restrict,
        ),
        argnames="expected,ondelete,onupdate",
    )
    def test_get_foreign_key_options(
        self, connection, metadata, expected, ondelete, onupdate
    ):
        options = {}
        if ondelete:
            options["ondelete"] = ondelete
        if onupdate:
            options["onupdate"] = onupdate

if expected is None:
            expected = options

Table(
            "x",
            metadata,
            Column("id", Integer, primary_key=True),
            test_needs_fk=True,
        )

Table(
            "table",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("x_id", Integer, sa.ForeignKey("x.id", name="xid")),
            Column("test", String(10)),
            test_needs_fk=True,
        )

Table(
            "user",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("name", String(50), nullable=False),
            Column("tid", Integer),
            sa.ForeignKeyConstraint(
                ["tid"], ["table.id"], name="myfk", **options
            ),
            test_needs_fk=True,
        )

metadata.create_all(connection)

insp = inspect(connection)

# test 'options' is always present for a backend
        # that can reflect these, since alembic looks for this
        opts = insp.get_foreign_keys("table")[0]["options"]

eq_(dict((k, opts[k]) for k in opts if opts[k]), {})

opts = insp.get_foreign_keys("user")[0]["options"]
        eq_(opts, expected)
        # eq_(dict((k, opts[k]) for k in opts if opts[k]), expected)

class NormalizedNameTest(fixtures.TablesTest):
    __requires__ = ("denormalized_names",)
    __backend__ = True

@classmethod
    def define_tables(cls, metadata):
        Table(
            quoted_name("t1", quote=True),
            metadata,
            Column("id", Integer, primary_key=True),
        )
        Table(
            quoted_name("t2", quote=True),
            metadata,
            Column("id", Integer, primary_key=True),
            Column("t1id", ForeignKey("t1.id")),
        )

def test_reflect_lowercase_forced_tables(self):

m2 = MetaData()
        t2_ref = Table(
            quoted_name("t2", quote=True), m2, autoload_with=config.db
        )
        t1_ref = m2.tables["t1"]
        assert t2_ref.c.t1id.references(t1_ref.c.id)

m3 = MetaData()
        m3.reflect(
            config.db, only=lambda name, m: name.lower() in ("t1", "t2")
        )
        assert m3.tables["t2"].c.t1id.references(m3.tables["t1"].c.id)

def test_get_table_names(self):
        tablenames = [
            t
            for t in inspect(config.db).get_table_names()
            if t.lower() in ("t1", "t2")
        ]

eq_(tablenames[0].upper(), tablenames[0].lower())
        eq_(tablenames[1].upper(), tablenames[1].lower())

class ComputedReflectionTest(fixtures.ComputedReflectionFixtureTest):
    def test_computed_col_default_not_set(self):
        insp = inspect(config.db)

cols = insp.get_columns("computed_default_table")
        col_data = {c["name"]: c for c in cols}
        is_true("42" in col_data["with_default"]["default"])
        is_(col_data["normal"]["default"], None)
        is_(col_data["computed_col"]["default"], None)

def test_get_column_returns_computed(self):
        insp = inspect(config.db)

cols = insp.get_columns("computed_default_table")
        data = {c["name"]: c for c in cols}
        for key in ("id", "normal", "with_default"):
            is_true("computed" not in data[key])
        compData = data["computed_col"]
        is_true("computed" in compData)
        is_true("sqltext" in compData["computed"])
        eq_(self.normalize(compData["computed"]["sqltext"]), "normal+42")
        eq_(
            "persisted" in compData["computed"],
            testing.requires.computed_columns_reflect_persisted.enabled,
        )
        if testing.requires.computed_columns_reflect_persisted.enabled:
            eq_(
                compData["computed"]["persisted"],
                testing.requires.computed_columns_default_persisted.enabled,
            )

def check_column(self, data, column, sqltext, persisted):
        is_true("computed" in data[column])
        compData = data[column]["computed"]
        eq_(self.normalize(compData["sqltext"]), sqltext)
        if testing.requires.computed_columns_reflect_persisted.enabled:
            is_true("persisted" in compData)
            is_(compData["persisted"], persisted)

def test_get_column_returns_persisted(self):
        insp = inspect(config.db)

cols = insp.get_columns("computed_column_table")
        data = {c["name"]: c for c in cols}

self.check_column(
            data,
            "computed_no_flag",
            "normal+42",
            testing.requires.computed_columns_default_persisted.enabled,
        )
        if testing.requires.computed_columns_virtual.enabled:
            self.check_column(
                data,
                "computed_virtual",
                "normal+2",
                False,
            )
        if testing.requires.computed_columns_stored.enabled:
            self.check_column(
                data,
                "computed_stored",
                "normal-42",
                True,
            )

@testing.requires.schemas
    def test_get_column_returns_persisted_with_schema(self):
        insp = inspect(config.db)

cols = insp.get_columns(
            "computed_column_table", schema=config.test_schema
        )
        data = {c["name"]: c for c in cols}

self.check_column(
            data,
            "computed_no_flag",
            "normal/42",
            testing.requires.computed_columns_default_persisted.enabled,
        )
        if testing.requires.computed_columns_virtual.enabled:
            self.check_column(
                data,
                "computed_virtual",
                "normal/2",
                False,
            )
        if testing.requires.computed_columns_stored.enabled:
            self.check_column(
                data,
                "computed_stored",
                "normal*42",
                True,
            )

class IdentityReflectionTest(fixtures.TablesTest):
    run_inserts = run_deletes = None

__backend__ = True
    __requires__ = ("identity_columns", "table_reflection")

@classmethod
    def define_tables(cls, metadata):
        Table(
            "t1",
            metadata,
            Column("normal", Integer),
            Column("id1", Integer, Identity()),
        )
        Table(
            "t2",
            metadata,
            Column(
                "id2",
                Integer,
                Identity(
                    always=True,
                    start=2,
                    increment=3,
                    minvalue=-2,
                    maxvalue=42,
                    cycle=True,
                    cache=4,
                ),
            ),
        )
        if testing.requires.schemas.enabled:
            Table(
                "t1",
                metadata,
                Column("normal", Integer),
                Column("id1", Integer, Identity(always=True, start=20)),
                schema=config.test_schema,
            )

def check(self, value, exp, approx):
        if testing.requires.identity_columns_standard.enabled:
            common_keys = (
                "always",
                "start",
                "increment",
                "minvalue",
                "maxvalue",
                "cycle",
                "cache",
            )
            for k in list(value):
                if k not in common_keys:
                    value.pop(k)
            if approx:
                eq_(len(value), len(exp))
                for k in value:
                    if k == "minvalue":
                        is_true(value[k] <= exp[k])
                    elif k in {"maxvalue", "cache"}:
                        is_true(value[k] >= exp[k])
                    else:
                        eq_(value[k], exp[k], k)
            else:
                eq_(value, exp)
        else:
            eq_(value["start"], exp["start"])
            eq_(value["increment"], exp["increment"])

def test_reflect_identity(self):
        insp = inspect(config.db)

cols = insp.get_columns("t1") + insp.get_columns("t2")
        for col in cols:
            if col["name"] == "normal":
                is_false("identity" in col)
            elif col["name"] == "id1":
                is_true(col["autoincrement"] in (True, "auto"))
                eq_(col["default"], None)
                is_true("identity" in col)
                self.check(
                    col["identity"],
                    dict(
                        always=False,
                        start=1,
                        increment=1,
                        minvalue=1,
                        maxvalue=2147483647,
                        cycle=False,
                        cache=1,
                    ),
                    approx=True,
                )
            elif col["name"] == "id2":
                is_true(col["autoincrement"] in (True, "auto"))
                eq_(col["default"], None)
                is_true("identity" in col)
                self.check(
                    col["identity"],
                    dict(
                        always=True,
                        start=2,
                        increment=3,
                        minvalue=-2,
                        maxvalue=42,
                        cycle=True,
                        cache=4,
                    ),
                    approx=False,
                )

@testing.requires.schemas
    def test_reflect_identity_schema(self):
        insp = inspect(config.db)

cols = insp.get_columns("t1", schema=config.test_schema)
        for col in cols:
            if col["name"] == "normal":
                is_false("identity" in col)
            elif col["name"] == "id1":
                is_true(col["autoincrement"] in (True, "auto"))
                eq_(col["default"], None)
                is_true("identity" in col)
                self.check(
                    col["identity"],
                    dict(
                        always=True,
                        start=20,
                        increment=1,
                        minvalue=1,
                        maxvalue=2147483647,
                        cycle=False,
                        cache=1,
                    ),
                    approx=True,
                )

class CompositeKeyReflectionTest(fixtures.TablesTest):
    __backend__ = True

@classmethod
    def define_tables(cls, metadata):
        tb1 = Table(
            "tb1",
            metadata,
            Column("id", Integer),
            Column("attr", Integer),
            Column("name", sql_types.VARCHAR(20)),
            sa.PrimaryKeyConstraint("name", "id", "attr", name="pk_tb1"),
            schema=None,
            test_needs_fk=True,
        )
        Table(
            "tb2",
            metadata,
            Column("id", Integer, primary_key=True),
            Column("pid", Integer),
            Column("pattr", Integer),
            Column("pname", sql_types.VARCHAR(20)),
            sa.ForeignKeyConstraint(
                ["pname", "pid", "pattr"],
                [tb1.c.name, tb1.c.id, tb1.c.attr],
                name="fk_tb1_name_id_attr",
            ),
            schema=None,
            test_needs_fk=True,
        )

@testing.requires.primary_key_constraint_reflection
    def test_pk_column_order(self):
        # test for issue #5661
        insp = inspect(self.bind)
        primary_key = insp.get_pk_constraint(self.tables.tb1.name)
        eq_(primary_key.get("constrained_columns"), ["name", "id", "attr"])

@testing.requires.foreign_key_constraint_reflection
    def test_fk_column_order(self):
        # test for issue #5661
        insp = inspect(self.bind)
        foreign_keys = insp.get_foreign_keys(self.tables.tb2.name)
        eq_(len(foreign_keys), 1)
        fkey1 = foreign_keys[0]
        eq_(fkey1.get("referred_columns"), ["name", "id", "attr"])
        eq_(fkey1.get("constrained_columns"), ["pname", "pid", "pattr"])

__all__ = (
    "ComponentReflectionTest",
    "ComponentReflectionTestExtra",
    "TableNoColumnsTest",
    "QuotedNameArgumentTest",
    "HasTableTest",
    "HasIndexTest",
    "NormalizedNameTest",
    "ComputedReflectionTest",
    "IdentityReflectionTest",
    "CompositeKeyReflectionTest",
)

Veranstaltungen | Fairnetzt-Hochrhein

Keine Veranstaltung gefunden!