PublicArchive/sqlalchemy
2
0
Fork 0
mirror of https://github.com/sqlalchemy/sqlalchemy.git synced 2026-05-23 17:11:39 -04:00
Code Issues Packages Projects Releases Wiki Activity
Files
ba3d4aa5baebacfef8a56de66a5c06cd1372e893
sqlalchemy/test/sql/test_constraints.py
T
Mike Bayer 77db0ef6ac - Fixed bug in enhanced constraint-attachment logic introduced in 
🎫`3341` where in the unusual case of a constraint that refers
to a mixture of :class:`.Column` objects and string column names
at the same time, the auto-attach-on-column-attach logic will be
skipped; for the constraint to be auto-attached in this case,
all columns must be assembled on the target table up front.
Added a new section to the migration document regarding the
original feature as well as this change.
fixes #3411
2015-05-02 10:27:03 -04:00

1495 lines
49 KiB
Python
Raw Blame History

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):
factory = lambda **kw: PrimaryKeyConstraint('a', **kw)
self._test_deferrable(factory)
def test_deferrable_table_fk(self):
factory = lambda **kw: 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):
factory = lambda **kw: UniqueConstraint('b', **kw)
self._test_deferrable(factory)
def test_deferrable_table_check(self):
factory = lambda **kw: 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_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)"
)
class ConstraintAPITest(fixtures.TestBase):
def test_double_fk_usage_raises(self):
f = ForeignKey('b.id')
Column('x', Integer, f)
assert_raises(exc.InvalidRequestError, Column, "y", Integer, f)
def test_auto_append_constraint(self):
m = MetaData()
t = Table('tbl', m,
Column('a', Integer),
Column('b', Integer)
)
t2 = Table('t2', m,
Column('a', Integer),
Column('b', Integer)
)
for c in (
UniqueConstraint(t.c.a),
CheckConstraint(t.c.a > 5),
ForeignKeyConstraint([t.c.a], [t2.c.a]),
PrimaryKeyConstraint(t.c.a)
):
assert c in t.constraints
t.append_constraint(c)
assert c in t.constraints
c = Index('foo', t.c.a)
assert c in t.indexes
def test_auto_append_lowercase_table(self):
t = table('t', column('a'))
t2 = table('t2', column('a'))
for c in (
UniqueConstraint(t.c.a),
CheckConstraint(t.c.a > 5),
ForeignKeyConstraint([t.c.a], [t2.c.a]),
PrimaryKeyConstraint(t.c.a),
Index('foo', t.c.a)
):
assert True
def test_tometadata_ok(self):
m = MetaData()
t = Table('tbl', m,
Column('a', Integer),
Column('b', Integer)
)
t2 = Table('t2', m,
Column('a', Integer),
Column('b', Integer)
)
UniqueConstraint(t.c.a)
CheckConstraint(t.c.a > 5)
ForeignKeyConstraint([t.c.a], [t2.c.a])
PrimaryKeyConstraint(t.c.a)
m2 = MetaData()
t3 = t.tometadata(m2)
eq_(len(t3.constraints), 4)
for c in t3.constraints:
assert c.table is t3
def test_check_constraint_copy(self):
m = MetaData()
t = Table('tbl', m,
Column('a', Integer),
Column('b', Integer)
)
ck = CheckConstraint(t.c.a > 5)
ck2 = ck.copy()
assert ck in t.constraints
assert ck2 not in t.constraints
def test_ambig_check_constraint_auto_append(self):
m = MetaData()
t = Table('tbl', m,
Column('a', Integer),
Column('b', Integer)
)
t2 = Table('t2', m,
Column('a', Integer),
Column('b', Integer)
)
c = CheckConstraint(t.c.a > t2.c.b)
assert c not in t.constraints
assert c not in t2.constraints
def test_auto_append_ck_on_col_attach_one(self):
m = MetaData()
a = Column('a', Integer)
b = Column('b', Integer)
ck = CheckConstraint(a > b)
t = Table('tbl', m, a, b)
assert ck in t.constraints
def test_auto_append_ck_on_col_attach_two(self):
m = MetaData()
a = Column('a', Integer)
b = Column('b', Integer)
c = Column('c', Integer)
ck = CheckConstraint(a > b + c)
t = Table('tbl', m, a)
assert ck not in t.constraints
t.append_column(b)
assert ck not in t.constraints
t.append_column(c)
assert ck in t.constraints
def test_auto_append_ck_on_col_attach_three(self):
m = MetaData()
a = Column('a', Integer)
b = Column('b', Integer)
c = Column('c', Integer)
ck = CheckConstraint(a > b + c)
t = Table('tbl', m, a)
assert ck not in t.constraints
t.append_column(b)
assert ck not in t.constraints
t2 = Table('t2', m)
t2.append_column(c)
# two different tables, so CheckConstraint does nothing.
assert ck not in t.constraints
def test_auto_append_uq_on_col_attach_one(self):
m = MetaData()
a = Column('a', Integer)
b = Column('b', Integer)
uq = UniqueConstraint(a, b)
t = Table('tbl', m, a, b)
assert uq in t.constraints
def test_auto_append_uq_on_col_attach_two(self):
m = MetaData()
a = Column('a', Integer)
b = Column('b', Integer)
c = Column('c', Integer)
uq = UniqueConstraint(a, b, c)
t = Table('tbl', m, a)
assert uq not in t.constraints
t.append_column(b)
assert uq not in t.constraints
t.append_column(c)
assert uq in t.constraints
def test_auto_append_uq_on_col_attach_three(self):
m = MetaData()
a = Column('a', Integer)
b = Column('b', Integer)
c = Column('c', Integer)
uq = UniqueConstraint(a, b, c)
t = Table('tbl', m, a)
assert uq not in t.constraints
t.append_column(b)
assert uq not in t.constraints
t2 = Table('t2', m)
# two different tables, so UniqueConstraint raises
assert_raises_message(
exc.ArgumentError,
r"Column\(s\) 't2\.c' are not part of table 'tbl'\.",
t2.append_column, c
)
def test_auto_append_uq_on_col_attach_four(self):
"""Test that a uniqueconstraint that names Column and string names
won't autoattach using deferred column attachment.
"""
m = MetaData()
a = Column('a', Integer)
b = Column('b', Integer)
c = Column('c', Integer)
uq = UniqueConstraint(a, 'b', 'c')
t = Table('tbl', m, a)
assert uq not in t.constraints
t.append_column(b)
assert uq not in t.constraints
t.append_column(c)
# we don't track events for previously unknown columns
# named 'c' to be attached
assert uq not in t.constraints
t.append_constraint(uq)
assert uq in t.constraints
eq_(
[cn for cn in t.constraints if isinstance(cn, UniqueConstraint)],
[uq]
)
def test_auto_append_uq_on_col_attach_five(self):
"""Test that a uniqueconstraint that names Column and string names
*will* autoattach if the table has all those names up front.
"""
m = MetaData()
a = Column('a', Integer)
b = Column('b', Integer)
c = Column('c', Integer)
t = Table('tbl', m, a, c, b)
uq = UniqueConstraint(a, 'b', 'c')
assert uq in t.constraints
t.append_constraint(uq)
assert uq in t.constraints
eq_(
[cn for cn in t.constraints if isinstance(cn, UniqueConstraint)],
[uq]
)
def test_index_asserts_cols_standalone(self):
metadata = MetaData()
t1 = Table('t1', metadata,
Column('x', Integer)
)
t2 = Table('t2', metadata,
Column('y', Integer)
)
assert_raises_message(
exc.ArgumentError,
r"Column\(s\) 't2.y' are not part of table 't1'.",
Index,
"bar", t1.c.x, t2.c.y
)
def test_index_asserts_cols_inline(self):
metadata = MetaData()
t1 = Table('t1', metadata,
Column('x', Integer)
)
assert_raises_message(
exc.ArgumentError,
"Index 'bar' is against table 't1', and "
"cannot be associated with table 't2'.",
Table, 't2', metadata,
Column('y', Integer),
Index('bar', t1.c.x)
)
def test_raise_index_nonexistent_name(self):
m = MetaData()
# the KeyError isn't ideal here, a nicer message
# perhaps
assert_raises(
KeyError,
Table, 't', m, Column('x', Integer), Index("foo", "q")
)
def test_raise_not_a_column(self):
assert_raises(
exc.ArgumentError,
Index, "foo", 5
)
def test_raise_expr_no_column(self):
idx = Index('foo', func.lower(5))
assert_raises_message(
exc.CompileError,
"Index 'foo' is not associated with any table.",
schema.CreateIndex(idx).compile, dialect=testing.db.dialect
)
assert_raises_message(
exc.CompileError,
"Index 'foo' is not associated with any table.",
schema.CreateIndex(idx).compile
)
def test_no_warning_w_no_columns(self):
idx = Index(name="foo")
assert_raises_message(
exc.CompileError,
"Index 'foo' is not associated with any table.",
schema.CreateIndex(idx).compile, dialect=testing.db.dialect
)
assert_raises_message(
exc.CompileError,
"Index 'foo' is not associated with any table.",
schema.CreateIndex(idx).compile
)
def test_raise_clauseelement_not_a_column(self):
m = MetaData()
t2 = Table('t2', m, Column('x', Integer))
class SomeClass(object):
def __clause_element__(self):
return t2
assert_raises(
exc.ArgumentError,
Index, "foo", SomeClass()
)
Reference in New Issue View Git Blame Copy Permalink