From: Mike Bayer Date: Mon, 11 Jun 2007 19:31:19 +0000 (+0000) Subject: added separate test for metadata X-Git-Tag: rel_0_4_6~210 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=ce2d1c715912918961cfb424920adc3d86f46358;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git added separate test for metadata some cleanup to query.py unit test added settable "engine" property on MetaData --- diff --git a/lib/sqlalchemy/schema.py b/lib/sqlalchemy/schema.py index 492210061e..b651774a24 100644 --- a/lib/sqlalchemy/schema.py +++ b/lib/sqlalchemy/schema.py @@ -1122,6 +1122,16 @@ class MetaData(SchemaItem): else: self._engine = engine_or_url + def _get_engine(self): + # we are checking is_bound() because _engine wires + # into SchemaItem's _engine mechanism, which raises an error, + # whereas we just want to return None. + if not self.is_bound(): + return None + return self._engine + + engine = property(_get_engine, connect) + def clear(self): self.tables.clear() @@ -1181,10 +1191,6 @@ class MetaData(SchemaItem): def _derived_metadata(self): return self - def _get_engine(self): - if not self.is_bound(): - return None - return self._engine class BoundMetaData(MetaData): """``MetaData`` for which the first argument is a required Engine, url string, or URL instance. @@ -1241,7 +1247,8 @@ thread-local basis. return self.context._engine else: return None - engine=property(_get_engine) + + engine = property(_get_engine, connect) class SchemaVisitor(sql.ClauseVisitor): """Define the visiting for ``SchemaItem`` objects.""" diff --git a/test/engine/alltests.py b/test/engine/alltests.py index 5b96bc1148..722f06256d 100644 --- a/test/engine/alltests.py +++ b/test/engine/alltests.py @@ -9,6 +9,7 @@ def suite(): 'engine.pool', 'engine.reconnect', 'engine.execute', + 'engine.metadata', 'engine.transaction', # schema/tables diff --git a/test/engine/metadata.py b/test/engine/metadata.py new file mode 100644 index 0000000000..95b601cd3d --- /dev/null +++ b/test/engine/metadata.py @@ -0,0 +1,32 @@ +import testbase +from sqlalchemy import * + +class MetaDataTest(testbase.PersistTest): + def test_global_metadata(self): + t1 = Table('table1', Column('col1', Integer, primary_key=True), + Column('col2', String(20))) + t2 = Table('table2', Column('col1', Integer, primary_key=True), + Column('col2', String(20))) + + assert t1.c.col1 + global_connect(testbase.db) + default_metadata.create_all() + try: + assert t1.count().scalar() == 0 + finally: + default_metadata.drop_all() + default_metadata.clear() + + def test_metadata_connect(self): + metadata = MetaData() + t1 = Table('table1', metadata, Column('col1', Integer, primary_key=True), + Column('col2', String(20))) + metadata.engine = testbase.db + metadata.create_all() + try: + assert t1.count().scalar() == 0 + finally: + metadata.drop_all() + +if __name__ == '__main__': + testbase.main() \ No newline at end of file diff --git a/test/sql/defaults.py b/test/sql/defaults.py index 992a3afc03..9867a52f36 100644 --- a/test/sql/defaults.py +++ b/test/sql/defaults.py @@ -125,6 +125,33 @@ class DefaultTest(PersistTest): l = l.fetchone() self.assert_(l['col3'] == 55) + @testbase.supported('postgres') + def testpassiveoverride(self): + """primarily for postgres, tests that when we get a primary key column back + from reflecting a table which has a default value on it, we pre-execute + that PassiveDefault upon insert, even though PassiveDefault says + "let the database execute this", because in postgres we must have all the primary + key values in memory before insert; otherwise we cant locate the just inserted row.""" + + try: + meta = BoundMetaData(testbase.db) + testbase.db.execute(""" + CREATE TABLE speedy_users + ( + speedy_user_id SERIAL PRIMARY KEY, + + user_name VARCHAR NOT NULL, + user_password VARCHAR NOT NULL + ); + """, None) + + t = Table("speedy_users", meta, autoload=True) + t.insert().execute(user_name='user', user_password='lala') + l = t.select().execute().fetchall() + self.assert_(l == [(1, 'user', 'lala')]) + finally: + testbase.db.execute("drop table speedy_users", None) + class AutoIncrementTest(PersistTest): @testbase.supported('postgres', 'mysql') def testnonautoincrement(self): diff --git a/test/sql/query.py b/test/sql/query.py index d788544c09..5d5a734108 100644 --- a/test/sql/query.py +++ b/test/sql/query.py @@ -24,91 +24,49 @@ class QueryTest(PersistTest): Column('address', String(30))) metadata.create_all() - def setUp(self): - self.users = users def tearDown(self): - self.users.delete().execute() + users.delete().execute() def tearDownAll(self): metadata.drop_all() def testinsert(self): - self.users.insert().execute(user_id = 7, user_name = 'jack') - print repr(self.users.select().execute().fetchall()) + users.insert().execute(user_id = 7, user_name = 'jack') + assert users.count().scalar() == 1 def testupdate(self): - self.users.insert().execute(user_id = 7, user_name = 'jack') - print repr(self.users.select().execute().fetchall()) + users.insert().execute(user_id = 7, user_name = 'jack') + assert users.count().scalar() == 1 - self.users.update(self.users.c.user_id == 7).execute(user_name = 'fred') - print repr(self.users.select().execute().fetchall()) + users.update(users.c.user_id == 7).execute(user_name = 'fred') + assert users.select(users.c.user_id==7).execute().fetchone()['user_name'] == 'fred' def testrowiteration(self): - self.users.insert().execute(user_id = 7, user_name = 'jack') - self.users.insert().execute(user_id = 8, user_name = 'ed') - self.users.insert().execute(user_id = 9, user_name = 'fred') - r = self.users.select().execute() + users.insert().execute(user_id = 7, user_name = 'jack') + users.insert().execute(user_id = 8, user_name = 'ed') + users.insert().execute(user_id = 9, user_name = 'fred') + r = users.select().execute() l = [] for row in r: l.append(row) self.assert_(len(l) == 3) def test_fetchmany(self): - self.users.insert().execute(user_id = 7, user_name = 'jack') - self.users.insert().execute(user_id = 8, user_name = 'ed') - self.users.insert().execute(user_id = 9, user_name = 'fred') - r = self.users.select().execute() + users.insert().execute(user_id = 7, user_name = 'jack') + users.insert().execute(user_id = 8, user_name = 'ed') + users.insert().execute(user_id = 9, user_name = 'fred') + r = users.select().execute() l = [] for row in r.fetchmany(size=2): l.append(row) self.assert_(len(l) == 2, "fetchmany(size=2) got %s rows" % len(l)) def test_compiled_execute(self): - s = select([self.users], self.users.c.user_id==bindparam('id')).compile() + users.insert().execute(user_id = 7, user_name = 'jack') + s = select([users], users.c.user_id==bindparam('id')).compile() c = testbase.db.connect() - print repr(c.execute(s, id=7).fetchall()) - - def test_global_metadata(self): - t1 = Table('table1', Column('col1', Integer, primary_key=True), - Column('col2', String(20))) - t2 = Table('table2', Column('col1', Integer, primary_key=True), - Column('col2', String(20))) - - assert t1.c.col1 - global_connect(testbase.db) - default_metadata.create_all() - try: - assert t1.count().scalar() == 0 - finally: - default_metadata.drop_all() - default_metadata.clear() - - @testbase.supported('postgres') - def testpassiveoverride(self): - """primarily for postgres, tests that when we get a primary key column back - from reflecting a table which has a default value on it, we pre-execute - that PassiveDefault upon insert, even though PassiveDefault says - "let the database execute this", because in postgres we must have all the primary - key values in memory before insert; otherwise we cant locate the just inserted row.""" - try: - meta = BoundMetaData(testbase.db) - testbase.db.execute(""" - CREATE TABLE speedy_users - ( - speedy_user_id SERIAL PRIMARY KEY, - - user_name VARCHAR NOT NULL, - user_password VARCHAR NOT NULL - ); - """, None) - - t = Table("speedy_users", meta, autoload=True) - t.insert().execute(user_name='user', user_password='lala') - l = t.select().execute().fetchall() - self.assert_(l == [(1, 'user', 'lala')]) - finally: - testbase.db.execute("drop table speedy_users", None) + assert c.execute(s, id=7).fetchall()[0]['user_id'] == 7 @testbase.supported('postgres') def testschema(self): @@ -133,61 +91,61 @@ class QueryTest(PersistTest): def test_repeated_bindparams(self): """test that a BindParam can be used more than once. this should be run for dbs with both positional and named paramstyles.""" - self.users.insert().execute(user_id = 7, user_name = 'jack') - self.users.insert().execute(user_id = 8, user_name = 'fred') + users.insert().execute(user_id = 7, user_name = 'jack') + users.insert().execute(user_id = 8, user_name = 'fred') u = bindparam('userid') - s = self.users.select(or_(self.users.c.user_name==u, self.users.c.user_name==u)) + s = users.select(or_(users.c.user_name==u, users.c.user_name==u)) r = s.execute(userid='fred').fetchall() assert len(r) == 1 def test_bindparam_shortname(self): """test the 'shortname' field on BindParamClause.""" - self.users.insert().execute(user_id = 7, user_name = 'jack') - self.users.insert().execute(user_id = 8, user_name = 'fred') + users.insert().execute(user_id = 7, user_name = 'jack') + users.insert().execute(user_id = 8, user_name = 'fred') u = bindparam('userid', shortname='someshortname') - s = self.users.select(self.users.c.user_name==u) + s = users.select(users.c.user_name==u) r = s.execute(someshortname='fred').fetchall() assert len(r) == 1 def testdelete(self): - self.users.insert().execute(user_id = 7, user_name = 'jack') - self.users.insert().execute(user_id = 8, user_name = 'fred') - print repr(self.users.select().execute().fetchall()) + users.insert().execute(user_id = 7, user_name = 'jack') + users.insert().execute(user_id = 8, user_name = 'fred') + print repr(users.select().execute().fetchall()) - self.users.delete(self.users.c.user_name == 'fred').execute() + users.delete(users.c.user_name == 'fred').execute() - print repr(self.users.select().execute().fetchall()) + print repr(users.select().execute().fetchall()) def testselectlimit(self): - self.users.insert().execute(user_id=1, user_name='john') - self.users.insert().execute(user_id=2, user_name='jack') - self.users.insert().execute(user_id=3, user_name='ed') - self.users.insert().execute(user_id=4, user_name='wendy') - self.users.insert().execute(user_id=5, user_name='laura') - self.users.insert().execute(user_id=6, user_name='ralph') - self.users.insert().execute(user_id=7, user_name='fido') - r = self.users.select(limit=3, order_by=[self.users.c.user_id]).execute().fetchall() + users.insert().execute(user_id=1, user_name='john') + users.insert().execute(user_id=2, user_name='jack') + users.insert().execute(user_id=3, user_name='ed') + users.insert().execute(user_id=4, user_name='wendy') + users.insert().execute(user_id=5, user_name='laura') + users.insert().execute(user_id=6, user_name='ralph') + users.insert().execute(user_id=7, user_name='fido') + r = users.select(limit=3, order_by=[users.c.user_id]).execute().fetchall() self.assert_(r == [(1, 'john'), (2, 'jack'), (3, 'ed')], repr(r)) @testbase.unsupported('mssql') def testselectlimitoffset(self): - self.users.insert().execute(user_id=1, user_name='john') - self.users.insert().execute(user_id=2, user_name='jack') - self.users.insert().execute(user_id=3, user_name='ed') - self.users.insert().execute(user_id=4, user_name='wendy') - self.users.insert().execute(user_id=5, user_name='laura') - self.users.insert().execute(user_id=6, user_name='ralph') - self.users.insert().execute(user_id=7, user_name='fido') - r = self.users.select(limit=3, offset=2, order_by=[self.users.c.user_id]).execute().fetchall() + users.insert().execute(user_id=1, user_name='john') + users.insert().execute(user_id=2, user_name='jack') + users.insert().execute(user_id=3, user_name='ed') + users.insert().execute(user_id=4, user_name='wendy') + users.insert().execute(user_id=5, user_name='laura') + users.insert().execute(user_id=6, user_name='ralph') + users.insert().execute(user_id=7, user_name='fido') + r = users.select(limit=3, offset=2, order_by=[users.c.user_id]).execute().fetchall() self.assert_(r==[(3, 'ed'), (4, 'wendy'), (5, 'laura')]) - r = self.users.select(offset=5, order_by=[self.users.c.user_id]).execute().fetchall() + r = users.select(offset=5, order_by=[users.c.user_id]).execute().fetchall() self.assert_(r==[(6, 'ralph'), (7, 'fido')]) @testbase.supported('mssql') def testselectlimitoffset_mssql(self): try: - r = self.users.select(limit=3, offset=2, order_by=[self.users.c.user_id]).execute().fetchall() + r = users.select(limit=3, offset=2, order_by=[users.c.user_id]).execute().fetchall() assert False # InvalidRequestError should have been raised except exceptions.InvalidRequestError: pass @@ -210,29 +168,29 @@ class QueryTest(PersistTest): datetable.drop() def test_column_accessor(self): - self.users.insert().execute(user_id=1, user_name='john') - self.users.insert().execute(user_id=2, user_name='jack') - r = self.users.select(self.users.c.user_id==2).execute().fetchone() - self.assert_(r.user_id == r['user_id'] == r[self.users.c.user_id] == 2) - self.assert_(r.user_name == r['user_name'] == r[self.users.c.user_name] == 'jack') + users.insert().execute(user_id=1, user_name='john') + users.insert().execute(user_id=2, user_name='jack') + r = users.select(users.c.user_id==2).execute().fetchone() + self.assert_(r.user_id == r['user_id'] == r[users.c.user_id] == 2) + self.assert_(r.user_name == r['user_name'] == r[users.c.user_name] == 'jack') r = text("select * from query_users where user_id=2", engine=testbase.db).execute().fetchone() - self.assert_(r.user_id == r['user_id'] == r[self.users.c.user_id] == 2) - self.assert_(r.user_name == r['user_name'] == r[self.users.c.user_name] == 'jack') + self.assert_(r.user_id == r['user_id'] == r[users.c.user_id] == 2) + self.assert_(r.user_name == r['user_name'] == r[users.c.user_name] == 'jack') def test_keys(self): - self.users.insert().execute(user_id=1, user_name='foo') - r = self.users.select().execute().fetchone() + users.insert().execute(user_id=1, user_name='foo') + r = users.select().execute().fetchone() self.assertEqual([x.lower() for x in r.keys()], ['user_id', 'user_name']) def test_items(self): - self.users.insert().execute(user_id=1, user_name='foo') - r = self.users.select().execute().fetchone() + users.insert().execute(user_id=1, user_name='foo') + r = users.select().execute().fetchone() self.assertEqual([(x[0].lower(), x[1]) for x in r.items()], [('user_id', 1), ('user_name', 'foo')]) def test_len(self): - self.users.insert().execute(user_id=1, user_name='foo') - r = self.users.select().execute().fetchone() + users.insert().execute(user_id=1, user_name='foo') + r = users.select().execute().fetchone() self.assertEqual(len(r), 2) r.close() r = testbase.db.execute('select user_name, user_id from query_users', {}).fetchone() @@ -330,8 +288,8 @@ class QueryTest(PersistTest): def test_column_order_with_simple_query(self): # should return values in column definition order - self.users.insert().execute(user_id=1, user_name='foo') - r = self.users.select(self.users.c.user_id==1).execute().fetchone() + users.insert().execute(user_id=1, user_name='foo') + r = users.select(users.c.user_id==1).execute().fetchone() self.assertEqual(r[0], 1) self.assertEqual(r[1], 'foo') self.assertEqual([x.lower() for x in r.keys()], ['user_id', 'user_name']) @@ -339,7 +297,7 @@ class QueryTest(PersistTest): def test_column_order_with_text_query(self): # should return values in query order - self.users.insert().execute(user_id=1, user_name='foo') + users.insert().execute(user_id=1, user_name='foo') r = testbase.db.execute('select user_name, user_id from query_users', {}).fetchone() self.assertEqual(r[0], 'foo') self.assertEqual(r[1], 1)