- Moved an ORM test out of engine...

diff --git a/test/engine/bind.py b/test/engine/bind.py
index 300a4eae6e845c2309190c8db3f4119c7be80b24..71043cb83b8540d6433c1299034fadd8c531f9a2 100644 (file)
--- a/test/engine/bind.py
+++ b/test/engine/bind.py
@@ -1,14 +1,14 @@
-"""tests the "bind" attribute/argument across schema, SQL, and ORM sessions,
+"""tests the "bind" attribute/argument across schema and SQL,
 including the deprecated versions of these arguments"""
 
 import testenv; testenv.configure_for_tests()
 from sqlalchemy import engine, exc
 from sqlalchemy import MetaData, ThreadLocalMetaData
-from testlib.sa import Table, Column, Integer, String, func, Sequence, text
-from testlib import TestBase, testing
+from testlib.sa import Table, Column, Integer, text
+from testlib import sa, testing
 
 
-class BindTest(TestBase):
+class BindTest(testing.TestBase):
     def test_create_drop_explicit(self):
         metadata = MetaData()
         table = Table('test_table', metadata,
@@ -185,7 +185,7 @@ class BindTest(TestBase):
         try:
             for elem in [
                 table.select,
-                lambda **kwargs:func.current_timestamp(**kwargs).select(),
+                lambda **kwargs: sa.func.current_timestamp(**kwargs).select(),
 #                func.current_timestamp().select,
                 lambda **kwargs:text("select * from test_table", **kwargs)
             ]:
@@ -218,48 +218,6 @@ class BindTest(TestBase):
                 bind.close()
             metadata.drop_all(bind=testing.db)
 
-    def test_session(self):
-        from sqlalchemy.orm import create_session, mapper
-        metadata = MetaData()
-        table = Table('test_table', metadata,
-            Column('foo', Integer, Sequence('foo_seq', optional=True), primary_key=True),
-            Column('data', String(30)))
-        class Foo(object):
-            pass
-        mapper(Foo, table)
-        metadata.create_all(bind=testing.db)
-        try:
-            for bind in (testing.db,
-                testing.db.connect()
-                ):
-                try:
-                    for args in ({'bind':bind},):
-                        sess = create_session(**args)
-                        assert sess.bind is bind
-                        f = Foo()
-                        sess.save(f)
-                        sess.flush()
-                        assert sess.get(Foo, f.foo) is f
-                finally:
-                    if isinstance(bind, engine.Connection):
-                        bind.close()
-
-                if isinstance(bind, engine.Connection):
-                    bind.close()
-
-            sess = create_session()
-            f = Foo()
-            sess.save(f)
-            try:
-                sess.flush()
-                assert False
-            except exc.InvalidRequestError, e:
-                assert str(e).startswith("Could not locate any Engine or Connection bound to mapper")
-        finally:
-            if isinstance(bind, engine.Connection):
-                bind.close()
-            metadata.drop_all(bind=testing.db)
-
 
 if __name__ == '__main__':
     testenv.main()

diff --git a/test/orm/bind.py b/test/orm/bind.py
new file mode 100644 (file)
index 0000000..47fba91
--- /dev/null
+++ b/test/orm/bind.py
@@ -0,0 +1,55 @@
+import testenv; testenv.configure_for_tests()
+from testlib.sa import MetaData, Table, Column, Integer
+from testlib.sa.orm import mapper, create_session
+from testlib import sa, testing
+from orm import _base
+
+
+class BindTest(_base.MappedTest):
+    def define_tables(self, metadata):
+        Table('test_table', metadata,
+              Column('id', Integer, primary_key=True,
+                     test_needs_autoincrement=True),
+              Column('data', Integer))
+
+    def setup_classes(self):
+        class Foo(_base.BasicEntity):
+            pass
+
+    @testing.resolve_artifact_names
+    def setup_mappers(self):
+        meta = MetaData()
+        test_table.tometadata(meta)
+
+        assert meta.tables['test_table'].bind is None
+        mapper(Foo, meta.tables['test_table'])
+
+    @testing.resolve_artifact_names
+    def test_session_bind(self):
+        engine = self.metadata.bind
+
+        for bind in (engine, engine.connect()):
+            try:
+                sess = create_session(bind=bind)
+                assert sess.bind is bind
+                f = Foo()
+                sess.save(f)
+                sess.flush()
+                assert sess.get(Foo, f.id) is f
+            finally:
+                if hasattr(bind, 'close'):
+                    bind.close()
+
+    @testing.resolve_artifact_names
+    def test_session_unbound(self):
+        sess = create_session()
+        sess.add(Foo())
+        self.assertRaisesMessage(
+            sa.exc.UnboundExecutionError,
+            ('Could not locate a bind configured on Mapper|Foo|test_table '
+             'or this Session'),
+            sess.flush)
+
+
+if __name__ == '__main__':
+    testenv.main()