diff options
Diffstat (limited to 'test/engine/bind.py')
| -rw-r--r-- | test/engine/bind.py | 172 |
1 files changed, 172 insertions, 0 deletions
diff --git a/test/engine/bind.py b/test/engine/bind.py new file mode 100644 index 000000000..d8afff3d7 --- /dev/null +++ b/test/engine/bind.py @@ -0,0 +1,172 @@ +"""tests the "bind" attribute/argument across schema, SQL, and ORM sessions, +including the deprecated versions of these arguments""" + +import testbase +import unittest, sys, datetime +import tables +db = testbase.db +from sqlalchemy import * + +class BindTest(testbase.PersistTest): + def test_create_drop_explicit(self): + metadata = MetaData() + table = Table('test_table', metadata, + Column('foo', Integer)) + for bind in ( + testbase.db, + testbase.db.connect() + ): + for args in [ + ([], {'connectable':bind}), + ([], {'bind':bind}), + ([bind], {}) + ]: + metadata.create_all(*args[0], **args[1]) + assert table.exists(*args[0], **args[1]) + metadata.drop_all(*args[0], **args[1]) + table.create(*args[0], **args[1]) + table.drop(*args[0], **args[1]) + assert not table.exists(*args[0], **args[1]) + + def test_create_drop_err(self): + metadata = MetaData() + table = Table('test_table', metadata, + Column('foo', Integer)) + + for meth in [ + metadata.create_all, + table.exists, + metadata.drop_all, + table.create, + table.drop, + ]: + try: + meth() + assert False + except exceptions.InvalidRequestError, e: + assert str(e) == "This SchemaItem is not connected to any Engine or Connection." + + def test_create_drop_bound(self): + + for meta in (MetaData,ThreadLocalMetaData): + for bind in ( + testbase.db, + testbase.db.connect() + ): + metadata = meta() + table = Table('test_table', metadata, + Column('foo', Integer)) + metadata.bind = bind + assert metadata.bind is metadata.engine is table.bind is table.engine is bind + metadata.create_all() + assert table.exists() + metadata.drop_all() + table.create() + table.drop() + assert not table.exists() + + metadata = meta() + table = Table('test_table', metadata, + Column('foo', Integer)) + + metadata.connect(bind) + assert metadata.bind is metadata.engine is table.bind is table.engine is bind + metadata.create_all() + assert table.exists() + metadata.drop_all() + table.create() + table.drop() + assert not table.exists() + + def test_create_drop_constructor_bound(self): + for bind in ( + testbase.db, + testbase.db.connect() + ): + for args in ( + ([bind], {}), + ([], {'engine_or_url':bind}), + ([], {'bind':bind}), + ([], {'engine':bind}) + ): + metadata = MetaData(*args[0], **args[1]) + table = Table('test_table', metadata, + Column('foo', Integer)) + + assert metadata.bind is metadata.engine is table.bind is table.engine is bind + metadata.create_all() + assert table.exists() + metadata.drop_all() + table.create() + table.drop() + assert not table.exists() + + + def test_clauseelement(self): + metadata = MetaData() + table = Table('test_table', metadata, + Column('foo', Integer)) + metadata.create_all(bind=testbase.db) + try: + for elem in [ + table.select, + lambda **kwargs:func.current_timestamp(**kwargs).select(), +# func.current_timestamp().select, + lambda **kwargs:text("select * from test_table", **kwargs) + ]: + for bind in ( + testbase.db, + testbase.db.connect() + ): + e = elem(bind=bind) + assert e.bind is e.engine is bind + e.execute() + e = elem(engine=bind) + assert e.bind is e.engine is bind + e.execute() + + try: + e = elem() + assert e.bind is e.engine is None + e.execute() + assert False + except exceptions.InvalidRequestError, e: + assert str(e) == "This Compiled object is not bound to any Engine or Connection." + + finally: + metadata.drop_all(bind=testbase.db) + + def test_session(self): + metadata = MetaData() + table = Table('test_table', metadata, + Column('foo', Integer, primary_key=True), + Column('data', String(30))) + class Foo(object): + pass + mapper(Foo, table) + metadata.create_all(bind=testbase.db) + try: + for bind in (testbase.db, testbase.db.connect()): + for args in ({'bind':bind}, {'bind_to':bind}): + sess = create_session(**args) + assert sess.bind is sess.bind_to is bind + f = Foo() + sess.save(f) + sess.flush() + assert sess.get(Foo, f.foo) is f + + sess = create_session() + f = Foo() + sess.save(f) + try: + sess.flush() + assert False + except exceptions.InvalidRequestError, e: + assert str(e).startswith("Could not locate any Engine or Connection bound to mapper") + + finally: + metadata.drop_all(bind=testbase.db) + + +if __name__ == '__main__': + testbase.main()
\ No newline at end of file |