from sqlalchemy.test.testing import assert_raises,assert_raises_message
from sqlalchemy import *
from sqlalchemy import exc,schema
from sqlalchemy.test import *
from sqlalchemy.test import config,engines
from sqlalchemy.engine import ddl
from sqlalchemy.test.testing import eq_
from sqlalchemy.test.assertsql import AllOf,RegexSQL,ExactSQL,CompiledSQL
from sqlalchemy.dialects.postgresql import base
class ConstraintTest(TestBase, AssertsExecutionResults, AssertsCompiledSQL):
def setup(self):
global metadata
metadata = MetaData(testing.db)
def teardown(self):
metadata.drop_all()
def test_constraint(self):
employees = Table('employees', metadata,
Column('id', Integer),
Column('soc', String(40)),
Column('name', String(30)),
PrimaryKeyConstraint('id', 'soc')
)
elements = 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'])
)
metadata.create_all()
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_circular_constraint(self):
a = Table("a", metadata,
Column('id', Integer, primary_key=True),
Column('bid', Integer),
ForeignKeyConstraint(["bid"], ["b.id"], name="afk")
)
b = Table("b", metadata,
Column('id', Integer, primary_key=True),
Column("aid", Integer),
ForeignKeyConstraint(["aid"], ["a.id"], use_alter=True, name="bfk")
)
metadata.create_all()
def test_circular_constraint_2(self):
a = Table("a", metadata,
Column('id', Integer, primary_key=True),
Column('bid', Integer, ForeignKey("b.id")),
)
b = Table("b", metadata,
Column('id', Integer, primary_key=True),
Column("aid", Integer, ForeignKey("a.id", use_alter=True, name="bfk")),
)
metadata.create_all()
@testing.fails_on('mysql', 'FIXME: unknown')
def test_check_constraint(self):
foo = Table('foo', metadata,
Column('id', Integer, primary_key=True),
Column('x', Integer),
Column('y', Integer),
CheckConstraint('x>y'))
bar = Table('bar', metadata,
Column('id', Integer, primary_key=True),
Column('x', Integer, CheckConstraint('x>7')),
Column('z', Integer)
)
metadata.create_all()
foo.insert().execute(id=1,x=9,y=5)
assert_raises(exc.SQLError, foo.insert().execute, id=2,x=5,y=9)
bar.insert().execute(id=1,x=10)
assert_raises(exc.SQLError, bar.insert().execute, id=2,x=5)
def test_unique_constraint(self):
foo = Table('foo', metadata,
Column('id', Integer, primary_key=True),
Column('value', String(30), unique=True))
bar = Table('bar', metadata,
Column('id', Integer, primary_key=True),
Column('value', String(30)),
Column('value2', String(30)),
UniqueConstraint('value', 'value2', name='uix1')
)
metadata.create_all()
foo.insert().execute(id=1, value='value1')
foo.insert().execute(id=2, value='value2')
bar.insert().execute(id=1, value='a', value2='a')
bar.insert().execute(id=2, value='a', value2='b')
assert_raises(exc.SQLError, foo.insert().execute, id=3, value='value1')
assert_raises(exc.SQLError, bar.insert().execute, id=3, value='a', value2='b')
def test_index_create(self):
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)))
employees.create()
i = Index('employee_name_index',
employees.c.last_name, employees.c.first_name)
i.create()
assert i in employees.indexes
i2 = Index('employee_email_index',
employees.c.email_address, unique=True)
i2.create()
assert i2 in employees.indexes
def test_index_create_camelcase(self):
"""test that mixed-case index identifiers are legal"""
employees = Table('companyEmployees', metadata,
Column('id', Integer, primary_key=True),
Column('firstName', String(30)),
Column('lastName', String(30)),
Column('emailAddress', String(30)))
employees.create()
i = Index('employeeNameIndex',
employees.c.lastName, employees.c.firstName)
i.create()
i = Index('employeeEmailIndex',
employees.c.emailAddress, unique=True)
i.create()
# Check that the table is useable. This is mostly for pg,
# which can be somewhat sticky with mixed-case identifiers
employees.insert().execute(firstName='Joe', lastName='Smith', id=0)
ss = employees.select().execute().fetchall()
assert ss[0].firstName == 'Joe'
assert ss[0].lastName == 'Smith'
def test_index_create_inline(self):
"""Test indexes defined with tables"""
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(
ExactSQL('CREATE UNIQUE INDEX ix_events_name ON events (name)'),
ExactSQL('CREATE INDEX ix_events_location ON events (location)'),
ExactSQL('CREATE UNIQUE INDEX sport_announcer ON events (sport, announcer)'),
ExactSQL('CREATE INDEX idx_winners ON events (winner)')
)
)
# verify that the table is functional
events.insert().execute(id=1, name='hockey finals', location='rink',
sport='hockey', announcer='some canadian',
winner='sweden')
ss = events.select().execute().fetchall()
def test_too_long_idx_name(self):
dialect = testing.db.dialect.__class__()
dialect.max_identifier_length = 20
t1 = Table("sometable", MetaData(), Column("foo", Integer))
self.assert_compile(
schema.CreateIndex(Index("this_name_is_too_long_for_what_were_doing", t1.c.foo)),
"CREATE INDEX this_name_is_t_1 ON sometable (foo)",
dialect=dialect
)
self.assert_compile(
schema.CreateIndex(Index("this_other_name_is_too_long_for_what_were_doing", t1.c.foo)),
"CREATE INDEX this_other_nam_1 ON sometable (foo)",
dialect=dialect
)
class ConstraintCompilationTest(TestBase, AssertsCompiledSQL):
def _test_deferrable(self, constraint_factory):
t = Table('tbl', MetaData(),
Column('a', Integer),
Column('b', Integer),
constraint_factory(deferrable=True))
sql = str(schema.CreateTable(t).compile(bind=testing.db))
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(bind=testing.db))
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(bind=testing.db))
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(bind=testing.db))
assert 'NOT DEFERRABLE' not in sql
assert 'INITIALLY DEFERRED' in sql
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_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_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()
t = Table('t', m,
Column('a', Integer),
)
t2 = Table('t2', m,
Column('a', Integer, ForeignKey('t.a', use_alter=True, name='fk_ta')),
Column('b', Integer, ForeignKey('t.a', name='fk_tb')), # to ensure create ordering ...
)
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 test_add_drop_constraint(self):
m = MetaData()
t = Table('tbl', m,
Column('a', Integer),
Column('b', Integer)
)
t2 = Table('t2', m,
Column('a', Integer),
Column('b', Integer)
)
constraint = 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"
")"
)
self.assert_compile(
schema.AddConstraint(constraint),
"ALTER TABLE tbl ADD CONSTRAINT my_test_constraint "
"CHECK (a < b) DEFERRABLE INITIALLY DEFERRED"
)
# 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"
")"
)
self.assert_compile(
schema.DropConstraint(constraint),
"ALTER TABLE tbl DROP CONSTRAINT my_test_constraint"
)
self.assert_compile(
schema.DropConstraint(constraint, cascade=True),
"ALTER TABLE tbl DROP CONSTRAINT my_test_constraint CASCADE"
)
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)"
)
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)"
)
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)"
)
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)"
)
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)"
)
|