from sqlalchemy.testing import assert_raises, assert_raises_message
from sqlalchemy import Table, Integer, String, Column, PrimaryKeyConstraint,\
    ForeignKeyConstraint, ForeignKey, UniqueConstraint, Index, MetaData, \
    CheckConstraint, func, text
from sqlalchemy import exc, schema
from sqlalchemy.testing import fixtures, AssertsExecutionResults, \
    AssertsCompiledSQL
from sqlalchemy import testing
from sqlalchemy.engine import default
from sqlalchemy.testing import engines
from sqlalchemy.testing.assertions import expect_warnings
from sqlalchemy.testing import eq_
from sqlalchemy.testing.assertsql import (AllOf,
                                          RegexSQL,
                                          CompiledSQL,
                                          DialectSQL)
from sqlalchemy.sql import table, column


class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
    __dialect__ = 'default'
    __backend__ = True

    @testing.provide_metadata
    def test_pk_fk_constraint_create(self):
        metadata = self.metadata

        Table('employees', metadata,
              Column('id', Integer),
              Column('soc', String(40)),
              Column('name', String(30)),
              PrimaryKeyConstraint('id', 'soc')
              )
        Table('elements', metadata,
              Column('id', Integer),
              Column('stuff', String(30)),
              Column('emp_id', Integer),
              Column('emp_soc', String(40)),
              PrimaryKeyConstraint('id', name='elements_primkey'),
              ForeignKeyConstraint(['emp_id', 'emp_soc'],
                                   ['employees.id', 'employees.soc'])
              )
        self.assert_sql_execution(
            testing.db,
            lambda: metadata.create_all(checkfirst=False),
            CompiledSQL('CREATE TABLE employees ('
                        'id INTEGER NOT NULL, '
                        'soc VARCHAR(40) NOT NULL, '
                        'name VARCHAR(30), '
                        'PRIMARY KEY (id, soc)'
                        ')'
                        ),
            CompiledSQL('CREATE TABLE elements ('
                        'id INTEGER NOT NULL, '
                        'stuff VARCHAR(30), '
                        'emp_id INTEGER, '
                        'emp_soc VARCHAR(40), '
                        'CONSTRAINT elements_primkey PRIMARY KEY (id), '
                        'FOREIGN KEY(emp_id, emp_soc) '
                        'REFERENCES employees (id, soc)'
                        ')'
                        )
        )

    @testing.force_drop_names('a', 'b')
    def test_fk_cant_drop_cycled_unnamed(self):
        metadata = MetaData()

        Table("a", metadata,
              Column('id', Integer, primary_key=True),
              Column('bid', Integer),
              ForeignKeyConstraint(["bid"], ["b.id"])
              )
        Table(
            "b", metadata,
            Column('id', Integer, primary_key=True),
            Column("aid", Integer),
            ForeignKeyConstraint(["aid"], ["a.id"]))
        metadata.create_all(testing.db)
        if testing.db.dialect.supports_alter:
            assert_raises_message(
                exc.CircularDependencyError,
                "Can't sort tables for DROP; an unresolvable foreign key "
                "dependency exists between tables: a, b.  Please ensure "
                "that the ForeignKey and ForeignKeyConstraint objects "
                "involved in the cycle have names so that they can be "
                "dropped using DROP CONSTRAINT.",
                metadata.drop_all, testing.db
            )
        else:
            with expect_warnings(
                    "Can't sort tables for DROP; an unresolvable "
                    "foreign key dependency "):
                with self.sql_execution_asserter() as asserter:
                    metadata.drop_all(testing.db, checkfirst=False)

            asserter.assert_(
                AllOf(
                    CompiledSQL("DROP TABLE a"),
                    CompiledSQL("DROP TABLE b")
                )
            )

    @testing.provide_metadata
    def test_fk_table_auto_alter_constraint_create(self):
        metadata = self.metadata

        Table("a", metadata,
              Column('id', Integer, primary_key=True),
              Column('bid', Integer),
              ForeignKeyConstraint(["bid"], ["b.id"])
              )
        Table(
            "b", metadata,
            Column('id', Integer, primary_key=True),
            Column("aid", Integer),
            ForeignKeyConstraint(["aid"], ["a.id"], name="bfk"))
        self._assert_cyclic_constraint(
            metadata, auto=True, sqlite_warning=True)

    @testing.provide_metadata
    def test_fk_column_auto_alter_inline_constraint_create(self):
        metadata = self.metadata

        Table("a", metadata,
              Column('id', Integer, primary_key=True),
              Column('bid', Integer, ForeignKey("b.id")),
              )
        Table("b", metadata,
              Column('id', Integer, primary_key=True),
              Column("aid", Integer,
                     ForeignKey("a.id", name="bfk")
                     ),
              )
        self._assert_cyclic_constraint(
            metadata, auto=True, sqlite_warning=True)

    @testing.provide_metadata
    def test_fk_column_use_alter_inline_constraint_create(self):
        metadata = self.metadata

        Table("a", metadata,
              Column('id', Integer, primary_key=True),
              Column('bid', Integer, ForeignKey("b.id")),
              )
        Table("b", metadata,
              Column('id', Integer, primary_key=True),
              Column("aid", Integer,
                     ForeignKey("a.id", name="bfk", use_alter=True)
                     ),
              )
        self._assert_cyclic_constraint(metadata, auto=False)

    @testing.provide_metadata
    def test_fk_table_use_alter_constraint_create(self):
        metadata = self.metadata

        Table("a", metadata,
              Column('id', Integer, primary_key=True),
              Column('bid', Integer),
              ForeignKeyConstraint(["bid"], ["b.id"])
              )
        Table(
            "b", metadata,
            Column('id', Integer, primary_key=True),
            Column("aid", Integer),
            ForeignKeyConstraint(
                ["aid"], ["a.id"], use_alter=True, name="bfk"))
        self._assert_cyclic_constraint(metadata)

    @testing.provide_metadata
    def test_fk_column_use_alter_constraint_create(self):
        metadata = self.metadata

        Table("a", metadata,
              Column('id', Integer, primary_key=True),
              Column('bid', Integer, ForeignKey("b.id")),
              )
        Table("b", metadata,
              Column('id', Integer, primary_key=True),
              Column("aid", Integer,
                     ForeignKey("a.id", use_alter=True, name="bfk")
                     ),
              )
        self._assert_cyclic_constraint(metadata, auto=False)

    def _assert_cyclic_constraint(
            self, metadata, auto=False, sqlite_warning=False):
        if testing.db.dialect.supports_alter:
            self._assert_cyclic_constraint_supports_alter(metadata, auto=auto)
        else:
            self._assert_cyclic_constraint_no_alter(
                metadata, auto=auto, sqlite_warning=sqlite_warning)

    def _assert_cyclic_constraint_supports_alter(self, metadata, auto=False):
        table_assertions = []
        if auto:
            table_assertions = [
                CompiledSQL('CREATE TABLE b ('
                            'id INTEGER NOT NULL, '
                            'aid INTEGER, '
                            'PRIMARY KEY (id)'
                            ')'
                            ),
                CompiledSQL(
                    'CREATE TABLE a ('
                    'id INTEGER NOT NULL, '
                    'bid INTEGER, '
                    'PRIMARY KEY (id)'
                    ')'
                )
            ]
        else:
            table_assertions = [
                CompiledSQL('CREATE TABLE b ('
                            'id INTEGER NOT NULL, '
                            'aid INTEGER, '
                            'PRIMARY KEY (id)'
                            ')'
                            ),
                CompiledSQL(
                    'CREATE TABLE a ('
                    'id INTEGER NOT NULL, '
                    'bid INTEGER, '
                    'PRIMARY KEY (id), '
                    'FOREIGN KEY(bid) REFERENCES b (id)'
                    ')'
                )
            ]

        assertions = [AllOf(*table_assertions)]
        fk_assertions = []
        fk_assertions.append(
            CompiledSQL('ALTER TABLE b ADD CONSTRAINT bfk '
                        'FOREIGN KEY(aid) REFERENCES a (id)')
        )
        if auto:
            fk_assertions.append(
                CompiledSQL('ALTER TABLE a ADD '
                            'FOREIGN KEY(bid) REFERENCES b (id)')
            )
        assertions.append(AllOf(*fk_assertions))

        with self.sql_execution_asserter() as asserter:
            metadata.create_all(checkfirst=False)
        asserter.assert_(*assertions)

        assertions = [
            CompiledSQL('ALTER TABLE b DROP CONSTRAINT bfk'),
            CompiledSQL("DROP TABLE a"),
            CompiledSQL("DROP TABLE b")
        ]

        with self.sql_execution_asserter() as asserter:
            metadata.drop_all(checkfirst=False),
        asserter.assert_(*assertions)

    def _assert_cyclic_constraint_no_alter(
            self, metadata, auto=False, sqlite_warning=False):
        table_assertions = []
        if auto:
            table_assertions.append(
                DialectSQL(
                    'CREATE TABLE b ('
                    'id INTEGER NOT NULL, '
                    'aid INTEGER, '
                    'PRIMARY KEY (id), '
                    'CONSTRAINT bfk FOREIGN KEY(aid) REFERENCES a (id)'
                    ')'
                )
            )
            table_assertions.append(
                DialectSQL(
                    'CREATE TABLE a ('
                    'id INTEGER NOT NULL, '
                    'bid INTEGER, '
                    'PRIMARY KEY (id), '
                    'FOREIGN KEY(bid) REFERENCES b (id)'
                    ')'
                )
            )
        else:
            table_assertions.append(
                DialectSQL(
                    'CREATE TABLE b ('
                    'id INTEGER NOT NULL, '
                    'aid INTEGER, '
                    'PRIMARY KEY (id), '
                    'CONSTRAINT bfk FOREIGN KEY(aid) REFERENCES a (id)'
                    ')'
                )
            )

            table_assertions.append(
                DialectSQL(
                    'CREATE TABLE a ('
                    'id INTEGER NOT NULL, '
                    'bid INTEGER, '
                    'PRIMARY KEY (id), '
                    'FOREIGN KEY(bid) REFERENCES b (id)'
                    ')'
                )
            )

        assertions = [AllOf(*table_assertions)]

        with self.sql_execution_asserter() as asserter:
            metadata.create_all(checkfirst=False)
        asserter.assert_(*assertions)

        assertions = [AllOf(
            CompiledSQL("DROP TABLE a"),
            CompiledSQL("DROP TABLE b")
        )]

        if sqlite_warning:
            with expect_warnings("Can't sort tables for DROP; "):
                with self.sql_execution_asserter() as asserter:
                    metadata.drop_all(checkfirst=False),
        else:
            with self.sql_execution_asserter() as asserter:
                metadata.drop_all(checkfirst=False),
        asserter.assert_(*assertions)

    @testing.force_drop_names("a", "b")
    def test_cycle_unnamed_fks(self):
        metadata = MetaData(testing.db)

        Table("a", metadata,
              Column('id', Integer, primary_key=True),
              Column('bid', Integer, ForeignKey("b.id")),
              )

        Table("b", metadata,
              Column('id', Integer, primary_key=True),
              Column("aid", Integer, ForeignKey("a.id")),
              )

        assertions = [
            AllOf(
                CompiledSQL(
                    'CREATE TABLE b ('
                    'id INTEGER NOT NULL, '
                    'aid INTEGER, '
                    'PRIMARY KEY (id)'
                    ')'
                ),
                CompiledSQL(
                    'CREATE TABLE a ('
                    'id INTEGER NOT NULL, '
                    'bid INTEGER, '
                    'PRIMARY KEY (id)'
                    ')'
                )
            ),
            AllOf(
                CompiledSQL('ALTER TABLE b ADD '
                            'FOREIGN KEY(aid) REFERENCES a (id)'),
                CompiledSQL('ALTER TABLE a ADD '
                            'FOREIGN KEY(bid) REFERENCES b (id)')
            ),
        ]
        with self.sql_execution_asserter() as asserter:
            metadata.create_all(checkfirst=False)

        if testing.db.dialect.supports_alter:
            asserter.assert_(*assertions)

            assert_raises_message(
                exc.CircularDependencyError,
                "Can't sort tables for DROP; an unresolvable foreign key "
                "dependency exists between tables: a, b.  "
                "Please ensure that the "
                "ForeignKey and ForeignKeyConstraint objects involved in the "
                "cycle have names so that they can be dropped using "
                "DROP CONSTRAINT.",
                metadata.drop_all, checkfirst=False
            )
        else:
            with expect_warnings(
                    "Can't sort tables for DROP; an unresolvable "
                    "foreign key dependency exists between tables"):
                with self.sql_execution_asserter() as asserter:
                    metadata.drop_all(checkfirst=False)

            asserter.assert_(
                AllOf(
                    CompiledSQL("DROP TABLE b"),
                    CompiledSQL("DROP TABLE a"),
                )
            )

    @testing.force_drop_names("a", "b")
    def test_cycle_named_fks(self):
        metadata = MetaData(testing.db)

        Table("a", metadata,
              Column('id', Integer, primary_key=True),
              Column('bid', Integer, ForeignKey("b.id")),
              )

        Table("b", metadata,
              Column('id', Integer, primary_key=True),
              Column(
                  "aid", Integer,
                  ForeignKey("a.id", use_alter=True, name='aidfk')),
              )

        assertions = [
            AllOf(
                CompiledSQL(
                    'CREATE TABLE b ('
                    'id INTEGER NOT NULL, '
                    'aid INTEGER, '
                    'PRIMARY KEY (id)'
                    ')'
                ),
                CompiledSQL(
                    'CREATE TABLE a ('
                    'id INTEGER NOT NULL, '
                    'bid INTEGER, '
                    'PRIMARY KEY (id), '
                    'FOREIGN KEY(bid) REFERENCES b (id)'
                    ')'
                )
            ),
            CompiledSQL('ALTER TABLE b ADD CONSTRAINT aidfk '
                        'FOREIGN KEY(aid) REFERENCES a (id)'),
        ]
        with self.sql_execution_asserter() as asserter:
            metadata.create_all(checkfirst=False)

        if testing.db.dialect.supports_alter:
            asserter.assert_(*assertions)

            with self.sql_execution_asserter() as asserter:
                metadata.drop_all(checkfirst=False)

            asserter.assert_(
                CompiledSQL("ALTER TABLE b DROP CONSTRAINT aidfk"),
                AllOf(
                    CompiledSQL("DROP TABLE b"),
                    CompiledSQL("DROP TABLE a"),
                )
            )
        else:
            with self.sql_execution_asserter() as asserter:
                metadata.drop_all(checkfirst=False)

            asserter.assert_(
                AllOf(
                    CompiledSQL("DROP TABLE b"),
                    CompiledSQL("DROP TABLE a"),
                ),
            )

    @testing.requires.check_constraints
    @testing.provide_metadata
    def test_check_constraint_create(self):
        metadata = self.metadata

        Table('foo', metadata,
              Column('id', Integer, primary_key=True),
              Column('x', Integer),
              Column('y', Integer),
              CheckConstraint('x>y'))
        Table('bar', metadata,
              Column('id', Integer, primary_key=True),
              Column('x', Integer, CheckConstraint('x>7')),
              Column('z', Integer)
              )

        self.assert_sql_execution(
            testing.db,
            lambda: metadata.create_all(checkfirst=False),
            AllOf(
                CompiledSQL('CREATE TABLE foo ('
                            'id INTEGER NOT NULL, '
                            'x INTEGER, '
                            'y INTEGER, '
                            'PRIMARY KEY (id), '
                            'CHECK (x>y)'
                            ')'
                            ),
                CompiledSQL('CREATE TABLE bar ('
                            'id INTEGER NOT NULL, '
                            'x INTEGER CHECK (x>7), '
                            'z INTEGER, '
                            'PRIMARY KEY (id)'
                            ')'
                            )
            )
        )

    @testing.provide_metadata
    def test_unique_constraint_create(self):
        metadata = self.metadata

        Table('foo', metadata,
              Column('id', Integer, primary_key=True),
              Column('value', String(30), unique=True))
        Table('bar', metadata,
              Column('id', Integer, primary_key=True),
              Column('value', String(30)),
              Column('value2', String(30)),
              UniqueConstraint('value', 'value2', name='uix1')
              )

        self.assert_sql_execution(
            testing.db,
            lambda: metadata.create_all(checkfirst=False),
            AllOf(
                CompiledSQL('CREATE TABLE foo ('
                            'id INTEGER NOT NULL, '
                            'value VARCHAR(30), '
                            'PRIMARY KEY (id), '
                            'UNIQUE (value)'
                            ')'),
                CompiledSQL('CREATE TABLE bar ('
                            'id INTEGER NOT NULL, '
                            'value VARCHAR(30), '
                            'value2 VARCHAR(30), '
                            'PRIMARY KEY (id), '
                            'CONSTRAINT uix1 UNIQUE (value, value2)'
                            ')')
            )
        )

    @testing.provide_metadata
    def test_index_create(self):
        metadata = self.metadata

        employees = Table('employees', metadata,
                          Column('id', Integer, primary_key=True),
                          Column('first_name', String(30)),
                          Column('last_name', String(30)),
                          Column('email_address', String(30)))

        i = Index('employee_name_index',
                  employees.c.last_name, employees.c.first_name)
        assert i in employees.indexes

        i2 = Index('employee_email_index',
                   employees.c.email_address, unique=True)
        assert i2 in employees.indexes

        self.assert_sql_execution(
            testing.db,
            lambda: metadata.create_all(checkfirst=False),
            RegexSQL("^CREATE TABLE"),
            AllOf(
                CompiledSQL('CREATE INDEX employee_name_index ON '
                            'employees (last_name, first_name)', []),
                CompiledSQL('CREATE UNIQUE INDEX employee_email_index ON '
                            'employees (email_address)', [])
            )
        )

    @testing.provide_metadata
    def test_index_create_camelcase(self):
        """test that mixed-case index identifiers are legal"""

        metadata = self.metadata

        employees = Table('companyEmployees', metadata,
                          Column('id', Integer, primary_key=True),
                          Column('firstName', String(30)),
                          Column('lastName', String(30)),
                          Column('emailAddress', String(30)))

        Index('employeeNameIndex',
              employees.c.lastName, employees.c.firstName)

        Index('employeeEmailIndex',
              employees.c.emailAddress, unique=True)

        self.assert_sql_execution(
            testing.db, lambda: metadata.create_all(
                checkfirst=False), RegexSQL("^CREATE TABLE"), AllOf(
                CompiledSQL(
                    'CREATE INDEX "employeeNameIndex" ON '
                    '"companyEmployees" ("lastName", "firstName")', []),
                CompiledSQL(
                    'CREATE UNIQUE INDEX "employeeEmailIndex" ON '
                    '"companyEmployees" ("emailAddress")', [])))

    @testing.provide_metadata
    def test_index_create_inline(self):
        # test an index create using index=True, unique=True

        metadata = self.metadata

        events = Table('events', metadata,
                       Column('id', Integer, primary_key=True),
                       Column('name', String(30), index=True, unique=True),
                       Column('location', String(30), index=True),
                       Column('sport', String(30)),
                       Column('announcer', String(30)),
                       Column('winner', String(30)))

        Index('sport_announcer', events.c.sport, events.c.announcer,
              unique=True)
        Index('idx_winners', events.c.winner)

        eq_(
            set(ix.name for ix in events.indexes),
            set(['ix_events_name', 'ix_events_location',
                 'sport_announcer', 'idx_winners'])
        )

        self.assert_sql_execution(
            testing.db,
            lambda: events.create(testing.db),
            RegexSQL("^CREATE TABLE events"),
            AllOf(
                CompiledSQL('CREATE UNIQUE INDEX ix_events_name ON events '
                            '(name)'),
                CompiledSQL('CREATE INDEX ix_events_location ON events '
                            '(location)'),
                CompiledSQL('CREATE UNIQUE INDEX sport_announcer ON events '
                            '(sport, announcer)'),
                CompiledSQL('CREATE INDEX idx_winners ON events (winner)'),
            )
        )

    @testing.provide_metadata
    def test_index_functional_create(self):
        metadata = self.metadata

        t = Table('sometable', metadata,
                  Column('id', Integer, primary_key=True),
                  Column('data', String(50))
                  )
        Index('myindex', t.c.data.desc())
        self.assert_sql_execution(
            testing.db,
            lambda: t.create(testing.db),
            CompiledSQL('CREATE TABLE sometable (id INTEGER NOT NULL, '
                        'data VARCHAR(50), PRIMARY KEY (id))'),
            CompiledSQL('CREATE INDEX myindex ON sometable (data DESC)')
        )


class ConstraintCompilationTest(fixtures.TestBase, AssertsCompiledSQL):
    __dialect__ = 'default'

    def test_create_index_plain(self):
        t = Table('t', MetaData(), Column('x', Integer))
        i = Index("xyz", t.c.x)
        self.assert_compile(
            schema.CreateIndex(i),
            "CREATE INDEX xyz ON t (x)"
        )

    def test_drop_index_plain_unattached(self):
        self.assert_compile(
            schema.DropIndex(Index(name="xyz")),
            "DROP INDEX xyz"
        )

    def test_drop_index_plain(self):
        self.assert_compile(
            schema.DropIndex(Index(name="xyz")),
            "DROP INDEX xyz"
        )

    def test_create_index_schema(self):
        t = Table('t', MetaData(), Column('x', Integer), schema="foo")
        i = Index("xyz", t.c.x)
        self.assert_compile(
            schema.CreateIndex(i),
            "CREATE INDEX xyz ON foo.t (x)"
        )

    def test_drop_index_schema(self):
        t = Table('t', MetaData(), Column('x', Integer), schema="foo")
        i = Index("xyz", t.c.x)
        self.assert_compile(
            schema.DropIndex(i),
            "DROP INDEX foo.xyz"
        )

    def test_too_long_index_name(self):
        dialect = testing.db.dialect.__class__()

        for max_ident, max_index in [(22, None), (256, 22)]:
            dialect.max_identifier_length = max_ident
            dialect.max_index_name_length = max_index

            for tname, cname, exp in [
                ('sometable', 'this_name_is_too_long', 'ix_sometable_t_09aa'),
                ('sometable', 'this_name_alsois_long', 'ix_sometable_t_3cf1'),
            ]:

                t1 = Table(tname, MetaData(),
                           Column(cname, Integer, index=True),
                           )
                ix1 = list(t1.indexes)[0]

                self.assert_compile(
                    schema.CreateIndex(ix1),
                    "CREATE INDEX %s "
                    "ON %s (%s)" % (exp, tname, cname),
                    dialect=dialect
                )

        dialect.max_identifier_length = 22
        dialect.max_index_name_length = None

        t1 = Table('t', MetaData(), Column('c', Integer))
        assert_raises(
            exc.IdentifierError,
            schema.CreateIndex(Index(
                "this_other_name_is_too_long_for_what_were_doing",
                t1.c.c)).compile,
            dialect=dialect
        )

    def test_functional_index(self):
        metadata = MetaData()
        x = Table('x', metadata,
                  Column('q', String(50))
                  )
        idx = Index('y', func.lower(x.c.q))

        self.assert_compile(
            schema.CreateIndex(idx),
            "CREATE INDEX y ON x (lower(q))"
        )

        self.assert_compile(
            schema.CreateIndex(idx),
            "CREATE INDEX y ON x (lower(q))",
            dialect=testing.db.dialect
        )

    def test_index_against_text_separate(self):
        metadata = MetaData()
        idx = Index('y', text("some_function(q)"))
        t = Table('x', metadata,
                  Column('q', String(50))
                  )
        t.append_constraint(idx)
        self.assert_compile(
            schema.CreateIndex(idx),
            "CREATE INDEX y ON x (some_function(q))"
        )

    def test_index_against_text_inline(self):
        metadata = MetaData()
        idx = Index('y', text("some_function(q)"))
        x = Table('x', metadata,
                  Column('q', String(50)),
                  idx
                  )

        self.assert_compile(
            schema.CreateIndex(idx),
            "CREATE INDEX y ON x (some_function(q))"
        )

    def test_index_declaration_inline(self):
        metadata = MetaData()

        t1 = Table('t1', metadata,
                   Column('x', Integer),
                   Column('y', Integer),
                   Index('foo', 'x', 'y')
                   )
        self.assert_compile(
            schema.CreateIndex(list(t1.indexes)[0]),
            "CREATE INDEX foo ON t1 (x, y)"
        )

    def _test_deferrable(self, constraint_factory):
        dialect = default.DefaultDialect()

        t = Table('tbl', MetaData(),
                  Column('a', Integer),
                  Column('b', Integer),
                  constraint_factory(deferrable=True))

        sql = str(schema.CreateTable(t).compile(dialect=dialect))
        assert 'DEFERRABLE' in sql, sql
        assert 'NOT DEFERRABLE' not in sql, sql

        t = Table('tbl', MetaData(),
                  Column('a', Integer),
                  Column('b', Integer),
                  constraint_factory(deferrable=False))

        sql = str(schema.CreateTable(t).compile(dialect=dialect))
        assert 'NOT DEFERRABLE' in sql

        t = Table('tbl', MetaData(),
                  Column('a', Integer),
                  Column('b', Integer),
                  constraint_factory(deferrable=True, initially='IMMEDIATE'))
        sql = str(schema.CreateTable(t).compile(dialect=dialect))
        assert 'NOT DEFERRABLE' not in sql
        assert 'INITIALLY IMMEDIATE' in sql

        t = Table('tbl', MetaData(),
                  Column('a', Integer),
                  Column('b', Integer),
                  constraint_factory(deferrable=True, initially='DEFERRED'))
        sql = str(schema.CreateTable(t).compile(dialect=dialect))

        assert 'NOT DEFERRABLE' not in sql
        assert 'INITIALLY DEFERRED' in sql

    def test_column_level_ck_name(self):
        t = Table(
            'tbl',
            MetaData(),
            Column(
                'a',
                Integer,
                CheckConstraint(
                    "a > 5",
                    name="ck_a_greater_five")))
        self.assert_compile(
            schema.CreateTable(t),
            "CREATE TABLE tbl (a INTEGER CONSTRAINT "
            "ck_a_greater_five CHECK (a > 5))"
        )

    def test_deferrable_pk(self):
        def factory(**kw): return PrimaryKeyConstraint('a', **kw)
        self._test_deferrable(factory)

    def test_deferrable_table_fk(self):
        def factory(**kw): return ForeignKeyConstraint(['b'], ['tbl.a'], **kw)
        self._test_deferrable(factory)

    def test_deferrable_column_fk(self):
        t = Table('tbl', MetaData(),
                  Column('a', Integer),
                  Column('b', Integer,
                         ForeignKey('tbl.a', deferrable=True,
                                    initially='DEFERRED')))

        self.assert_compile(
            schema.CreateTable(t),
            "CREATE TABLE tbl (a INTEGER, b INTEGER, "
            "FOREIGN KEY(b) REFERENCES tbl "
            "(a) DEFERRABLE INITIALLY DEFERRED)",
        )

    def test_fk_match_clause(self):
        t = Table('tbl', MetaData(),
                  Column('a', Integer),
                  Column('b', Integer,
                         ForeignKey('tbl.a', match="SIMPLE")))

        self.assert_compile(
            schema.CreateTable(t),
            "CREATE TABLE tbl (a INTEGER, b INTEGER, "
            "FOREIGN KEY(b) REFERENCES tbl "
            "(a) MATCH SIMPLE)",
        )

        self.assert_compile(
            schema.AddConstraint(list(t.foreign_keys)[0].constraint),
            "ALTER TABLE tbl ADD FOREIGN KEY(b) "
            "REFERENCES tbl (a) MATCH SIMPLE"
        )

    def test_create_table_omit_fks(self):
        fkcs = [
            ForeignKeyConstraint(['a'], ['remote.id'], name='foo'),
            ForeignKeyConstraint(['b'], ['remote.id'], name='bar'),
            ForeignKeyConstraint(['c'], ['remote.id'], name='bat'),
        ]
        m = MetaData()
        t = Table(
            't', m,
            Column('a', Integer),
            Column('b', Integer),
            Column('c', Integer),
            *fkcs
        )
        Table('remote', m, Column('id', Integer, primary_key=True))

        self.assert_compile(
            schema.CreateTable(t, include_foreign_key_constraints=[]),
            "CREATE TABLE t (a INTEGER, b INTEGER, c INTEGER)"
        )
        self.assert_compile(
            schema.CreateTable(t, include_foreign_key_constraints=fkcs[0:2]),
            "CREATE TABLE t (a INTEGER, b INTEGER, c INTEGER, "
            "CONSTRAINT foo FOREIGN KEY(a) REFERENCES remote (id), "
            "CONSTRAINT bar FOREIGN KEY(b) REFERENCES remote (id))"
        )

    def test_deferrable_unique(self):
        def factory(**kw): return UniqueConstraint('b', **kw)
        self._test_deferrable(factory)

    def test_deferrable_table_check(self):
        def factory(**kw): return CheckConstraint('a < b', **kw)
        self._test_deferrable(factory)

    def test_multiple(self):
        m = MetaData()
        Table("foo", m,
              Column('id', Integer, primary_key=True),
              Column('bar', Integer, primary_key=True)
              )
        tb = Table("some_table", m,
                   Column('id', Integer, primary_key=True),
                   Column('foo_id', Integer, ForeignKey('foo.id')),
                   Column('foo_bar', Integer, ForeignKey('foo.bar')),
                   )
        self.assert_compile(
            schema.CreateTable(tb),
            "CREATE TABLE some_table ("
            "id INTEGER NOT NULL, "
            "foo_id INTEGER, "
            "foo_bar INTEGER, "
            "PRIMARY KEY (id), "
            "FOREIGN KEY(foo_id) REFERENCES foo (id), "
            "FOREIGN KEY(foo_bar) REFERENCES foo (bar))"
        )

    def test_empty_pkc(self):
        # test that an empty primary key is ignored
        metadata = MetaData()
        tbl = Table('test', metadata,
                    Column('x', Integer, autoincrement=False),
                    Column('y', Integer, autoincrement=False),
                    PrimaryKeyConstraint())
        self.assert_compile(schema.CreateTable(tbl),
                            "CREATE TABLE test (x INTEGER, y INTEGER)"
                            )

    def test_empty_uc(self):
        # test that an empty constraint is ignored
        metadata = MetaData()
        tbl = Table('test', metadata,
                    Column('x', Integer, autoincrement=False),
                    Column('y', Integer, autoincrement=False),
                    UniqueConstraint())
        self.assert_compile(schema.CreateTable(tbl),
                            "CREATE TABLE test (x INTEGER, y INTEGER)"
                            )

    def test_deferrable_column_check(self):
        t = Table('tbl', MetaData(),
                  Column('a', Integer),
                  Column('b', Integer,
                         CheckConstraint('a < b',
                                         deferrable=True,
                                         initially='DEFERRED')))

        self.assert_compile(
            schema.CreateTable(t),
            "CREATE TABLE tbl (a INTEGER, b INTEGER CHECK (a < b) "
            "DEFERRABLE INITIALLY DEFERRED)"
        )

    def test_use_alter(self):
        m = MetaData()
        Table('t', m,
              Column('a', Integer),
              )

        Table('t2', m,
              Column('a', Integer, ForeignKey('t.a', use_alter=True,
                                              name='fk_ta')),
              Column('b', Integer, ForeignKey('t.a', name='fk_tb'))
              )

        e = engines.mock_engine(dialect_name='postgresql')
        m.create_all(e)
        m.drop_all(e)

        e.assert_sql([
            'CREATE TABLE t (a INTEGER)',
            'CREATE TABLE t2 (a INTEGER, b INTEGER, CONSTRAINT fk_tb '
            'FOREIGN KEY(b) REFERENCES t (a))',
            'ALTER TABLE t2 '
            'ADD CONSTRAINT fk_ta FOREIGN KEY(a) REFERENCES t (a)',
            'ALTER TABLE t2 DROP CONSTRAINT fk_ta',
            'DROP TABLE t2',
            'DROP TABLE t'
        ])

    def _constraint_create_fixture(self):
        m = MetaData()

        t = Table('tbl', m,
                  Column('a', Integer),
                  Column('b', Integer)
                  )

        t2 = Table('t2', m,
                   Column('a', Integer),
                   Column('b', Integer)
                   )

        return t, t2

    def test_render_ck_constraint_inline(self):
        t, t2 = self._constraint_create_fixture()

        CheckConstraint('a < b', name="my_test_constraint",
                        deferrable=True, initially='DEFERRED',
                        table=t)

        # before we create an AddConstraint,
        # the CONSTRAINT comes out inline
        self.assert_compile(
            schema.CreateTable(t),
            "CREATE TABLE tbl ("
            "a INTEGER, "
            "b INTEGER, "
            "CONSTRAINT my_test_constraint CHECK (a < b) "
            "DEFERRABLE INITIALLY DEFERRED"
            ")"
        )

    def test_render_ck_constraint_external(self):
        t, t2 = self._constraint_create_fixture()

        constraint = CheckConstraint('a < b', name="my_test_constraint",
                                     deferrable=True, initially='DEFERRED',
                                     table=t)

        self.assert_compile(
            schema.AddConstraint(constraint),
            "ALTER TABLE tbl ADD CONSTRAINT my_test_constraint "
            "CHECK (a < b) DEFERRABLE INITIALLY DEFERRED"
        )

    def test_external_ck_constraint_cancels_internal(self):
        t, t2 = self._constraint_create_fixture()

        constraint = CheckConstraint('a < b', name="my_test_constraint",
                                     deferrable=True, initially='DEFERRED',
                                     table=t)

        schema.AddConstraint(constraint)

        # once we make an AddConstraint,
        # inline compilation of the CONSTRAINT
        # is disabled
        self.assert_compile(
            schema.CreateTable(t),
            "CREATE TABLE tbl ("
            "a INTEGER, "
            "b INTEGER"
            ")"
        )

    def test_render_drop_constraint(self):
        t, t2 = self._constraint_create_fixture()

        constraint = CheckConstraint('a < b', name="my_test_constraint",
                                     deferrable=True, initially='DEFERRED',
                                     table=t)

        self.assert_compile(
            schema.DropConstraint(constraint),
            "ALTER TABLE tbl DROP CONSTRAINT my_test_constraint"
        )

    def test_render_drop_constraint_cascade(self):
        t, t2 = self._constraint_create_fixture()

        constraint = CheckConstraint('a < b', name="my_test_constraint",
                                     deferrable=True, initially='DEFERRED',
                                     table=t)

        self.assert_compile(
            schema.DropConstraint(constraint, cascade=True),
            "ALTER TABLE tbl DROP CONSTRAINT my_test_constraint CASCADE"
        )

    def test_render_add_fk_constraint_stringcol(self):
        t, t2 = self._constraint_create_fixture()

        constraint = ForeignKeyConstraint(["b"], ["t2.a"])
        t.append_constraint(constraint)
        self.assert_compile(
            schema.AddConstraint(constraint),
            "ALTER TABLE tbl ADD FOREIGN KEY(b) REFERENCES t2 (a)"
        )

    def test_render_add_fk_constraint_realcol(self):
        t, t2 = self._constraint_create_fixture()

        constraint = ForeignKeyConstraint([t.c.a], [t2.c.b])
        t.append_constraint(constraint)
        self.assert_compile(
            schema.AddConstraint(constraint),
            "ALTER TABLE tbl ADD FOREIGN KEY(a) REFERENCES t2 (b)"
        )

    def test_render_add_uq_constraint_stringcol(self):
        t, t2 = self._constraint_create_fixture()

        constraint = UniqueConstraint("a", "b", name="uq_cst")
        t2.append_constraint(constraint)
        self.assert_compile(
            schema.AddConstraint(constraint),
            "ALTER TABLE t2 ADD CONSTRAINT uq_cst UNIQUE (a, b)"
        )

    def test_render_add_uq_constraint_realcol(self):
        t, t2 = self._constraint_create_fixture()

        constraint = UniqueConstraint(t2.c.a, t2.c.b, name="uq_cs2")
        self.assert_compile(
            schema.AddConstraint(constraint),
            "ALTER TABLE t2 ADD CONSTRAINT uq_cs2 UNIQUE (a, b)"
        )

    def test_render_add_pk_constraint(self):
        t, t2 = self._constraint_create_fixture()

        assert t.c.a.primary_key is False
        constraint = PrimaryKeyConstraint(t.c.a)
        assert t.c.a.primary_key is True
        self.assert_compile(
            schema.AddConstraint(constraint),
            "ALTER TABLE tbl ADD PRIMARY KEY (a)"
        )

    def test_render_check_constraint_sql_literal(self):
        t, t2 = self._constraint_create_fixture()

        constraint = CheckConstraint(t.c.a > 5)

        self.assert_compile(
            schema.AddConstraint(constraint),
            "ALTER TABLE tbl ADD CHECK (a > 5)"
        )

    def test_render_check_constraint_inline_sql_literal(self):
        t, t2 = self._constraint_create_fixture()

        m = MetaData()
        t = Table(
            't', m,
            Column('a', Integer, CheckConstraint(Column('a', Integer) > 5)))

        self.assert_compile(
            schema.CreateColumn(t.c.a),
            "a INTEGER CHECK (a > 5)"
        )

    def test_render_index_sql_literal(self):
        t, t2 = self._constraint_create_fixture()

        constraint = Index('name', t.c.a + 5)

        self.assert_compile(
            schema.CreateIndex(constraint),
            "CREATE INDEX name ON tbl (a + 5)"
        )