summaryrefslogtreecommitdiff
path: root/test/sql/constraints.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2009-06-10 21:18:24 +0000
committerMike Bayer <mike_mp@zzzcomputing.com>2009-06-10 21:18:24 +0000
commit45cec095b4904ba71425d2fe18c143982dd08f43 (patch)
treeaf5e540fdcbf1cb2a3337157d69d4b40be010fa8 /test/sql/constraints.py
parent698a3c1ac665e7cd2ef8d5ad3ebf51b7fe6661f4 (diff)
downloadsqlalchemy-45cec095b4904ba71425d2fe18c143982dd08f43.tar.gz
- unit tests have been migrated from unittest to nose.
See README.unittests for information on how to run the tests. [ticket:970]
Diffstat (limited to 'test/sql/constraints.py')
-rw-r--r--test/sql/constraints.py337
1 files changed, 0 insertions, 337 deletions
diff --git a/test/sql/constraints.py b/test/sql/constraints.py
deleted file mode 100644
index d019aa037..000000000
--- a/test/sql/constraints.py
+++ /dev/null
@@ -1,337 +0,0 @@
-import testenv; testenv.configure_for_tests()
-from sqlalchemy import *
-from sqlalchemy import exc
-from testlib import *
-from testlib import config, engines
-
-class ConstraintTest(TestBase, AssertsExecutionResults):
-
- 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')
-
- self.assertRaises(exc.InvalidRequestError, Table, "a", metadata,
- Column('x', Integer, f),
- 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)
- try:
- foo.insert().execute(id=2,x=5,y=9)
- assert False
- except exc.SQLError:
- assert True
-
- bar.insert().execute(id=1,x=10)
- try:
- bar.insert().execute(id=2,x=5)
- assert False
- except exc.SQLError:
- assert True
-
- 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')
- try:
- foo.insert().execute(id=3, value='value1')
- assert False
- except exc.SQLError:
- assert True
- try:
- bar.insert().execute(id=3, value='a', value2='b')
- assert False
- except exc.SQLError:
- assert True
-
- 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)
-
- index_names = [ ix.name for ix in events.indexes ]
- assert 'ix_events_name' in index_names
- assert 'ix_events_location' in index_names
- assert 'sport_announcer' in index_names
- assert 'idx_winners' in index_names
- assert len(index_names) == 4
-
- capt = []
- connection = testing.db.connect()
- # TODO: hacky, put a real connection proxy in
- ex = connection._Connection__execute_context
- def proxy(context):
- capt.append(context.statement)
- capt.append(repr(context.parameters))
- ex(context)
- connection._Connection__execute_context = proxy
- schemagen = testing.db.dialect.schemagenerator(testing.db.dialect, connection)
- schemagen.traverse(events)
-
- assert capt[0].strip().startswith('CREATE TABLE events')
-
- s = set([capt[x].strip() for x in [2,4,6,8]])
-
- assert s == set([
- 'CREATE UNIQUE INDEX ix_events_name ON events (name)',
- 'CREATE INDEX ix_events_location ON events (location)',
- 'CREATE UNIQUE INDEX sport_announcer ON events (sport, announcer)',
- '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
-
- schemagen = dialect.schemagenerator(dialect, None)
- schemagen.execute = lambda : None
-
- t1 = Table("sometable", MetaData(), Column("foo", Integer))
- schemagen.visit_index(Index("this_name_is_too_long_for_what_were_doing", t1.c.foo))
- self.assertEquals(schemagen.buffer.getvalue(), "CREATE INDEX this_name_is_t_1 ON sometable (foo)")
- schemagen.buffer.truncate(0)
- schemagen.visit_index(Index("this_other_name_is_too_long_for_what_were_doing", t1.c.foo))
- self.assertEquals(schemagen.buffer.getvalue(), "CREATE INDEX this_other_nam_2 ON sometable (foo)")
-
- schemadrop = dialect.schemadropper(dialect, None)
- schemadrop.execute = lambda: None
- self.assertRaises(exc.IdentifierError, schemadrop.visit_index, Index("this_name_is_too_long_for_what_were_doing", t1.c.foo))
-
-
-class ConstraintCompilationTest(TestBase, AssertsExecutionResults):
- class accum(object):
- def __init__(self):
- self.statements = []
- def __call__(self, sql, *a, **kw):
- self.statements.append(sql)
- def __contains__(self, substring):
- for s in self.statements:
- if substring in s:
- return True
- return False
- def __str__(self):
- return '\n'.join([repr(x) for x in self.statements])
- def clear(self):
- del self.statements[:]
-
- def setUp(self):
- self.sql = self.accum()
- opts = config.db_opts.copy()
- opts['strategy'] = 'mock'
- opts['executor'] = self.sql
- self.engine = engines.testing_engine(options=opts)
-
-
- def _test_deferrable(self, constraint_factory):
- meta = MetaData(self.engine)
- t = Table('tbl', meta,
- Column('a', Integer),
- Column('b', Integer),
- constraint_factory(deferrable=True))
- t.create()
- assert 'DEFERRABLE' in self.sql, self.sql
- assert 'NOT DEFERRABLE' not in self.sql, self.sql
- self.sql.clear()
- meta.clear()
-
- t = Table('tbl', meta,
- Column('a', Integer),
- Column('b', Integer),
- constraint_factory(deferrable=False))
- t.create()
- assert 'NOT DEFERRABLE' in self.sql
- self.sql.clear()
- meta.clear()
-
- t = Table('tbl', meta,
- Column('a', Integer),
- Column('b', Integer),
- constraint_factory(deferrable=True, initially='IMMEDIATE'))
- t.create()
- assert 'NOT DEFERRABLE' not in self.sql
- assert 'INITIALLY IMMEDIATE' in self.sql
- self.sql.clear()
- meta.clear()
-
- t = Table('tbl', meta,
- Column('a', Integer),
- Column('b', Integer),
- constraint_factory(deferrable=True, initially='DEFERRED'))
- t.create()
-
- assert 'NOT DEFERRABLE' not in self.sql
- assert 'INITIALLY DEFERRED' in self.sql, self.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):
- meta = MetaData(self.engine)
- t = Table('tbl', meta,
- Column('a', Integer),
- Column('b', Integer,
- ForeignKey('tbl.a', deferrable=True,
- initially='DEFERRED')))
- t.create()
- assert 'DEFERRABLE' in self.sql, self.sql
- assert 'INITIALLY DEFERRED' in self.sql, self.sql
-
- 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):
- meta = MetaData(self.engine)
- t = Table('tbl', meta,
- Column('a', Integer),
- Column('b', Integer,
- CheckConstraint('a < b',
- deferrable=True,
- initially='DEFERRED')))
- t.create()
- assert 'DEFERRABLE' in self.sql, self.sql
- assert 'INITIALLY DEFERRED' in self.sql, self.sql
-
-
-if __name__ == "__main__":
- testenv.main()