From: Mike Bayer <mike_mp@zzzcomputing.com>
Date: Tue, 14 Oct 2014 16:07:38 +0000 (-0400)
Subject: - move BindTest into orm/test_binds
X-Git-Tag: rel_1_0_0b1~70^2~24
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=1e5ffa029a2b9adc1a6a3c83fb7e88a99d4e6448;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git

- move BindTest into orm/test_binds
---

diff --git a/test/orm/test_bind.py b/test/orm/test_bind.py
index 0d869130bd..3e5af0cba4 100644
--- a/test/orm/test_bind.py
+++ b/test/orm/test_bind.py
@@ -1,14 +1,205 @@
-from sqlalchemy.testing import assert_raises, assert_raises_message
+from sqlalchemy.testing import assert_raises_message
 from sqlalchemy import MetaData, Integer
 from sqlalchemy.testing.schema import Table
 from sqlalchemy.testing.schema import Column
 from sqlalchemy.orm import mapper, create_session
 import sqlalchemy as sa
 from sqlalchemy import testing
-from sqlalchemy.testing import fixtures
+from sqlalchemy.testing import fixtures, eq_, engines
+from sqlalchemy.orm import relationship, Session, backref, sessionmaker
+from test.orm import _fixtures
 
 
-class BindTest(fixtures.MappedTest):
+class BindIntegrationTest(_fixtures.FixtureTest):
+    run_inserts = None
+
+    def test_mapped_binds(self):
+        Address, addresses, users, User = (self.classes.Address,
+                                           self.tables.addresses,
+                                           self.tables.users,
+                                           self.classes.User)
+
+        # ensure tables are unbound
+        m2 = sa.MetaData()
+        users_unbound = users.tometadata(m2)
+        addresses_unbound = addresses.tometadata(m2)
+
+        mapper(Address, addresses_unbound)
+        mapper(User, users_unbound, properties={
+            'addresses': relationship(Address,
+                                      backref=backref("user", cascade="all"),
+                                      cascade="all")})
+
+        sess = Session(binds={User: self.metadata.bind,
+                              Address: self.metadata.bind})
+
+        u1 = User(id=1, name='ed')
+        sess.add(u1)
+        eq_(sess.query(User).filter(User.id == 1).all(),
+            [User(id=1, name='ed')])
+
+        # test expression binding
+
+        sess.execute(users_unbound.insert(), params=dict(id=2,
+                                                         name='jack'))
+        eq_(sess.execute(users_unbound.select(users_unbound.c.id
+                                              == 2)).fetchall(), [(2, 'jack')])
+
+        eq_(sess.execute(users_unbound.select(User.id == 2)).fetchall(),
+            [(2, 'jack')])
+
+        sess.execute(users_unbound.delete())
+        eq_(sess.execute(users_unbound.select()).fetchall(), [])
+
+        sess.close()
+
+    def test_table_binds(self):
+        Address, addresses, users, User = (self.classes.Address,
+                                           self.tables.addresses,
+                                           self.tables.users,
+                                           self.classes.User)
+
+        # ensure tables are unbound
+        m2 = sa.MetaData()
+        users_unbound = users.tometadata(m2)
+        addresses_unbound = addresses.tometadata(m2)
+
+        mapper(Address, addresses_unbound)
+        mapper(User, users_unbound, properties={
+            'addresses': relationship(Address,
+                                      backref=backref("user", cascade="all"),
+                                      cascade="all")})
+
+        Session = sessionmaker(binds={users_unbound: self.metadata.bind,
+                                      addresses_unbound: self.metadata.bind})
+        sess = Session()
+
+        u1 = User(id=1, name='ed')
+        sess.add(u1)
+        eq_(sess.query(User).filter(User.id == 1).all(),
+            [User(id=1, name='ed')])
+
+        sess.execute(users_unbound.insert(), params=dict(id=2, name='jack'))
+
+        eq_(sess.execute(users_unbound.select(users_unbound.c.id
+                                              == 2)).fetchall(), [(2, 'jack')])
+
+        eq_(sess.execute(users_unbound.select(User.id == 2)).fetchall(),
+            [(2, 'jack')])
+
+        sess.execute(users_unbound.delete())
+        eq_(sess.execute(users_unbound.select()).fetchall(), [])
+
+        sess.close()
+
+    def test_bind_from_metadata(self):
+        users, User = self.tables.users, self.classes.User
+
+        mapper(User, users)
+
+        session = create_session()
+        session.execute(users.insert(), dict(name='Johnny'))
+
+        assert len(session.query(User).filter_by(name='Johnny').all()) == 1
+
+        session.execute(users.delete())
+
+        assert len(session.query(User).filter_by(name='Johnny').all()) == 0
+        session.close()
+
+    def test_bind_arguments(self):
+        users, Address, addresses, User = (self.tables.users,
+                                           self.classes.Address,
+                                           self.tables.addresses,
+                                           self.classes.User)
+
+        mapper(User, users)
+        mapper(Address, addresses)
+
+        e1 = engines.testing_engine()
+        e2 = engines.testing_engine()
+        e3 = engines.testing_engine()
+
+        sess = Session(e3)
+        sess.bind_mapper(User, e1)
+        sess.bind_mapper(Address, e2)
+
+        assert sess.connection().engine is e3
+        assert sess.connection(bind=e1).engine is e1
+        assert sess.connection(mapper=Address, bind=e1).engine is e1
+        assert sess.connection(mapper=Address).engine is e2
+        assert sess.connection(clause=addresses.select()).engine is e2
+        assert sess.connection(mapper=User,
+                               clause=addresses.select()).engine is e1
+        assert sess.connection(mapper=User,
+                               clause=addresses.select(),
+                               bind=e2).engine is e2
+
+        sess.close()
+
+    @engines.close_open_connections
+    def test_bound_connection(self):
+        users, User = self.tables.users, self.classes.User
+
+        mapper(User, users)
+        c = testing.db.connect()
+        sess = create_session(bind=c)
+        sess.begin()
+        transaction = sess.transaction
+        u = User(name='u1')
+        sess.add(u)
+        sess.flush()
+        assert transaction._connection_for_bind(testing.db) \
+            is transaction._connection_for_bind(c) is c
+
+        assert_raises_message(sa.exc.InvalidRequestError,
+                              'Session already has a Connection '
+                              'associated',
+                              transaction._connection_for_bind,
+                              testing.db.connect())
+        transaction.rollback()
+        assert len(sess.query(User).all()) == 0
+        sess.close()
+
+    def test_bound_connection_transactional(self):
+        User, users = self.classes.User, self.tables.users
+
+        mapper(User, users)
+        c = testing.db.connect()
+
+        sess = create_session(bind=c, autocommit=False)
+        u = User(name='u1')
+        sess.add(u)
+        sess.flush()
+        sess.close()
+        assert not c.in_transaction()
+        assert c.scalar("select count(1) from users") == 0
+
+        sess = create_session(bind=c, autocommit=False)
+        u = User(name='u2')
+        sess.add(u)
+        sess.flush()
+        sess.commit()
+        assert not c.in_transaction()
+        assert c.scalar("select count(1) from users") == 1
+        c.execute("delete from users")
+        assert c.scalar("select count(1) from users") == 0
+
+        c = testing.db.connect()
+
+        trans = c.begin()
+        sess = create_session(bind=c, autocommit=True)
+        u = User(name='u3')
+        sess.add(u)
+        sess.flush()
+        assert c.in_transaction()
+        trans.commit()
+        assert not c.in_transaction()
+        assert c.scalar("select count(1) from users") == 1
+
+
+class SessionBindTest(fixtures.MappedTest):
+
     @classmethod
     def define_tables(cls, metadata):
         Table('test_table', metadata,
@@ -58,5 +249,3 @@ class BindTest(fixtures.MappedTest):
             ('Could not locate a bind configured on Mapper|Foo|test_table '
              'or this Session'),
             sess.flush)
-
-
diff --git a/test/orm/test_session.py b/test/orm/test_session.py
index 74a7a7442b..06d1d73342 100644
--- a/test/orm/test_session.py
+++ b/test/orm/test_session.py
@@ -18,194 +18,6 @@ from sqlalchemy.testing import fixtures
 from test.orm import _fixtures
 from sqlalchemy import event, ForeignKey
 
-class BindTest(_fixtures.FixtureTest):
-    run_inserts = None
-
-    def test_mapped_binds(self):
-        Address, addresses, users, User = (self.classes.Address,
-                                self.tables.addresses,
-                                self.tables.users,
-                                self.classes.User)
-
-
-        # ensure tables are unbound
-        m2 = sa.MetaData()
-        users_unbound = users.tometadata(m2)
-        addresses_unbound = addresses.tometadata(m2)
-
-        mapper(Address, addresses_unbound)
-        mapper(User, users_unbound, properties={
-            'addresses': relationship(Address,
-                                 backref=backref("user", cascade="all"),
-                                 cascade="all")})
-
-        sess = Session(binds={User: self.metadata.bind,
-                                      Address: self.metadata.bind})
-
-        u1 = User(id=1, name='ed')
-        sess.add(u1)
-        eq_(sess.query(User).filter(User.id == 1).all(),
-            [User(id=1, name='ed')])
-
-        # test expression binding
-
-        sess.execute(users_unbound.insert(), params=dict(id=2,
-                     name='jack'))
-        eq_(sess.execute(users_unbound.select(users_unbound.c.id
-            == 2)).fetchall(), [(2, 'jack')])
-
-        eq_(sess.execute(users_unbound.select(User.id == 2)).fetchall(),
-            [(2, 'jack')])
-
-        sess.execute(users_unbound.delete())
-        eq_(sess.execute(users_unbound.select()).fetchall(), [])
-
-        sess.close()
-
-    def test_table_binds(self):
-        Address, addresses, users, User = (self.classes.Address,
-                                self.tables.addresses,
-                                self.tables.users,
-                                self.classes.User)
-
-
-        # ensure tables are unbound
-        m2 = sa.MetaData()
-        users_unbound = users.tometadata(m2)
-        addresses_unbound = addresses.tometadata(m2)
-
-        mapper(Address, addresses_unbound)
-        mapper(User, users_unbound, properties={
-            'addresses': relationship(Address,
-                                 backref=backref("user", cascade="all"),
-                                 cascade="all")})
-
-        Session = sessionmaker(binds={users_unbound: self.metadata.bind,
-                                      addresses_unbound: self.metadata.bind})
-        sess = Session()
-
-        u1 = User(id=1, name='ed')
-        sess.add(u1)
-        eq_(sess.query(User).filter(User.id == 1).all(),
-            [User(id=1, name='ed')])
-
-        sess.execute(users_unbound.insert(), params=dict(id=2, name='jack'))
-
-        eq_(sess.execute(users_unbound.select(users_unbound.c.id
-            == 2)).fetchall(), [(2, 'jack')])
-
-        eq_(sess.execute(users_unbound.select(User.id == 2)).fetchall(),
-            [(2, 'jack')])
-
-        sess.execute(users_unbound.delete())
-        eq_(sess.execute(users_unbound.select()).fetchall(), [])
-
-        sess.close()
-
-    def test_bind_from_metadata(self):
-        users, User = self.tables.users, self.classes.User
-
-        mapper(User, users)
-
-        session = create_session()
-        session.execute(users.insert(), dict(name='Johnny'))
-
-        assert len(session.query(User).filter_by(name='Johnny').all()) == 1
-
-        session.execute(users.delete())
-
-        assert len(session.query(User).filter_by(name='Johnny').all()) == 0
-        session.close()
-
-    def test_bind_arguments(self):
-        users, Address, addresses, User = (self.tables.users,
-                                self.classes.Address,
-                                self.tables.addresses,
-                                self.classes.User)
-
-        mapper(User, users)
-        mapper(Address, addresses)
-
-        e1 = engines.testing_engine()
-        e2 = engines.testing_engine()
-        e3 = engines.testing_engine()
-
-        sess = Session(e3)
-        sess.bind_mapper(User, e1)
-        sess.bind_mapper(Address, e2)
-
-        assert sess.connection().engine is e3
-        assert sess.connection(bind=e1).engine is e1
-        assert sess.connection(mapper=Address, bind=e1).engine is e1
-        assert sess.connection(mapper=Address).engine is e2
-        assert sess.connection(clause=addresses.select()).engine is e2
-        assert sess.connection(mapper=User,
-                                clause=addresses.select()).engine is e1
-        assert sess.connection(mapper=User,
-                                clause=addresses.select(),
-                                bind=e2).engine is e2
-
-        sess.close()
-
-    @engines.close_open_connections
-    def test_bound_connection(self):
-        users, User = self.tables.users, self.classes.User
-
-        mapper(User, users)
-        c = testing.db.connect()
-        sess = create_session(bind=c)
-        sess.begin()
-        transaction = sess.transaction
-        u = User(name='u1')
-        sess.add(u)
-        sess.flush()
-        assert transaction._connection_for_bind(testing.db) \
-            is transaction._connection_for_bind(c) is c
-
-        assert_raises_message(sa.exc.InvalidRequestError,
-                              'Session already has a Connection '
-                              'associated',
-                              transaction._connection_for_bind,
-                              testing.db.connect())
-        transaction.rollback()
-        assert len(sess.query(User).all()) == 0
-        sess.close()
-
-    def test_bound_connection_transactional(self):
-        User, users = self.classes.User, self.tables.users
-
-        mapper(User, users)
-        c = testing.db.connect()
-
-        sess = create_session(bind=c, autocommit=False)
-        u = User(name='u1')
-        sess.add(u)
-        sess.flush()
-        sess.close()
-        assert not c.in_transaction()
-        assert c.scalar("select count(1) from users") == 0
-
-        sess = create_session(bind=c, autocommit=False)
-        u = User(name='u2')
-        sess.add(u)
-        sess.flush()
-        sess.commit()
-        assert not c.in_transaction()
-        assert c.scalar("select count(1) from users") == 1
-        c.execute("delete from users")
-        assert c.scalar("select count(1) from users") == 0
-
-        c = testing.db.connect()
-
-        trans = c.begin()
-        sess = create_session(bind=c, autocommit=True)
-        u = User(name='u3')
-        sess.add(u)
-        sess.flush()
-        assert c.in_transaction()
-        trans.commit()
-        assert not c.in_transaction()
-        assert c.scalar("select count(1) from users") == 1
 
 class ExecutionTest(_fixtures.FixtureTest):
     run_inserts = None