merge tip

22 files changed, 1323 insertions, 207 deletions

diff --git a/test/aaa_profiling/test_zoomark_orm.py b/test/aaa_profiling/test_zoomark_orm.py
index 0b699eead..3e30efa24 100644
--- a/test/aaa_profiling/test_zoomark_orm.py
+++ b/test/aaa_profiling/test_zoomark_orm.py
@@ -366,7 +366,7 @@ class ZooMarkTest(TestBase):
     def test_profile_5_aggregates(self):
         self.test_baseline_5_aggregates()
 
-    @profiling.function_call_count(3172)
+    @profiling.function_call_count(2929)
     def test_profile_6_editing(self):
         self.test_baseline_6_editing()
 
diff --git a/test/dialect/test_firebird.py b/test/dialect/test_firebird.py
index a9b9fa262..41a50e6a3 100644
--- a/test/dialect/test_firebird.py
+++ b/test/dialect/test_firebird.py
@@ -94,7 +94,8 @@ class DomainReflectionTest(TestBase, AssertsExecutionResults):
 
 
 class BuggyDomainReflectionTest(TestBase, AssertsExecutionResults):
-    "Test Firebird domains, see [ticket:1663] and http://tracker.firebirdsql.org/browse/CORE-356"
+    """Test Firebird domains (and some other reflection bumps), 
+    see [ticket:1663] and http://tracker.firebirdsql.org/browse/CORE-356"""
 
     __only_on__ = 'firebird'
 
@@ -168,6 +169,12 @@ CREATE DOMAIN DOM_ID INTEGER NOT NULL
 CREATE TABLE A (
 ID DOM_ID /* INTEGER NOT NULL */ DEFAULT 0 )
 """
+    
+    # the 'default' keyword is lower case here
+    TABLE_B = """\
+CREATE TABLE B (
+ID DOM_ID /* INTEGER NOT NULL */ default 0 )
+"""
 
     @classmethod
     def setup_class(cls):
@@ -181,11 +188,13 @@ ID DOM_ID /* INTEGER NOT NULL */ DEFAULT 0 )
 
         con.execute(cls.DOM_ID)
         con.execute(cls.TABLE_A)
+        con.execute(cls.TABLE_B)
 
     @classmethod
     def teardown_class(cls):
         con = testing.db.connect()
         con.execute('DROP TABLE a')
+        con.execute("DROP TABLE b")
         con.execute('DROP DOMAIN dom_id')
         con.execute('DROP TABLE def_error_nodom')
         con.execute('DROP TABLE def_error')
@@ -213,7 +222,14 @@ ID DOM_ID /* INTEGER NOT NULL */ DEFAULT 0 )
         table_a = Table('a', metadata, autoload=True)
 
         eq_(table_a.c.id.server_default.arg.text, "0")
+    
+    def test_lowercase_default_name(self):
+        metadata = MetaData(testing.db)
+
+        table_b = Table('b', metadata, autoload=True)
 
+        eq_(table_b.c.id.server_default.arg.text, "0")
+        
 
 class CompileTest(TestBase, AssertsCompiledSQL):
 
diff --git a/test/dialect/test_oracle.py b/test/dialect/test_oracle.py
index 384066c41..29d18b988 100644
--- a/test/dialect/test_oracle.py
+++ b/test/dialect/test_oracle.py
@@ -1121,7 +1121,80 @@ class UnsupportedIndexReflectTest(TestBase):
                            'TEST_INDEX_REFLECT (UPPER(DATA))')
         m2 = MetaData(testing.db)
         t2 = Table('test_index_reflect', m2, autoload=True)
-        
+    
+class RoundTripIndexTest(TestBase):
+    __only_on__ = 'oracle'
+
+    def test_basic(self):
+        engine = testing.db
+        metadata = MetaData(engine)
+
+        table=Table("sometable", metadata,
+            Column("id_a", Unicode(255), primary_key=True),
+            Column("id_b", Unicode(255), primary_key=True, unique=True),
+            Column("group", Unicode(255), primary_key=True),
+            Column("col", Unicode(255)),
+            UniqueConstraint('col','group'),
+        )
+
+        # "group" is a keyword, so lower case
+        normalind = Index('tableind', table.c.id_b, table.c.group) 
+
+        # create
+        metadata.create_all()
+        try:
+            # round trip, create from reflection
+            mirror = MetaData(engine)
+            mirror.reflect()
+            metadata.drop_all()
+            mirror.create_all()
+
+            # inspect the reflected creation
+            inspect = MetaData(engine)
+            inspect.reflect()
+
+            def obj_definition(obj):
+                return obj.__class__, tuple([c.name for c in
+                        obj.columns]), getattr(obj, 'unique', None)
+
+            # find what the primary k constraint name should be
+            primaryconsname = engine.execute(
+               text("""SELECT constraint_name
+                   FROM all_constraints
+                   WHERE table_name = :table_name
+                   AND owner = :owner
+                   AND constraint_type = 'P' """),
+               table_name=table.name.upper(),
+               owner=engine.url.username.upper()).fetchall()[0][0]
+
+            reflectedtable = inspect.tables[table.name]
+
+            # make a dictionary of the reflected objects:
+
+            reflected = dict([(obj_definition(i), i) for i in
+                             reflectedtable.indexes
+                             | reflectedtable.constraints])
+
+            # assert we got primary key constraint and its name, Error
+            # if not in dict
+
+            assert reflected[(PrimaryKeyConstraint, ('id_a', 'id_b',
+                             'group'), None)].name.upper() \
+                == primaryconsname.upper()
+
+            # Error if not in dict
+
+            assert reflected[(Index, ('id_b', 'group'), False)].name \
+                == normalind.name
+            assert (Index, ('id_b', ), True) in reflected
+            assert (Index, ('col', 'group'), True) in reflected
+            assert len(reflectedtable.constraints) == 1
+            assert len(reflectedtable.indexes) == 3
+
+        finally:
+            metadata.drop_all()
+
+
         
 class SequenceTest(TestBase, AssertsCompiledSQL):
 
diff --git a/test/dialect/test_postgresql.py b/test/dialect/test_postgresql.py
index a605594d4..9ad46c189 100644
--- a/test/dialect/test_postgresql.py
+++ b/test/dialect/test_postgresql.py
@@ -1725,7 +1725,7 @@ class ServerSideCursorsTest(TestBase, AssertsExecutionResults):
             result.close()
             result = \
                 sess.query(Foo).execution_options(stream_results=True).\
-                    subquery().execute()
+                    statement.execute()
             assert result.cursor.name
             result.close()
         finally:
diff --git a/test/engine/test_metadata.py b/test/engine/test_metadata.py
index 7ea753621..528d56244 100644
--- a/test/engine/test_metadata.py
+++ b/test/engine/test_metadata.py
@@ -1,8 +1,11 @@
-from sqlalchemy.test.testing import assert_raises, assert_raises_message
+from sqlalchemy.test.testing import assert_raises
+from sqlalchemy.test.testing import assert_raises_message
+from sqlalchemy.test.testing import emits_warning
+
 import pickle
 from sqlalchemy import Integer, String, UniqueConstraint, \
     CheckConstraint, ForeignKey, MetaData, Sequence, \
-    ForeignKeyConstraint, ColumnDefault
+    ForeignKeyConstraint, ColumnDefault, Index
 from sqlalchemy.test.schema import Table, Column
 from sqlalchemy import schema, exc
 import sqlalchemy as tsa
@@ -246,6 +249,62 @@ class MetaDataTest(TestBase, ComparesTables):
         eq_(str(table_c.join(table2_c).onclause),
             'someschema.mytable.myid = someschema.othertable.myid')
 
+    def test_tometadata_kwargs(self):
+        meta = MetaData()
+
+        table = Table('mytable', meta,
+            Column('myid', Integer, primary_key=True),
+            mysql_engine='InnoDB',
+        )
+
+        meta2 = MetaData()
+        table_c = table.tometadata(meta2)
+
+        eq_(table.kwargs,table_c.kwargs)
+
+    def test_tometadata_indexes(self):
+        meta = MetaData()
+
+        table = Table('mytable', meta,
+            Column('id', Integer, primary_key=True),
+            Column('data1', Integer, index=True),
+            Column('data2', Integer),
+        )
+        Index('multi',table.c.data1,table.c.data2),
+        
+        meta2 = MetaData()
+        table_c = table.tometadata(meta2)
+
+        def _get_key(i):
+            return [i.name,i.unique] + \
+                    sorted(i.kwargs.items()) + \
+                    i.columns.keys()
+        
+        eq_(
+            sorted([_get_key(i) for i in table.indexes]),
+            sorted([_get_key(i) for i in table_c.indexes])
+        )
+
+    @emits_warning("Table '.+' already exists within the given MetaData")
+    def test_tometadata_already_there(self):
+        
+        meta1 = MetaData()
+        table1 = Table('mytable', meta1,
+            Column('myid', Integer, primary_key=True),
+        )
+        meta2 = MetaData()
+        table2 = Table('mytable', meta2,
+            Column('yourid', Integer, primary_key=True),
+        )
+
+        meta3 = MetaData()
+        
+        table_c = table1.tometadata(meta2)
+        table_d = table2.tometadata(meta2)
+
+        # d'oh!
+        assert table_c is table_d
+
     def test_tometadata_default_schema(self):
         meta = MetaData()
 
diff --git a/test/ext/test_declarative.py b/test/ext/test_declarative.py
index 71e31233b..65ec92ca2 100644
--- a/test/ext/test_declarative.py
+++ b/test/ext/test_declarative.py
@@ -369,11 +369,14 @@ class DeclarativeTest(DeclarativeTestBase):
 
         hasattr(User.addresses, 'property')
 
-        # the exeption is preserved
-
-        assert_raises_message(sa.exc.InvalidRequestError,
-                              r"suppressed within a hasattr\(\)",
-                              compile_mappers)
+        # the exception is preserved.  Remains the 
+        # same through repeated calls.
+        for i in range(3):
+            assert_raises_message(sa.exc.InvalidRequestError,
+                            "^One or more mappers failed to initialize - "
+                            "can't proceed with initialization of other "
+                            "mappers.  Original exception was: When initializing.*",
+                            compile_mappers)
 
     def test_custom_base(self):
         class MyBase(object):
@@ -1316,7 +1319,42 @@ class DeclarativeInheritanceTest(DeclarativeTestBase):
             primary_language = Column('primary_language', String(50))
 
         assert class_mapper(Engineer).inherits is class_mapper(Person)
+    
+    @testing.fails_if(lambda: True, "Not implemented until 0.7")
+    def test_foreign_keys_with_col(self):
+        """Test that foreign keys that reference a literal 'id' subclass 
+        'id' attribute behave intuitively.  
+        
+        See ticket 1892.
+        
+        """
+        class Booking(Base):
+            __tablename__ = 'booking'
+            id = Column(Integer, primary_key=True)
+
+        class PlanBooking(Booking):
+            __tablename__ = 'plan_booking'
+            id = Column(Integer, ForeignKey(Booking.id),
+                            primary_key=True)
+
+        # referencing PlanBooking.id gives us the column
+        # on plan_booking, not booking
+        class FeatureBooking(Booking):
+            __tablename__ = 'feature_booking'
+            id = Column(Integer, ForeignKey(Booking.id),
+                                        primary_key=True)
+            plan_booking_id = Column(Integer,
+                                ForeignKey(PlanBooking.id))
+            
+            plan_booking = relationship(PlanBooking,
+                        backref='feature_bookings')
+        
+        assert FeatureBooking.__table__.c.plan_booking_id.\
+                    references(PlanBooking.__table__.c.id)
 
+        assert FeatureBooking.__table__.c.id.\
+                    references(Booking.__table__.c.id)
+      
     def test_with_undefined_foreignkey(self):
 
         class Parent(Base):
diff --git a/test/orm/inheritance/test_basic.py b/test/orm/inheritance/test_basic.py
index b4aaf13ba..c6dec16b7 100644
--- a/test/orm/inheritance/test_basic.py
+++ b/test/orm/inheritance/test_basic.py
@@ -981,7 +981,8 @@ class OverrideColKeyTest(_base.MappedTest):
         # s2 gets a new id, base_id is overwritten by the ultimate
         # PK col
         assert s2.id == s2.base_id != 15
-        
+    
+    @testing.emits_warning(r'Implicit')
     def test_override_implicit(self):
         # this is how the pattern looks intuitively when 
         # using declarative.
@@ -1143,7 +1144,9 @@ class OptimizedLoadTest(_base.MappedTest):
         
         # redefine Sub's "id" to favor the "id" col in the subtable.
         # "id" is also part of the primary join condition
-        mapper(Sub, sub, inherits=Base, polymorphic_identity='sub', properties={'id':sub.c.id})
+        mapper(Sub, sub, inherits=Base, 
+                        polymorphic_identity='sub',
+                        properties={'id':[sub.c.id, base.c.id]})
         sess = sessionmaker()()
         s1 = Sub(data='s1data', sub='s1sub')
         sess.add(s1)
diff --git a/test/orm/inheritance/test_single.py b/test/orm/inheritance/test_single.py
index 4b7078eb5..2efde2b32 100644
--- a/test/orm/inheritance/test_single.py
+++ b/test/orm/inheritance/test_single.py
@@ -8,7 +8,7 @@ from test.orm._base import MappedTest, ComparableEntity
 from sqlalchemy.test.schema import Table, Column
 
 
-class SingleInheritanceTest(MappedTest):
+class SingleInheritanceTest(testing.AssertsCompiledSQL, MappedTest):
     @classmethod
     def define_tables(cls, metadata):
         Table('employees', metadata,
@@ -26,6 +26,7 @@ class SingleInheritanceTest(MappedTest):
 
     @classmethod
     def setup_classes(cls):
+        global Employee, Manager, Engineer, JuniorEngineer
         class Employee(ComparableEntity):
             pass
         class Manager(Employee):
@@ -114,6 +115,31 @@ class SingleInheritanceTest(MappedTest):
         #    session.query(Employee.name, Manager.manager_data, Engineer.engineer_info).all(), 
         #    []
         # )
+
+    @testing.resolve_artifact_names
+    def test_from_self(self):
+        sess = create_session()
+        self.assert_compile(sess.query(Engineer).from_self(),
+                            'SELECT anon_1.employees_employee_id AS '
+                            'anon_1_employees_employee_id, '
+                            'anon_1.employees_name AS '
+                            'anon_1_employees_name, '
+                            'anon_1.employees_manager_data AS '
+                            'anon_1_employees_manager_data, '
+                            'anon_1.employees_engineer_info AS '
+                            'anon_1_employees_engineer_info, '
+                            'anon_1.employees_type AS '
+                            'anon_1_employees_type FROM (SELECT '
+                            'employees.employee_id AS '
+                            'employees_employee_id, employees.name AS '
+                            'employees_name, employees.manager_data AS '
+                            'employees_manager_data, '
+                            'employees.engineer_info AS '
+                            'employees_engineer_info, employees.type '
+                            'AS employees_type FROM employees) AS '
+                            'anon_1 WHERE anon_1.employees_type IN '
+                            '(:type_1, :type_2)',
+                            use_default_dialect=True)
         
     @testing.resolve_artifact_names
     def test_select_from(self):
@@ -182,6 +208,56 @@ class SingleInheritanceTest(MappedTest):
         assert len(rq.join(Report.employee.of_type(Manager)).all()) == 1
         assert len(rq.join(Report.employee.of_type(Engineer)).all()) == 0
 
+class RelationshipFromSingleTest(testing.AssertsCompiledSQL, MappedTest):
+    @classmethod
+    def define_tables(cls, metadata):
+        Table('employee', metadata,
+            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
+            Column('name', String(50)),
+            Column('type', String(20)),
+        )
+        
+        Table('employee_stuff', metadata,
+            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
+            Column('employee_id', Integer, ForeignKey('employee.id')),
+            Column('name', String(50)),
+        )
+    
+    @classmethod
+    def setup_classes(cls):
+        class Employee(ComparableEntity):
+            pass
+        class Manager(Employee):
+            pass
+        class Stuff(ComparableEntity):
+            pass
+
+    @testing.resolve_artifact_names
+    def test_subquery_load(self):
+        mapper(Employee, employee, polymorphic_on=employee.c.type, polymorphic_identity='employee')
+        mapper(Manager, inherits=Employee, polymorphic_identity='manager', properties={
+            'stuff':relationship(Stuff)
+        })
+        mapper(Stuff, employee_stuff)
+        
+        sess = create_session()
+        context = sess.query(Manager).options(subqueryload('stuff'))._compile_context()
+        subq = context.attributes[('subquery', (class_mapper(Employee), 'stuff'))]
+
+        self.assert_compile(subq,
+                            'SELECT employee_stuff.id AS '
+                            'employee_stuff_id, employee_stuff.employee'
+                            '_id AS employee_stuff_employee_id, '
+                            'employee_stuff.name AS '
+                            'employee_stuff_name, anon_1.employee_id '
+                            'AS anon_1_employee_id FROM (SELECT '
+                            'employee.id AS employee_id FROM employee '
+                            'WHERE employee.type IN (:type_1)) AS anon_1 '
+                            'JOIN employee_stuff ON anon_1.employee_id '
+                            '= employee_stuff.employee_id ORDER BY '
+                            'anon_1.employee_id',
+                            use_default_dialect=True
+                            )
 
 class RelationshipToSingleTest(MappedTest):
     @classmethod
diff --git a/test/orm/test_assorted_eager.py b/test/orm/test_assorted_eager.py
index 20736b8fe..0e389b74b 100644
--- a/test/orm/test_assorted_eager.py
+++ b/test/orm/test_assorted_eager.py
@@ -324,7 +324,7 @@ class EagerTest3(_base.MappedTest):
         arb_data = sa.select(
             [stats.c.data_id, sa.func.max(stats.c.somedata).label('max')],
             stats.c.data_id <= 5,
-            group_by=[stats.c.data_id]).alias('arb')
+            group_by=[stats.c.data_id])
 
         arb_result = arb_data.execute().fetchall()
 
@@ -334,6 +334,8 @@ class EagerTest3(_base.MappedTest):
         # extract just the "data_id" from it
         arb_result = [row['data_id'] for row in arb_result]
 
+        arb_data = arb_data.alias('arb')
+        
         # now query for Data objects using that above select, adding the
         # "order by max desc" separately
         q = (session.query(Data).
diff --git a/test/orm/test_attributes.py b/test/orm/test_attributes.py
index 3a8a320e3..742e9d874 100644
--- a/test/orm/test_attributes.py
+++ b/test/orm/test_attributes.py
@@ -693,14 +693,18 @@ class UtilTest(_base.ORMTest):
 
 class BackrefTest(_base.ORMTest):
 
-    def test_manytomany(self):
+    def test_m2m(self):
         class Student(object):pass
         class Course(object):pass
 
         attributes.register_class(Student)
         attributes.register_class(Course)
-        attributes.register_attribute(Student, 'courses', uselist=True, extension=attributes.GenericBackrefExtension('students'), useobject=True)
-        attributes.register_attribute(Course, 'students', uselist=True, extension=attributes.GenericBackrefExtension('courses'), useobject=True)
+        attributes.register_attribute(Student, 'courses', uselist=True,
+                extension=attributes.GenericBackrefExtension('students'
+                ), useobject=True)
+        attributes.register_attribute(Course, 'students', uselist=True,
+                extension=attributes.GenericBackrefExtension('courses'
+                ), useobject=True)
 
         s = Student()
         c = Course()
@@ -717,14 +721,18 @@ class BackrefTest(_base.ORMTest):
         s1.courses.remove(c)
         self.assert_(c.students == [s2,s3])
 
-    def test_onetomany(self):
+    def test_o2m(self):
         class Post(object):pass
         class Blog(object):pass
 
         attributes.register_class(Post)
         attributes.register_class(Blog)
-        attributes.register_attribute(Post, 'blog', uselist=False, extension=attributes.GenericBackrefExtension('posts'), trackparent=True, useobject=True)
-        attributes.register_attribute(Blog, 'posts', uselist=True, extension=attributes.GenericBackrefExtension('blog'), trackparent=True, useobject=True)
+        attributes.register_attribute(Post, 'blog', uselist=False,
+                extension=attributes.GenericBackrefExtension('posts'),
+                trackparent=True, useobject=True)
+        attributes.register_attribute(Blog, 'posts', uselist=True,
+                extension=attributes.GenericBackrefExtension('blog'),
+                trackparent=True, useobject=True)
         b = Blog()
         (p1, p2, p3) = (Post(), Post(), Post())
         b.posts.append(p1)
@@ -748,13 +756,17 @@ class BackrefTest(_base.ORMTest):
         p5.blog = None
         del p5.blog
 
-    def test_onetoone(self):
+    def test_o2o(self):
         class Port(object):pass
         class Jack(object):pass
         attributes.register_class(Port)
         attributes.register_class(Jack)
-        attributes.register_attribute(Port, 'jack', uselist=False, extension=attributes.GenericBackrefExtension('port'), useobject=True)
-        attributes.register_attribute(Jack, 'port', uselist=False, extension=attributes.GenericBackrefExtension('jack'), useobject=True)
+        attributes.register_attribute(Port, 'jack', uselist=False,
+                extension=attributes.GenericBackrefExtension('port'),
+                useobject=True)
+        attributes.register_attribute(Jack, 'port', uselist=False,
+                extension=attributes.GenericBackrefExtension('jack'),
+                useobject=True)
         p = Port()
         j = Jack()
         p.jack = j
@@ -764,6 +776,96 @@ class BackrefTest(_base.ORMTest):
         j.port = None
         self.assert_(p.jack is None)
 
+    def test_symmetric_o2o_inheritance(self):
+        """Test that backref 'initiator' catching goes against
+        a token that is global to all InstrumentedAttribute objects
+        within a particular class, not just the indvidual IA object
+        since we use distinct objects in an inheritance scenario.
+        
+        """
+        class Parent(object):
+            pass
+        class Child(object):
+            pass
+        class SubChild(Child):
+            pass
+
+        p_token = object()
+        c_token = object()
+        
+        attributes.register_class(Parent)
+        attributes.register_class(Child)
+        attributes.register_class(SubChild)
+        attributes.register_attribute(Parent, 'child', uselist=False,
+                extension=attributes.GenericBackrefExtension('parent'),
+                parent_token = p_token,
+                useobject=True)
+        attributes.register_attribute(Child, 'parent', uselist=False,
+                extension=attributes.GenericBackrefExtension('child'),
+                parent_token = c_token,
+                useobject=True)
+        attributes.register_attribute(SubChild, 'parent',
+                uselist=False,
+                extension=attributes.GenericBackrefExtension('child'),
+                parent_token = c_token,
+                useobject=True)
+        
+        p1 = Parent()
+        c1 = Child()
+        p1.child = c1
+        
+        c2 = SubChild()
+        c2.parent = p1
+
+    def test_symmetric_o2m_inheritance(self):
+        class Parent(object):
+            pass
+        class SubParent(Parent):
+            pass
+        class Child(object):
+            pass
+
+        p_token = object()
+        c_token = object()
+        
+        attributes.register_class(Parent)
+        attributes.register_class(SubParent)
+        attributes.register_class(Child)
+        attributes.register_attribute(Parent, 'children', uselist=True,
+                extension=attributes.GenericBackrefExtension('parent'),
+                parent_token = p_token,
+                useobject=True)
+        attributes.register_attribute(SubParent, 'children', uselist=True,
+                extension=attributes.GenericBackrefExtension('parent'),
+                parent_token = p_token,
+                useobject=True)
+        attributes.register_attribute(Child, 'parent', uselist=False,
+                extension=attributes.GenericBackrefExtension('children'),
+                parent_token = c_token,
+                useobject=True)
+        
+        p1 = Parent()
+        p2 = SubParent()
+        c1 = Child()
+        
+        p1.children.append(c1)
+
+        assert c1.parent is p1
+        assert c1 in p1.children
+        
+        p2.children.append(c1)
+        assert c1.parent is p2
+        
+        # note its still in p1.children -
+        # the event model currently allows only
+        # one level deep.  without the parent_token,
+        # it keeps going until a ValueError is raised
+        # and this condition changes.
+        assert c1 in p1.children
+        
+        
+        
+        
 class PendingBackrefTest(_base.ORMTest):
     def setup(self):
         global Post, Blog, called, lazy_load
diff --git a/test/orm/test_dynamic.py b/test/orm/test_dynamic.py
index 5d822fa3d..c06f6918a 100644
--- a/test/orm/test_dynamic.py
+++ b/test/orm/test_dynamic.py
@@ -21,6 +21,7 @@ class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL):
         q = sess.query(User)
 
         u = q.filter(User.id==7).first()
+        
         eq_([User(id=7,
                   addresses=[Address(id=1, email_address='jack@bean.com')])],
             q.filter(User.id==7).all())
diff --git a/test/orm/test_load_on_fks.py b/test/orm/test_load_on_fks.py
new file mode 100644
index 000000000..8e4f53b0d
--- /dev/null
+++ b/test/orm/test_load_on_fks.py
@@ -0,0 +1,273 @@
+from sqlalchemy import *
+from sqlalchemy.orm import *
+
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.test.testing import TestBase, eq_, AssertsExecutionResults, assert_raises
+from sqlalchemy.test import testing
+from sqlalchemy.orm.attributes import instance_state
+from sqlalchemy.orm.exc import FlushError
+from sqlalchemy.test.schema import Table, Column
+
+engine = testing.db
+
+
+class FlushOnPendingTest(AssertsExecutionResults, TestBase):
+    def setUp(self):
+        global Parent, Child, Base
+        Base= declarative_base()
+        
+        class Parent(Base):
+            __tablename__ = 'parent'
+    
+            id= Column(Integer, primary_key=True, test_needs_autoincrement=True)
+            name = Column(String(50), nullable=False)
+            children = relationship("Child", load_on_pending=True)
+            
+        class Child(Base):
+            __tablename__ = 'child'
+            id= Column(Integer, primary_key=True, test_needs_autoincrement=True)
+            parent_id = Column(Integer, ForeignKey('parent.id'))
+    
+        Base.metadata.create_all(engine)
+
+    def tearDown(self):
+        Base.metadata.drop_all(engine)
+
+    def test_annoying_autoflush_one(self):
+        sess = Session(engine)
+        
+        p1 = Parent()
+        sess.add(p1)
+        p1.children = []
+
+    def test_annoying_autoflush_two(self):
+        sess = Session(engine)
+        
+        p1 = Parent()
+        sess.add(p1)
+        assert p1.children == []
+
+    def test_dont_load_if_no_keys(self):
+        sess = Session(engine)
+        
+        p1 = Parent()
+        sess.add(p1)
+        
+        def go():
+            assert p1.children == []
+        self.assert_sql_count(testing.db, go, 0)
+
+class LoadOnFKsTest(AssertsExecutionResults, TestBase):
+    
+    def setUp(self):
+        global Parent, Child, Base
+        Base= declarative_base()
+        
+        class Parent(Base):
+            __tablename__ = 'parent'
+            __table_args__ = {'mysql_engine':'InnoDB'}
+            
+            id= Column(Integer, primary_key=True, test_needs_autoincrement=True)
+
+        class Child(Base):
+            __tablename__ = 'child'
+            __table_args__ = {'mysql_engine':'InnoDB'}
+
+            id= Column(Integer, primary_key=True, test_needs_autoincrement=True)
+            parent_id = Column(Integer, ForeignKey('parent.id'))
+    
+            parent = relationship(Parent, backref=backref("children"))
+    
+        Base.metadata.create_all(engine)
+
+        global sess, p1, p2, c1, c2
+        sess = Session(bind=engine)
+
+        p1 = Parent()
+        p2 = Parent()
+        c1, c2 = Child(), Child()
+        c1.parent = p1
+        sess.add_all([p1, p2])
+        assert c1 in sess
+
+        sess.commit()
+    
+    def tearDown(self):
+        sess.rollback()
+        Base.metadata.drop_all(engine)
+
+    def test_load_on_pending_disallows_backref_event(self):
+        Child.parent.property.load_on_pending = True
+        sess.autoflush = False
+        c3 = Child()
+        sess.add(c3)
+        c3.parent_id = p1.id
+        c3.parent = p1
+        
+        # a side effect of load-on-pending with no autoflush.
+        # a change to the backref event handler to check
+        # collection membership before assuming "old == new so return"
+        # would fix this - but this is wasteful and autoflush
+        # should be turned on.
+        assert c3 not in p1.children
+
+    def test_load_on_persistent_allows_backref_event(self):
+        Child.parent.property.load_on_pending = True
+        c3 = Child()
+        sess.add(c3)
+        c3.parent_id = p1.id
+        c3.parent = p1
+        
+        assert c3 in p1.children
+
+    def test_no_load_on_pending_allows_backref_event(self):
+        # users who stick with the program and don't use
+        # 'load_on_pending' get expected behavior
+        
+        sess.autoflush = False
+        c3 = Child()
+        sess.add(c3)
+        c3.parent_id = p1.id
+
+        c3.parent = p1
+        
+        assert c3 in p1.children
+    
+    def test_autoflush_on_pending(self):
+        c3 = Child()
+        sess.add(c3)
+        c3.parent_id = p1.id
+        
+        # pendings don't autoflush
+        assert c3.parent is None
+
+    def test_autoflush_on_pending(self):
+        Child.parent.property.load_on_pending = True
+        c3 = Child()
+        sess.add(c3)
+        c3.parent_id = p1.id
+        
+        # ...unless the flag is on
+        assert c3.parent is p1
+        
+    def test_load_on_pending_with_set(self):
+        Child.parent.property.load_on_pending = True
+
+        p1.children
+
+        c3 = Child()
+        sess.add(c3)
+        
+        c3.parent_id = p1.id
+
+        def go():
+            c3.parent = p1
+        self.assert_sql_count(testing.db, go, 0)
+        
+    def test_backref_doesnt_double(self):
+        Child.parent.property.load_on_pending = True
+        sess.autoflush = False
+        p1.children
+        c3 = Child()
+        sess.add(c3)
+        c3.parent = p1
+        c3.parent = p1
+        c3.parent = p1
+        c3.parent = p1
+        assert len(p1.children)== 2
+        
+    def test_m2o_lazy_loader_on_persistent(self):
+        """Compare the behaviors from the lazyloader using
+        the "committed" state in all cases, vs. the lazyloader
+        using the "current" state in all cases except during flush.
+        
+        """
+        for loadfk in (True, False):
+            for loadrel in (True, False):
+                for autoflush in (True, False):
+                    for manualflush in (True, False):
+                        for fake_autoexpire in (True, False):
+                            sess.autoflush = autoflush
+                        
+                            if loadfk:
+                                c1.parent_id
+                            if loadrel:
+                                c1.parent
+
+                            c1.parent_id = p2.id
+                    
+                            if manualflush:
+                                sess.flush()
+                        
+                            # fake_autoexpire refers to the eventual
+                            # auto-expire of 'parent' when c1.parent_id
+                            # is altered.
+                            if fake_autoexpire:
+                                sess.expire(c1, ['parent'])
+                            
+                            # old 0.6 behavior
+                            #if manualflush and (not loadrel or fake_autoexpire):
+                            #    # a flush occurs, we get p2
+                            #    assert c1.parent is p2
+                            #elif not loadrel and not loadfk: 
+                            #    # problematically - we get None since committed state
+                            #    # is empty when c1.parent_id was mutated, since we want
+                            #    # to save on selects.  this is
+                            #    # why the patch goes in in 0.6 - this is mostly a bug.
+                            #    assert c1.parent is None
+                            #else:
+                            #    # if things were loaded, autoflush doesn't even
+                            #    # happen.
+                            #    assert c1.parent is p1
+                            
+                            # new behavior
+                            if loadrel and not fake_autoexpire:
+                                assert c1.parent is p1
+                            else:
+                                assert c1.parent is p2
+                                
+                            sess.rollback()
+                    
+    def test_m2o_lazy_loader_on_pending(self):
+        for loadonpending in (False, True):
+            for autoflush in (False, True):
+                for manualflush in (False, True):
+                    Child.parent.property.load_on_pending = loadonpending
+                    sess.autoflush = autoflush
+                    c2 = Child()
+                    sess.add(c2)
+                    c2.parent_id = p2.id
+                
+                    if manualflush:
+                       sess.flush()
+                
+                    if loadonpending or manualflush:
+                        assert c2.parent is p2
+                    else:
+                        assert c2.parent is None
+                
+                    sess.rollback()
+
+    def test_m2o_lazy_loader_on_transient(self):
+        for loadonpending in (False, True):
+            for attach in (False, True):
+                for autoflush in (False, True):
+                    for manualflush in (False, True):
+                        Child.parent.property.load_on_pending = loadonpending
+                        sess.autoflush = autoflush
+                        c2 = Child()
+                    
+                        if attach:
+                            sess._attach(instance_state(c2))
+
+                        c2.parent_id = p2.id
+                
+                        if manualflush:
+                           sess.flush()
+                        
+                        if loadonpending and attach:
+                            assert c2.parent is p2
+                        else:
+                            assert c2.parent is None
+                
+                        sess.rollback()
diff --git a/test/orm/test_mapper.py b/test/orm/test_mapper.py
index e24906a1f..f041c8896 100644
--- a/test/orm/test_mapper.py
+++ b/test/orm/test_mapper.py
@@ -88,14 +88,25 @@ class MapperTest(_fixtures.FixtureTest):
 
     @testing.resolve_artifact_names
     def test_exceptions_sticky(self):
-        """test preservation of mapper compile errors raised during hasattr()."""
+        """test preservation of mapper compile errors raised during hasattr(),
+        as well as for redundant mapper compile calls.  Test that 
+        repeated calls don't stack up error messages.
+        
+        """
         
         mapper(Address, addresses, properties={
             'user':relationship(User)
         })
         
         hasattr(Address.user, 'property')
-        assert_raises_message(sa.exc.InvalidRequestError, r"suppressed within a hasattr\(\)", compile_mappers)
+        for i in range(3):
+            assert_raises_message(sa.exc.InvalidRequestError,
+                                  "^One or more mappers failed to "
+                                  "initialize - can't proceed with "
+                                  "initialization of other mappers.  "
+                                  "Original exception was: Class "
+                                  "'test.orm._fixtures.User' is not mapped$"
+                                  , compile_mappers)
     
     @testing.resolve_artifact_names
     def test_column_prefix(self):
@@ -157,13 +168,15 @@ class MapperTest(_fixtures.FixtureTest):
 
     @testing.resolve_artifact_names
     def test_column_not_present(self):
-        assert_raises_message(sa.exc.ArgumentError, "not represented in the mapper's table", mapper, User, users, properties={
-            'foo':addresses.c.user_id
-        })
+        assert_raises_message(sa.exc.ArgumentError,
+                              "not represented in the mapper's table",
+                              mapper, User, users, properties={'foo'
+                              : addresses.c.user_id})
         
     @testing.resolve_artifact_names
     def test_bad_constructor(self):
         """If the construction of a mapped class fails, the instance does not get placed in the session"""
+        
         class Foo(object):
             def __init__(self, one, two, _sa_session=None):
                 pass
@@ -482,18 +495,22 @@ class MapperTest(_fixtures.FixtureTest):
         class Manager(Employee): pass
         class Hoho(object): pass
         class Lala(object): pass
-
+        class Fub(object):pass
+        class Frob(object):pass
         class HasDef(object):
             def name(self):
                 pass
-            
+        class Empty(object):pass
+        
+        empty = mapper(Empty, t, properties={'empty_id' : t.c.id}, 
+                       include_properties=[])
         p_m = mapper(Person, t, polymorphic_on=t.c.type,
                      include_properties=('id', 'type', 'name'))
-        e_m = mapper(Employee, inherits=p_m, polymorphic_identity='employee',
-          properties={
-            'boss': relationship(Manager, backref=backref('peon', ), remote_side=t.c.id)
-          },
-          exclude_properties=('vendor_id',))
+        e_m = mapper(Employee, inherits=p_m,
+                     polymorphic_identity='employee', properties={'boss'
+                     : relationship(Manager, backref=backref('peon'),
+                     remote_side=t.c.id)},
+                     exclude_properties=('vendor_id', ))
 
         m_m = mapper(Manager, inherits=e_m, polymorphic_identity='manager',
                      include_properties=('id', 'type'))
@@ -506,8 +523,12 @@ class MapperTest(_fixtures.FixtureTest):
 
         hd_m = mapper(HasDef, t, column_prefix="h_")
         
+        fb_m = mapper(Fub, t, include_properties=(t.c.id, t.c.type))
+        frb_m = mapper(Frob, t, column_prefix='f_',
+                       exclude_properties=(t.c.boss_id,
+                       'employee_number', t.c.vendor_id))
+        
         p_m.compile()
-        #sa.orm.compile_mappers()
 
         def assert_props(cls, want):
             have = set([n for n in dir(cls) if not n.startswith('_')])
@@ -519,7 +540,8 @@ class MapperTest(_fixtures.FixtureTest):
             want = set(want)
             eq_(have, want)
         
-        assert_props(HasDef, ['h_boss_id', 'h_employee_number', 'h_id', 'name', 'h_name', 'h_vendor_id', 'h_type'])    
+        assert_props(HasDef, ['h_boss_id', 'h_employee_number', 'h_id', 
+                                'name', 'h_name', 'h_vendor_id', 'h_type'])    
         assert_props(Person, ['id', 'name', 'type'])
         assert_instrumented(Person, ['id', 'name', 'type'])
         assert_props(Employee, ['boss', 'boss_id', 'employee_number',
@@ -535,24 +557,58 @@ class MapperTest(_fixtures.FixtureTest):
         assert_props(Vendor, ['vendor_id', 'id', 'name', 'type'])
         assert_props(Hoho, ['id', 'name', 'type'])
         assert_props(Lala, ['p_employee_number', 'p_id', 'p_name', 'p_type'])
-
+        assert_props(Fub, ['id', 'type'])
+        assert_props(Frob, ['f_id', 'f_type', 'f_name', ])
         # excluding the discriminator column is currently not allowed
         class Foo(Person):
             pass
-        assert_raises(sa.exc.InvalidRequestError, mapper, Foo, inherits=Person, polymorphic_identity='foo', exclude_properties=('type',) )
+        assert_props(Empty, ['empty_id'])
+
+        assert_raises(
+            sa.exc.InvalidRequestError,
+            mapper,
+            Foo, inherits=Person, polymorphic_identity='foo',
+            exclude_properties=('type', ),
+            )
     
     @testing.resolve_artifact_names
-    def test_mapping_to_join(self):
-        """Mapping to a join"""
+    def test_mapping_to_join_raises(self):
+        """Test implicit merging of two cols warns."""
+        
         usersaddresses = sa.join(users, addresses,
                                  users.c.id == addresses.c.user_id)
-        mapper(User, usersaddresses, primary_key=[users.c.id])
+        assert_raises_message(
+            sa.exc.SAWarning,
+            "Implicitly",
+            mapper, User, usersaddresses, primary_key=[users.c.id]
+        )
+        sa.orm.clear_mappers()
+        
+        @testing.emits_warning(r'Implicitly')
+        def go():
+            # but it works despite the warning
+            mapper(User, usersaddresses, primary_key=[users.c.id])
+            l = create_session().query(User).order_by(users.c.id).all()
+            eq_(l, self.static.user_result[:3])
+        go()
+
+    @testing.resolve_artifact_names
+    def test_mapping_to_join(self):
+        """Mapping to a join"""
+
+        usersaddresses = sa.join(users, addresses, users.c.id
+                                 == addresses.c.user_id)
+        mapper(User, usersaddresses, primary_key=[users.c.id],
+               exclude_properties=[addresses.c.id])
         l = create_session().query(User).order_by(users.c.id).all()
         eq_(l, self.static.user_result[:3])
 
     @testing.resolve_artifact_names
     def test_mapping_to_join_no_pk(self):
-        m = mapper(Address, addresses.join(email_bounces))
+        m = mapper(Address, 
+                    addresses.join(email_bounces), 
+                    properties={'id':[addresses.c.id, email_bounces.c.id]}
+                )
         m.compile()
         assert addresses in m._pks_by_table
         assert email_bounces not in m._pks_by_table
diff --git a/test/orm/test_naturalpks.py b/test/orm/test_naturalpks.py
index b305375da..d02ecb707 100644
--- a/test/orm/test_naturalpks.py
+++ b/test/orm/test_naturalpks.py
@@ -8,7 +8,7 @@ import sqlalchemy as sa
 from sqlalchemy.test import testing
 from sqlalchemy import Integer, String, ForeignKey, Unicode
 from sqlalchemy.test.schema import Table, Column
-from sqlalchemy.orm import mapper, relationship, create_session, backref
+from sqlalchemy.orm import mapper, relationship, create_session, backref, Session
 from sqlalchemy.orm.session import make_transient
 from sqlalchemy.test.testing import eq_
 from test.orm import _base, _fixtures
@@ -499,24 +499,49 @@ class SelfReferentialTest(_base.MappedTest):
             pass
 
     @testing.resolve_artifact_names
-    def test_one_to_many(self):
+    def test_one_to_many_on_m2o(self):
         mapper(Node, nodes, properties={
             'children': relationship(Node,
                                  backref=sa.orm.backref('parentnode',
                                             remote_side=nodes.c.name,
                                             passive_updates=False),
-                                 passive_updates=False)})
+                                 )})
 
-        sess = create_session()
+        sess = Session()
+        n1 = Node(name='n1')
+        sess.add(n1)
+        n2 = Node(name='n11', parentnode=n1)
+        n3 = Node(name='n12', parentnode=n1)
+        n4 = Node(name='n13', parentnode=n1)
+        sess.add_all([n2, n3, n4])
+        sess.commit()
+        
+        n1.name = 'new n1'
+        sess.commit()
+        eq_(['new n1', 'new n1', 'new n1'],
+            [n.parent
+             for n in sess.query(Node).filter(
+                 Node.name.in_(['n11', 'n12', 'n13']))])
+
+    @testing.resolve_artifact_names
+    def test_one_to_many_on_o2m(self):
+        mapper(Node, nodes, properties={
+            'children': relationship(Node,
+                                 backref=sa.orm.backref('parentnode',
+                                            remote_side=nodes.c.name),
+                                passive_updates=False
+                                 )})
+
+        sess = Session()
         n1 = Node(name='n1')
         n1.children.append(Node(name='n11'))
         n1.children.append(Node(name='n12'))
         n1.children.append(Node(name='n13'))
         sess.add(n1)
-        sess.flush()
+        sess.commit()
 
         n1.name = 'new n1'
-        sess.flush()
+        sess.commit()
         eq_(n1.children[1].parent, 'new n1')
         eq_(['new n1', 'new n1', 'new n1'],
             [n.parent
@@ -540,18 +565,16 @@ class SelfReferentialTest(_base.MappedTest):
             }
         )
 
-        sess = create_session()
+        sess = Session()
         n1 = Node(name='n1')
         n11 = Node(name='n11', parentnode=n1)
         n12 = Node(name='n12', parentnode=n1)
         n13 = Node(name='n13', parentnode=n1)
         sess.add_all([n1, n11, n12, n13])
-        sess.flush()
+        sess.commit()
 
         n1.name = 'new n1'
-        sess.flush()
-        if passive:
-            sess.expire_all()
+        sess.commit()
         eq_(['new n1', 'new n1', 'new n1'],
              [n.parent
               for n in sess.query(Node).filter(
diff --git a/test/orm/test_query.py b/test/orm/test_query.py
index 65be1e00a..8facd4e69 100644
--- a/test/orm/test_query.py
+++ b/test/orm/test_query.py
@@ -1375,6 +1375,9 @@ class ParentTest(QueryTest):
         o = sess.query(Order).with_parent(u1, property='orders').all()
         assert [Order(description="order 1"), Order(description="order 3"), Order(description="order 5")] == o
 
+        o = sess.query(Order).with_parent(u1, property=User.orders).all()
+        assert [Order(description="order 1"), Order(description="order 3"), Order(description="order 5")] == o
+
         o = sess.query(Order).filter(with_parent(u1, User.orders)).all()
         assert [Order(description="order 1"), Order(description="order 3"), Order(description="order 5")] == o
         
@@ -1396,7 +1399,9 @@ class ParentTest(QueryTest):
             q = sess.query(Item).with_parent(u1)
             assert False
         except sa_exc.InvalidRequestError, e:
-            assert str(e) == "Could not locate a property which relates instances of class 'Item' to instances of class 'User'"
+            assert str(e) \
+                == "Could not locate a property which relates "\
+                "instances of class 'Item' to instances of class 'User'"
 
     def test_m2m(self):
         sess = create_session()
@@ -1404,6 +1409,50 @@ class ParentTest(QueryTest):
         k = sess.query(Keyword).with_parent(i1).all()
         assert [Keyword(name='red'), Keyword(name='small'), Keyword(name='square')] == k
 
+    def test_with_transient(self):
+        sess = Session()
+        
+        q = sess.query(User)
+        u1 = q.filter_by(name='jack').one()
+        utrans = User(id=u1.id)
+        o = sess.query(Order).with_parent(utrans, 'orders')
+        eq_(
+            [Order(description="order 1"), Order(description="order 3"), Order(description="order 5")],
+            o.all()
+        )
+
+        o = sess.query(Order).filter(with_parent(utrans, 'orders'))
+        eq_(
+            [Order(description="order 1"), Order(description="order 3"), Order(description="order 5")],
+            o.all()
+        )
+        
+    def test_with_pending_autoflush(self):
+        sess = Session()
+
+        o1 = sess.query(Order).first()
+        opending = Order(id=20, user_id=o1.user_id)
+        sess.add(opending)
+        eq_(
+            sess.query(User).with_parent(opending, 'user').one(),
+            User(id=o1.user_id)
+        )
+        eq_(
+            sess.query(User).filter(with_parent(opending, 'user')).one(),
+            User(id=o1.user_id)
+        )
+
+    def test_with_pending_no_autoflush(self):
+        sess = Session(autoflush=False)
+
+        o1 = sess.query(Order).first()
+        opending = Order(user_id=o1.user_id)
+        sess.add(opending)
+        eq_(
+            sess.query(User).with_parent(opending, 'user').one(),
+            User(id=o1.user_id)
+        )
+        
 class InheritedJoinTest(_base.MappedTest, AssertsCompiledSQL):
     run_setup_mappers = 'once'
     
diff --git a/test/orm/test_session.py b/test/orm/test_session.py
index 4976db131..6ac42a6b3 100644
--- a/test/orm/test_session.py
+++ b/test/orm/test_session.py
@@ -4,7 +4,7 @@ from sqlalchemy.test.util import gc_collect
 import inspect
 import pickle
 from sqlalchemy.orm import create_session, sessionmaker, attributes, \
-    make_transient
+    make_transient, Session
 from sqlalchemy.orm.attributes import instance_state
 import sqlalchemy as sa
 from sqlalchemy.test import engines, testing, config
@@ -713,12 +713,39 @@ class SessionTest(_fixtures.FixtureTest):
         sess.flush()
         sess.rollback()
         assert_raises_message(sa.exc.InvalidRequestError,
-                              'inactive due to a rollback in a '
-                              'subtransaction', sess.begin,
-                              subtransactions=True)
+                              "This Session's transaction has been "
+                              r"rolled back by a nested rollback\(\) "
+                              "call.  To begin a new transaction, "
+                              r"issue Session.rollback\(\) first.",
+                              sess.begin, subtransactions=True)
         sess.close()
 
     @testing.resolve_artifact_names
+    def test_preserve_flush_error(self):
+        mapper(User, users)
+        sess = Session()
+
+        sess.add(User(id=5))
+        assert_raises(
+            sa.exc.DBAPIError,
+            sess.commit
+        )
+        
+        for i in range(5):
+            assert_raises_message(sa.exc.InvalidRequestError,
+                              "^This Session's transaction has been "
+                              r"rolled back due to a previous exception during flush. To "
+                              "begin a new transaction with this "
+                              "Session, first issue "
+                              r"Session.rollback\(\). Original exception "
+                              "was:",
+                              sess.commit)
+        sess.rollback()
+        sess.add(User(id=5, name='some name'))
+        sess.commit()
+        
+        
+    @testing.resolve_artifact_names
     def test_no_autocommit_with_explicit_commit(self):
         mapper(User, users)
         session = create_session(autocommit=False)
@@ -1383,6 +1410,22 @@ class SessionTest(_fixtures.FixtureTest):
         assert b in sess
         assert len(list(sess)) == 1
 
+    @testing.resolve_artifact_names
+    def test_identity_map_mutate(self):
+        mapper(User, users)
+
+        sess = Session()
+        
+        sess.add_all([User(name='u1'), User(name='u2'), User(name='u3')])
+        sess.commit()
+        
+        u1, u2, u3 = sess.query(User).all()
+        for i, (key, value) in enumerate(sess.identity_map.iteritems()):
+            if i == 2:
+                del u3
+                gc_collect()
+        
+        
 class DisposedStates(_base.MappedTest):
     run_setup_mappers = 'once'
     run_inserts = 'once'
diff --git a/test/orm/test_unitofwork.py b/test/orm/test_unitofwork.py
index ea6397517..c95836055 100644
--- a/test/orm/test_unitofwork.py
+++ b/test/orm/test_unitofwork.py
@@ -11,7 +11,8 @@ from sqlalchemy.test import engines, testing, pickleable
 from sqlalchemy import Integer, String, ForeignKey, literal_column
 from sqlalchemy.test.schema import Table
 from sqlalchemy.test.schema import Column
-from sqlalchemy.orm import mapper, relationship, create_session, column_property, attributes
+from sqlalchemy.orm import mapper, relationship, create_session, \
+    column_property, attributes, Session, reconstructor, object_session
 from sqlalchemy.test.testing import eq_, ne_
 from test.orm import _base, _fixtures
 from test.engine import _base as engine_base
@@ -698,10 +699,15 @@ class PassiveDeletesTest(_base.MappedTest):
         assert mytable.count().scalar() == 0
         assert myothertable.count().scalar() == 0
     
+    @testing.emits_warning(r".*'passive_deletes' is normally configured on one-to-many")
     @testing.resolve_artifact_names
     def test_backwards_pd(self):
-        # the unusual scenario where a trigger or something might be deleting
-        # a many-to-one on deletion of the parent row
+        """Test that passive_deletes=True disables a delete from an m2o.
+        
+        This is not the usual usage and it now raises a warning, but test
+        that it works nonetheless.
+
+        """
         mapper(MyOtherClass, myothertable, properties={
             'myclass':relationship(MyClass, cascade="all, delete", passive_deletes=True)
         })
@@ -721,8 +727,17 @@ class PassiveDeletesTest(_base.MappedTest):
         session.delete(mco)
         session.flush()
         
+        # mytable wasn't deleted, is the point.
         assert mytable.count().scalar() == 1
         assert myothertable.count().scalar() == 0
+    
+    @testing.resolve_artifact_names
+    def test_aaa_m2o_emits_warning(self):
+        mapper(MyOtherClass, myothertable, properties={
+            'myclass':relationship(MyClass, cascade="all, delete", passive_deletes=True)
+        })
+        mapper(MyClass, mytable)
+        assert_raises(sa.exc.SAWarning, sa.orm.compile_mappers)
         
 class ExtraPassiveDeletesTest(_base.MappedTest):
     __requires__ = ('foreign_keys',)
@@ -2111,7 +2126,68 @@ class BooleanColTest(_base.MappedTest):
         sess.flush()
         eq_(sess.query(T).filter(T.value==True).all(), [T(value=True, name="t1"),T(value=True, name="t3")])
 
-
+class DontAllowFlushOnLoadingObjectTest(_base.MappedTest):
+    """Test that objects with NULL identity keys aren't permitted to complete a flush.
+    
+    User-defined callables that execute during a load may modify state
+    on instances which results in their being autoflushed, before attributes
+    are populated.  If the primary key identifiers are missing, an explicit assertion
+    is needed to check that the object doesn't go through the flush process with
+    no net changes and gets placed in the identity map with an incorrect 
+    identity key.
+    
+    """
+    @classmethod
+    def define_tables(cls, metadata):
+        t1 = Table('t1', metadata,
+            Column('id', Integer, primary_key=True),
+            Column('data', String(30)),
+        )
+    
+    @testing.resolve_artifact_names
+    def test_flush_raises(self):
+        class T1(_base.ComparableEntity):
+            @reconstructor
+            def go(self):
+                # blow away 'id', no change event.
+                # this simulates a callable occuring
+                # before 'id' was even populated, i.e. a callable
+                # within an attribute_mapped_collection
+                self.__dict__.pop('id', None)
+                
+                # generate a change event, perhaps this occurs because
+                # someone wrote a broken attribute_mapped_collection that 
+                # inappropriately fires off change events when it should not,
+                # now we're dirty
+                self.data = 'foo bar'
+                
+                # blow away that change, so an UPDATE does not occur
+                # (since it would break)
+                self.__dict__.pop('data', None)
+                
+                # flush ! any lazyloader here would trigger
+                # autoflush, for example.
+                sess.flush()
+                
+        mapper(T1, t1)
+        
+        sess = Session()
+        sess.add(T1(data='test', id=5))
+        sess.commit()
+        sess.close()
+        
+        # make sure that invalid state doesn't get into the session
+        # with the wrong key.  If the identity key is not NULL, at least
+        # the population process would continue after the erroneous flush
+        # and thing would right themselves.
+        assert_raises_message(sa.orm.exc.FlushError,
+                              'has a NULL identity key.  Check if this '
+                              'flush is occuring at an inappropriate '
+                              'time, such as during a load operation.',
+                              sess.query(T1).first)
+        
+        
+    
 class RowSwitchTest(_base.MappedTest):
     @classmethod
     def define_tables(cls, metadata):
diff --git a/test/sql/test_case_statement.py b/test/sql/test_case_statement.py
index 645822fa7..1a106ee5e 100644
--- a/test/sql/test_case_statement.py
+++ b/test/sql/test_case_statement.py
@@ -32,14 +32,14 @@ class CaseTest(TestBase, AssertsCompiledSQL):
     @testing.fails_on('firebird', 'FIXME: unknown')
     @testing.fails_on('maxdb', 'FIXME: unknown')
     @testing.requires.subqueries
-    def testcase(self):
+    def test_case(self):
         inner = select([case([
                 [info_table.c.pk < 3,
                         'lessthan3'],
         [and_(info_table.c.pk >= 3, info_table.c.pk < 7),
                         'gt3']]).label('x'),
         info_table.c.pk, info_table.c.info],
-                from_obj=[info_table]).alias('q_inner')
+                from_obj=[info_table])
 
         inner_result = inner.execute().fetchall()
 
@@ -59,7 +59,7 @@ class CaseTest(TestBase, AssertsCompiledSQL):
             ('gt3', 6, 'pk_6_data')
         ]
 
-        outer = select([inner])
+        outer = select([inner.alias('q_inner')])
 
         outer_result = outer.execute().fetchall()
 
@@ -79,7 +79,7 @@ class CaseTest(TestBase, AssertsCompiledSQL):
                         6]],
                 else_ = 0).label('x'),
         info_table.c.pk, info_table.c.info],
-                from_obj=[info_table]).alias('q_inner')
+                from_obj=[info_table])
 
         else_result = w_else.execute().fetchall()
 
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index 09432e1d4..07ceb9767 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -89,7 +89,7 @@ class SelectTest(TestBase, AssertsCompiledSQL):
     def test_invalid_col_argument(self):
         assert_raises(exc.ArgumentError, select, table1)
         assert_raises(exc.ArgumentError, select, table1.c.myid)
-        
+    
     def test_from_subquery(self):
         """tests placing select statements in the column clause of another select, for the
         purposes of selecting from the exported columns of that select."""
@@ -263,13 +263,27 @@ sq.myothertable_othername AS sq_myothertable_othername FROM (" + sqstring + ") A
         s3 = select([s2], use_labels=True)
         s4 = s3.alias()
         s5 = select([s4], use_labels=True)
-        self.assert_compile(s5, "SELECT anon_1.anon_2_myid AS anon_1_anon_2_myid, anon_1.anon_2_name AS anon_1_anon_2_name, "\
-        "anon_1.anon_2_description AS anon_1_anon_2_description FROM (SELECT anon_2.myid AS anon_2_myid, anon_2.name AS anon_2_name, "\
-        "anon_2.description AS anon_2_description FROM (SELECT mytable.myid AS myid, mytable.name AS name, mytable.description "\
-        "AS description FROM mytable) AS anon_2) AS anon_1")
+        self.assert_compile(s5,
+                            'SELECT anon_1.anon_2_myid AS '
+                            'anon_1_anon_2_myid, anon_1.anon_2_name AS '
+                            'anon_1_anon_2_name, anon_1.anon_2_descript'
+                            'ion AS anon_1_anon_2_description FROM '
+                            '(SELECT anon_2.myid AS anon_2_myid, '
+                            'anon_2.name AS anon_2_name, '
+                            'anon_2.description AS anon_2_description '
+                            'FROM (SELECT mytable.myid AS myid, '
+                            'mytable.name AS name, mytable.description '
+                            'AS description FROM mytable) AS anon_2) '
+                            'AS anon_1')
         
     def test_dont_overcorrelate(self):
-        self.assert_compile(select([table1], from_obj=[table1, table1.select()]), """SELECT mytable.myid, mytable.name, mytable.description FROM mytable, (SELECT mytable.myid AS myid, mytable.name AS name, mytable.description AS description FROM mytable)""")
+        self.assert_compile(select([table1], from_obj=[table1,
+                            table1.select()]),
+                            "SELECT mytable.myid, mytable.name, "
+                            "mytable.description FROM mytable, (SELECT "
+                            "mytable.myid AS myid, mytable.name AS "
+                            "name, mytable.description AS description "
+                            "FROM mytable)")
     
     def test_full_correlate(self):
         # intentional
@@ -301,31 +315,56 @@ sq.myothertable_othername AS sq_myothertable_othername FROM (" + sqstring + ") A
                     "EXISTS (SELECT mytable.myid FROM mytable WHERE mytable.myid = :myid_1)"
                 )
         
-        self.assert_compile(exists([table1.c.myid], table1.c.myid==5).select(), "SELECT EXISTS (SELECT mytable.myid FROM mytable WHERE mytable.myid = :myid_1)", params={'mytable_myid':5})
-
-        self.assert_compile(select([table1, exists([1], from_obj=table2)]), "SELECT mytable.myid, mytable.name, mytable.description, EXISTS (SELECT 1 FROM myothertable) FROM mytable", params={})
-
-        self.assert_compile(select([table1, exists([1], from_obj=table2).label('foo')]), "SELECT mytable.myid, mytable.name, mytable.description, EXISTS (SELECT 1 FROM myothertable) AS foo FROM mytable", params={})
-
-        self.assert_compile(
-          table1.select(exists().where(table2.c.otherid == table1.c.myid).correlate(table1)),
-          "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE EXISTS (SELECT * FROM myothertable WHERE myothertable.otherid = mytable.myid)"
-        )
-
-        self.assert_compile(
-          table1.select(exists().where(table2.c.otherid == table1.c.myid).correlate(table1)),
-          "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE EXISTS (SELECT * FROM myothertable WHERE myothertable.otherid = mytable.myid)"
-        )
-
-        self.assert_compile(
-          table1.select(exists().where(table2.c.otherid == table1.c.myid).correlate(table1)).replace_selectable(table2, table2.alias()),
-          "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE EXISTS (SELECT * FROM myothertable AS myothertable_1 WHERE myothertable_1.otherid = mytable.myid)"
-        )
-
-        self.assert_compile(
-          table1.select(exists().where(table2.c.otherid == table1.c.myid).correlate(table1)).select_from(table1.join(table2, table1.c.myid==table2.c.otherid)).replace_selectable(table2, table2.alias()),
-          "SELECT mytable.myid, mytable.name, mytable.description FROM mytable JOIN myothertable AS myothertable_1 ON mytable.myid = myothertable_1.otherid WHERE EXISTS (SELECT * FROM myothertable AS myothertable_1 WHERE myothertable_1.otherid = mytable.myid)"
-        )
+        self.assert_compile(exists([table1.c.myid], table1.c.myid
+                            == 5).select(),
+                            'SELECT EXISTS (SELECT mytable.myid FROM '
+                            'mytable WHERE mytable.myid = :myid_1)',
+                            params={'mytable_myid': 5})
+        self.assert_compile(select([table1, exists([1],
+                            from_obj=table2)]),
+                            'SELECT mytable.myid, mytable.name, '
+                            'mytable.description, EXISTS (SELECT 1 '
+                            'FROM myothertable) FROM mytable',
+                            params={})
+        self.assert_compile(select([table1, exists([1],
+                            from_obj=table2).label('foo')]),
+                            'SELECT mytable.myid, mytable.name, '
+                            'mytable.description, EXISTS (SELECT 1 '
+                            'FROM myothertable) AS foo FROM mytable',
+                            params={})
+
+        self.assert_compile(table1.select(exists().where(table2.c.otherid
+                            == table1.c.myid).correlate(table1)),
+                            'SELECT mytable.myid, mytable.name, '
+                            'mytable.description FROM mytable WHERE '
+                            'EXISTS (SELECT * FROM myothertable WHERE '
+                            'myothertable.otherid = mytable.myid)')
+        self.assert_compile(table1.select(exists().where(table2.c.otherid
+                            == table1.c.myid).correlate(table1)),
+                            'SELECT mytable.myid, mytable.name, '
+                            'mytable.description FROM mytable WHERE '
+                            'EXISTS (SELECT * FROM myothertable WHERE '
+                            'myothertable.otherid = mytable.myid)')
+        self.assert_compile(table1.select(exists().where(table2.c.otherid
+                            == table1.c.myid).correlate(table1)).replace_selectable(table2,
+                            table2.alias()),
+                            'SELECT mytable.myid, mytable.name, '
+                            'mytable.description FROM mytable WHERE '
+                            'EXISTS (SELECT * FROM myothertable AS '
+                            'myothertable_1 WHERE myothertable_1.otheri'
+                            'd = mytable.myid)')
+        self.assert_compile(table1.select(exists().where(table2.c.otherid
+                            == table1.c.myid).correlate(table1)).select_from(table1.join(table2,
+                            table1.c.myid
+                            == table2.c.otherid)).replace_selectable(table2,
+                            table2.alias()),
+                            'SELECT mytable.myid, mytable.name, '
+                            'mytable.description FROM mytable JOIN '
+                            'myothertable AS myothertable_1 ON '
+                            'mytable.myid = myothertable_1.otherid '
+                            'WHERE EXISTS (SELECT * FROM myothertable '
+                            'AS myothertable_1 WHERE '
+                            'myothertable_1.otherid = mytable.myid)')
         
         self.assert_compile(
             select([
@@ -334,62 +373,93 @@ sq.myothertable_othername AS sq_myothertable_othername FROM (" + sqstring + ") A
                     exists().where(table2.c.otherid=='bar')
                 )
             ]),
-            "SELECT (EXISTS (SELECT * FROM myothertable WHERE myothertable.otherid = :otherid_1)) "\
-            "OR (EXISTS (SELECT * FROM myothertable WHERE myothertable.otherid = :otherid_2)) AS anon_1"
+            "SELECT (EXISTS (SELECT * FROM myothertable "
+            "WHERE myothertable.otherid = :otherid_1)) "
+            "OR (EXISTS (SELECT * FROM myothertable WHERE "
+            "myothertable.otherid = :otherid_2)) AS anon_1"
         )
         
 
     def test_where_subquery(self):
-        s = select([addresses.c.street], addresses.c.user_id==users.c.user_id, correlate=True).alias('s')
-        self.assert_compile(
-            select([users, s.c.street], from_obj=s),
-            """SELECT users.user_id, users.user_name, users.password, s.street FROM users, (SELECT addresses.street AS street FROM addresses WHERE addresses.user_id = users.user_id) AS s""")
-
-        self.assert_compile(
-            table1.select(table1.c.myid == select([table1.c.myid], table1.c.name=='jack')),
-            "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid = (SELECT mytable.myid FROM mytable WHERE mytable.name = :name_1)"
-        )
-
-        self.assert_compile(
-            table1.select(table1.c.myid == select([table2.c.otherid], table1.c.name == table2.c.othername)),
-            "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid = (SELECT myothertable.otherid FROM myothertable WHERE mytable.name = myothertable.othername)"
-        )
-
-        self.assert_compile(
-            table1.select(exists([1], table2.c.otherid == table1.c.myid)),
-            "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE EXISTS (SELECT 1 FROM myothertable WHERE myothertable.otherid = mytable.myid)"
-        )
-
-
+        s = select([addresses.c.street], addresses.c.user_id
+                   == users.c.user_id, correlate=True).alias('s')
+        self.assert_compile(select([users, s.c.street], from_obj=s),
+                            "SELECT users.user_id, users.user_name, "
+                            "users.password, s.street FROM users, "
+                            "(SELECT addresses.street AS street FROM "
+                            "addresses WHERE addresses.user_id = "
+                            "users.user_id) AS s")
+        self.assert_compile(table1.select(table1.c.myid
+                            == select([table1.c.myid], table1.c.name
+                            == 'jack')),
+                            'SELECT mytable.myid, mytable.name, '
+                            'mytable.description FROM mytable WHERE '
+                            'mytable.myid = (SELECT mytable.myid FROM '
+                            'mytable WHERE mytable.name = :name_1)')
+        self.assert_compile(table1.select(table1.c.myid
+                            == select([table2.c.otherid], table1.c.name
+                            == table2.c.othername)),
+                            'SELECT mytable.myid, mytable.name, '
+                            'mytable.description FROM mytable WHERE '
+                            'mytable.myid = (SELECT '
+                            'myothertable.otherid FROM myothertable '
+                            'WHERE mytable.name = myothertable.othernam'
+                            'e)')
+        self.assert_compile(table1.select(exists([1], table2.c.otherid
+                            == table1.c.myid)),
+                            'SELECT mytable.myid, mytable.name, '
+                            'mytable.description FROM mytable WHERE '
+                            'EXISTS (SELECT 1 FROM myothertable WHERE '
+                            'myothertable.otherid = mytable.myid)')
         talias = table1.alias('ta')
-        s = subquery('sq2', [talias], exists([1], table2.c.otherid == talias.c.myid))
-        self.assert_compile(
-            select([s, table1])
-            ,"SELECT sq2.myid, sq2.name, sq2.description, mytable.myid, mytable.name, mytable.description FROM (SELECT ta.myid AS myid, ta.name AS name, ta.description AS description FROM mytable AS ta WHERE EXISTS (SELECT 1 FROM myothertable WHERE myothertable.otherid = ta.myid)) AS sq2, mytable")
-
-        s = select([addresses.c.street], addresses.c.user_id==users.c.user_id, correlate=True).alias('s')
-        self.assert_compile(
-            select([users, s.c.street], from_obj=s),
-            """SELECT users.user_id, users.user_name, users.password, s.street FROM users, (SELECT addresses.street AS street FROM addresses WHERE addresses.user_id = users.user_id) AS s""")
-
-        # test constructing the outer query via append_column(), which occurs in the ORM's Query object
-        s = select([], exists([1], table2.c.otherid==table1.c.myid), from_obj=table1)
+        s = subquery('sq2', [talias], exists([1], table2.c.otherid
+                     == talias.c.myid))
+        self.assert_compile(select([s, table1]),
+                            'SELECT sq2.myid, sq2.name, '
+                            'sq2.description, mytable.myid, '
+                            'mytable.name, mytable.description FROM '
+                            '(SELECT ta.myid AS myid, ta.name AS name, '
+                            'ta.description AS description FROM '
+                            'mytable AS ta WHERE EXISTS (SELECT 1 FROM '
+                            'myothertable WHERE myothertable.otherid = '
+                            'ta.myid)) AS sq2, mytable')
+        s = select([addresses.c.street], addresses.c.user_id
+                   == users.c.user_id, correlate=True).alias('s')
+        self.assert_compile(select([users, s.c.street], from_obj=s),
+                            "SELECT users.user_id, users.user_name, "
+                            "users.password, s.street FROM users, "
+                            "(SELECT addresses.street AS street FROM "
+                            "addresses WHERE addresses.user_id = "
+                            "users.user_id) AS s")
+
+        # test constructing the outer query via append_column(), which
+        # occurs in the ORM's Query object
+
+        s = select([], exists([1], table2.c.otherid == table1.c.myid),
+                   from_obj=table1)
         s.append_column(table1)
-        self.assert_compile(
-            s,
-            "SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE EXISTS (SELECT 1 FROM myothertable WHERE myothertable.otherid = mytable.myid)"
-        )
+        self.assert_compile(s,
+                            'SELECT mytable.myid, mytable.name, '
+                            'mytable.description FROM mytable WHERE '
+                            'EXISTS (SELECT 1 FROM myothertable WHERE '
+                            'myothertable.otherid = mytable.myid)')
 
 
     def test_orderby_subquery(self):
-        self.assert_compile(
-            table1.select(order_by=[select([table2.c.otherid], table1.c.myid==table2.c.otherid)]),
-            "SELECT mytable.myid, mytable.name, mytable.description FROM mytable ORDER BY (SELECT myothertable.otherid FROM myothertable WHERE mytable.myid = myothertable.otherid)"
-        )
-        self.assert_compile(
-            table1.select(order_by=[desc(select([table2.c.otherid], table1.c.myid==table2.c.otherid))]),
-            "SELECT mytable.myid, mytable.name, mytable.description FROM mytable ORDER BY (SELECT myothertable.otherid FROM myothertable WHERE mytable.myid = myothertable.otherid) DESC"
-        )
+        self.assert_compile(table1.select(order_by=[select([table2.c.otherid],
+                            table1.c.myid == table2.c.otherid)]),
+                            'SELECT mytable.myid, mytable.name, '
+                            'mytable.description FROM mytable ORDER BY '
+                            '(SELECT myothertable.otherid FROM '
+                            'myothertable WHERE mytable.myid = '
+                            'myothertable.otherid)')
+        self.assert_compile(table1.select(order_by=[desc(select([table2.c.otherid],
+                            table1.c.myid == table2.c.otherid))]),
+                            'SELECT mytable.myid, mytable.name, '
+                            'mytable.description FROM mytable ORDER BY '
+                            '(SELECT myothertable.otherid FROM '
+                            'myothertable WHERE mytable.myid = '
+                            'myothertable.otherid) DESC')
 
     @testing.uses_deprecated('scalar option')
     def test_scalar_select(self):
@@ -401,41 +471,76 @@ sq.myothertable_othername AS sq_myothertable_othername FROM (" + sqstring + ") A
         )
 
         s = select([table1.c.myid], correlate=False).as_scalar()
-        self.assert_compile(select([table1, s]), "SELECT mytable.myid, mytable.name, mytable.description, (SELECT mytable.myid FROM mytable) AS anon_1 FROM mytable")
-
+        self.assert_compile(select([table1, s]),
+                            'SELECT mytable.myid, mytable.name, '
+                            'mytable.description, (SELECT mytable.myid '
+                            'FROM mytable) AS anon_1 FROM mytable')
         s = select([table1.c.myid]).as_scalar()
-        self.assert_compile(select([table2, s]), "SELECT myothertable.otherid, myothertable.othername, (SELECT mytable.myid FROM mytable) AS anon_1 FROM myothertable")
-
+        self.assert_compile(select([table2, s]),
+                            'SELECT myothertable.otherid, '
+                            'myothertable.othername, (SELECT '
+                            'mytable.myid FROM mytable) AS anon_1 FROM '
+                            'myothertable')
         s = select([table1.c.myid]).correlate(None).as_scalar()
-        self.assert_compile(select([table1, s]), "SELECT mytable.myid, mytable.name, mytable.description, (SELECT mytable.myid FROM mytable) AS anon_1 FROM mytable")
-
-        # test that aliases use as_scalar() when used in an explicitly scalar context
-        s = select([table1.c.myid]).alias()
-        self.assert_compile(select([table1.c.myid]).where(table1.c.myid==s), "SELECT mytable.myid FROM mytable WHERE mytable.myid = (SELECT mytable.myid FROM mytable)")
-        self.assert_compile(select([table1.c.myid]).where(s > table1.c.myid), "SELECT mytable.myid FROM mytable WHERE mytable.myid < (SELECT mytable.myid FROM mytable)")
+        self.assert_compile(select([table1, s]),
+                            'SELECT mytable.myid, mytable.name, '
+                            'mytable.description, (SELECT mytable.myid '
+                            'FROM mytable) AS anon_1 FROM mytable')
 
+        # test that aliases use as_scalar() when used in an explicitly
+        # scalar context
 
+        s = select([table1.c.myid]).alias()
+        self.assert_compile(select([table1.c.myid]).where(table1.c.myid
+                            == s),
+                            'SELECT mytable.myid FROM mytable WHERE '
+                            'mytable.myid = (SELECT mytable.myid FROM '
+                            'mytable)')
+        self.assert_compile(select([table1.c.myid]).where(s
+                            > table1.c.myid),
+                            'SELECT mytable.myid FROM mytable WHERE '
+                            'mytable.myid < (SELECT mytable.myid FROM '
+                            'mytable)')
         s = select([table1.c.myid]).as_scalar()
-        self.assert_compile(select([table2, s]), "SELECT myothertable.otherid, myothertable.othername, (SELECT mytable.myid FROM mytable) AS anon_1 FROM myothertable")
+        self.assert_compile(select([table2, s]),
+                            'SELECT myothertable.otherid, '
+                            'myothertable.othername, (SELECT '
+                            'mytable.myid FROM mytable) AS anon_1 FROM '
+                            'myothertable')
 
         # test expressions against scalar selects
-        self.assert_compile(select([s - literal(8)]), "SELECT (SELECT mytable.myid FROM mytable) - :param_1 AS anon_1")
-        self.assert_compile(select([select([table1.c.name]).as_scalar() + literal('x')]), "SELECT (SELECT mytable.name FROM mytable) || :param_1 AS anon_1")
-        self.assert_compile(select([s > literal(8)]), "SELECT (SELECT mytable.myid FROM mytable) > :param_1 AS anon_1")
 
-        self.assert_compile(select([select([table1.c.name]).label('foo')]), "SELECT (SELECT mytable.name FROM mytable) AS foo")
+        self.assert_compile(select([s - literal(8)]),
+                            'SELECT (SELECT mytable.myid FROM mytable) '
+                            '- :param_1 AS anon_1')
+        self.assert_compile(select([select([table1.c.name]).as_scalar()
+                            + literal('x')]),
+                            'SELECT (SELECT mytable.name FROM mytable) '
+                            '|| :param_1 AS anon_1')
+        self.assert_compile(select([s > literal(8)]),
+                            'SELECT (SELECT mytable.myid FROM mytable) '
+                            '> :param_1 AS anon_1')
+        self.assert_compile(select([select([table1.c.name]).label('foo'
+                            )]),
+                            'SELECT (SELECT mytable.name FROM mytable) '
+                            'AS foo')
+
+        # scalar selects should not have any attributes on their 'c' or
+        # 'columns' attribute
 
-        # scalar selects should not have any attributes on their 'c' or 'columns' attribute
         s = select([table1.c.myid]).as_scalar()
         try:
             s.c.foo
         except exc.InvalidRequestError, err:
-            assert str(err) == 'Scalar Select expression has no columns; use this object directly within a column-level expression.'
-
+            assert str(err) \
+                == 'Scalar Select expression has no columns; use this '\
+                'object directly within a column-level expression.'
         try:
             s.columns.foo
         except exc.InvalidRequestError, err:
-            assert str(err) == 'Scalar Select expression has no columns; use this object directly within a column-level expression.'
+            assert str(err) \
+                == 'Scalar Select expression has no columns; use this '\
+                'object directly within a column-level expression.'
 
         zips = table('zips',
             column('zipcode'),
@@ -455,29 +560,55 @@ sq.myothertable_othername AS sq_myothertable_othername FROM (" + sqstring + ") A
                          order_by = ['dist', places.c.nm]
                          )
 
-        self.assert_compile(q,"SELECT places.id, places.nm, zips.zipcode, latlondist((SELECT zips.latitude FROM zips WHERE "
-        "zips.zipcode = :zipcode_1), (SELECT zips.longitude FROM zips WHERE zips.zipcode = :zipcode_2)) AS dist "
-        "FROM places, zips WHERE zips.zipcode = :zipcode_3 ORDER BY dist, places.nm")
+        self.assert_compile(q,
+                            'SELECT places.id, places.nm, '
+                            'zips.zipcode, latlondist((SELECT '
+                            'zips.latitude FROM zips WHERE '
+                            'zips.zipcode = :zipcode_1), (SELECT '
+                            'zips.longitude FROM zips WHERE '
+                            'zips.zipcode = :zipcode_2)) AS dist FROM '
+                            'places, zips WHERE zips.zipcode = '
+                            ':zipcode_3 ORDER BY dist, places.nm')
 
         zalias = zips.alias('main_zip')
         qlat = select([zips.c.latitude], zips.c.zipcode == zalias.c.zipcode).as_scalar()
         qlng = select([zips.c.longitude], zips.c.zipcode == zalias.c.zipcode).as_scalar()
-        q = select([places.c.id, places.c.nm, zalias.c.zipcode, func.latlondist(qlat, qlng).label('dist')],
-                         order_by = ['dist', places.c.nm]
-                         )
-        self.assert_compile(q, "SELECT places.id, places.nm, main_zip.zipcode, latlondist((SELECT zips.latitude FROM zips WHERE zips.zipcode = main_zip.zipcode), (SELECT zips.longitude FROM zips WHERE zips.zipcode = main_zip.zipcode)) AS dist FROM places, zips AS main_zip ORDER BY dist, places.nm")
+        q = select([places.c.id, places.c.nm, zalias.c.zipcode,
+                   func.latlondist(qlat, qlng).label('dist')],
+                   order_by=['dist', places.c.nm])
+        self.assert_compile(q,
+                            'SELECT places.id, places.nm, '
+                            'main_zip.zipcode, latlondist((SELECT '
+                            'zips.latitude FROM zips WHERE '
+                            'zips.zipcode = main_zip.zipcode), (SELECT '
+                            'zips.longitude FROM zips WHERE '
+                            'zips.zipcode = main_zip.zipcode)) AS dist '
+                            'FROM places, zips AS main_zip ORDER BY '
+                            'dist, places.nm')
 
         a1 = table2.alias('t2alias')
         s1 = select([a1.c.otherid], table1.c.myid==a1.c.otherid).as_scalar()
         j1 = table1.join(table2, table1.c.myid==table2.c.otherid)
         s2 = select([table1, s1], from_obj=j1)
-        self.assert_compile(s2, "SELECT mytable.myid, mytable.name, mytable.description, (SELECT t2alias.otherid FROM myothertable AS t2alias WHERE mytable.myid = t2alias.otherid) AS anon_1 FROM mytable JOIN myothertable ON mytable.myid = myothertable.otherid")
+        self.assert_compile(s2,
+                            'SELECT mytable.myid, mytable.name, '
+                            'mytable.description, (SELECT '
+                            't2alias.otherid FROM myothertable AS '
+                            't2alias WHERE mytable.myid = '
+                            't2alias.otherid) AS anon_1 FROM mytable '
+                            'JOIN myothertable ON mytable.myid = '
+                            'myothertable.otherid')
 
     def test_label_comparison(self):
         x = func.lala(table1.c.myid).label('foo')
-        self.assert_compile(select([x], x==5), "SELECT lala(mytable.myid) AS foo FROM mytable WHERE lala(mytable.myid) = :param_1")
+        self.assert_compile(select([x], x == 5),
+                            'SELECT lala(mytable.myid) AS foo FROM '
+                            'mytable WHERE lala(mytable.myid) = '
+                            ':param_1')
 
-        self.assert_compile(label('bar', column('foo', type_=String)) + "foo", "foo || :param_1")
+        self.assert_compile(
+                label('bar', column('foo', type_=String))+ 'foo', 
+                'foo || :param_1')
     
 
     def test_conjunctions(self):
@@ -491,7 +622,8 @@ sq.myothertable_othername AS sq_myothertable_othername FROM (" + sqstring + ") A
         )
     
         self.assert_compile(
-            and_(table1.c.myid == 12, table1.c.name=='asdf', table2.c.othername == 'foo', "sysdate() = today()"),
+            and_(table1.c.myid == 12, table1.c.name=='asdf', 
+                table2.c.othername == 'foo', "sysdate() = today()"),
             "mytable.myid = :myid_1 AND mytable.name = :name_1 "\
             "AND myothertable.othername = :othername_1 AND sysdate() = today()"
         )
@@ -499,11 +631,14 @@ sq.myothertable_othername AS sq_myothertable_othername FROM (" + sqstring + ") A
         self.assert_compile(
             and_(
                 table1.c.myid == 12,
-                or_(table2.c.othername=='asdf', table2.c.othername == 'foo', table2.c.otherid == 9),
+                or_(table2.c.othername=='asdf', 
+                    table2.c.othername == 'foo', table2.c.otherid == 9),
                 "sysdate() = today()",
             ),
-            "mytable.myid = :myid_1 AND (myothertable.othername = :othername_1 OR "\
-            "myothertable.othername = :othername_2 OR myothertable.otherid = :otherid_1) AND sysdate() = today()",
+            'mytable.myid = :myid_1 AND (myothertable.othername = '
+             ':othername_1 OR myothertable.othername = :othername_2 OR '
+             'myothertable.otherid = :otherid_1) AND sysdate() = '
+             'today()', 
             checkparams = {'othername_1': 'asdf', 'othername_2':'foo', 'otherid_1': 9, 'myid_1': 12}
         )
         
@@ -1766,18 +1901,74 @@ sq.myothertable_othername AS sq_myothertable_othername FROM (" + sqstring + ") A
         self.assert_compile(table.select(between((table.c.field == table.c.field), False, True)),
             "SELECT op.field FROM op WHERE (op.field = op.field) BETWEEN :param_1 AND :param_2")
     
+    def test_delayed_col_naming(self):
+        my_str = Column(String)
+        
+        sel1 = select([my_str])
+        
+        assert_raises_message(
+            exc.InvalidRequestError,
+            "Cannot initialize a sub-selectable with this Column",
+            lambda: sel1.c
+        )
+        
+        # calling label or as_scalar doesn't compile
+        # anything.  
+        sel2 = select([func.substr(my_str, 2, 3)]).label('my_substr')
+        
+        assert_raises_message(
+            exc.CompileError,
+            "Cannot compile Column object until it's 'name' is assigned.",
+            str, sel2
+        )
+        
+        sel3 = select([my_str]).as_scalar()
+        assert_raises_message(
+            exc.CompileError,
+            "Cannot compile Column object until it's 'name' is assigned.",
+            str, sel3
+        )
+        
+        my_str.name = 'foo'
+        
+        self.assert_compile(
+            sel1,
+            "SELECT foo",
+        )
+        self.assert_compile(
+            sel2,
+            '(SELECT substr(foo, :substr_2, :substr_3) AS substr_1)',
+        )
+        
+        self.assert_compile(
+            sel3,
+            "(SELECT foo)"
+        )
+        
     def test_naming(self):
-        s1 = select([table1.c.myid, table1.c.myid.label('foobar'), func.hoho(table1.c.name), func.lala(table1.c.name).label('gg')])
-        assert s1.c.keys() == ['myid', 'foobar', 'hoho(mytable.name)', 'gg']
+        f1 = func.hoho(table1.c.name)
+        s1 = select([table1.c.myid, table1.c.myid.label('foobar'),
+                    f1,
+                    func.lala(table1.c.name).label('gg')])
+        
+        eq_(
+            s1.c.keys(),
+            ['myid', 'foobar', str(f1), 'gg']
+        )
 
         meta = MetaData()
         t1 = Table('mytable', meta, Column('col1', Integer))
         
+        exprs = (
+            table1.c.myid==12,
+            func.hoho(table1.c.myid),
+            cast(table1.c.name, Numeric)
+        )
         for col, key, expr, label in (
             (table1.c.name, 'name', 'mytable.name', None),
-            (table1.c.myid==12, 'mytable.myid = :myid_1', 'mytable.myid = :myid_1', 'anon_1'),
-            (func.hoho(table1.c.myid), 'hoho(mytable.myid)', 'hoho(mytable.myid)', 'hoho_1'),
-            (cast(table1.c.name, Numeric), 'CAST(mytable.name AS NUMERIC)', 'CAST(mytable.name AS NUMERIC)', 'anon_1'),
+            (exprs[0], str(exprs[0]), 'mytable.myid = :myid_1', 'anon_1'),
+            (exprs[1], str(exprs[1]), 'hoho(mytable.myid)', 'hoho_1'),
+            (exprs[2], str(exprs[2]), 'CAST(mytable.name AS NUMERIC)', 'anon_1'),
             (t1.c.col1, 'col1', 'mytable.col1', None),
             (column('some wacky thing'), 'some wacky thing', '"some wacky thing"', '')
         ):
diff --git a/test/sql/test_query.py b/test/sql/test_query.py
index 0a496906d..a87931bb3 100644
--- a/test/sql/test_query.py
+++ b/test/sql/test_query.py
@@ -216,7 +216,7 @@ class QueryTest(TestBase):
                 {'user_name':'jack'},
             )
             assert r.closed
-        
+    
     def test_row_iteration(self):
         users.insert().execute(
             {'user_id':7, 'user_name':'jack'},
@@ -619,6 +619,16 @@ class QueryTest(TestBase):
         eq_(r[users.c.user_name], 'jack')
         eq_(r.user_name, 'jack')
 
+    @testing.requires.dbapi_lastrowid
+    def test_native_lastrowid(self):
+        r = testing.db.execute(
+            users.insert(),
+            {'user_id':1, 'user_name':'ed'}
+        )
+        
+        eq_(r.lastrowid, 1)
+        
+        
     def test_graceful_fetch_on_non_rows(self):
         """test that calling fetchone() etc. on a result that doesn't
         return rows fails gracefully.
@@ -743,6 +753,7 @@ class QueryTest(TestBase):
         r = testing.db.execute('select user_name from query_users').first()
         eq_(len(r), 1)
 
+    @testing.uses_deprecated(r'.*which subclass Executable')
     def test_cant_execute_join(self):
         try:
             users.join(addresses).execute()
@@ -784,7 +795,10 @@ class QueryTest(TestBase):
         )
         shadowed.create(checkfirst=True)
         try:
-            shadowed.insert().execute(shadow_id=1, shadow_name='The Shadow', parent='The Light', row='Without light there is no shadow', _parent='Hidden parent', _row='Hidden row')
+            shadowed.insert().execute(shadow_id=1, shadow_name='The Shadow', parent='The Light', 
+                                            row='Without light there is no shadow', 
+                                            _parent='Hidden parent', 
+                                            _row='Hidden row')
             r = shadowed.select(shadowed.c.shadow_id==1).execute().first()
             self.assert_(r.shadow_id == r['shadow_id'] == r[shadowed.c.shadow_id] == 1)
             self.assert_(r.shadow_name == r['shadow_name'] == r[shadowed.c.shadow_name] == 'The Shadow')
diff --git a/test/sql/test_types.py b/test/sql/test_types.py
index a80e761d7..af460628e 100644
--- a/test/sql/test_types.py
+++ b/test/sql/test_types.py
@@ -17,11 +17,13 @@ from sqlalchemy.test.util import round_decimal
 
 class AdaptTest(TestBase):
     def test_uppercase_rendering(self):
-        """Test that uppercase types from types.py always render as their type.
+        """Test that uppercase types from types.py always render as their
+        type.
         
-        As of SQLA 0.6, using an uppercase type means you want specifically that
-        type.  If the database in use doesn't support that DDL, it (the DB backend) 
-        should raise an error - it means you should be using a lowercased (genericized) type.
+        As of SQLA 0.6, using an uppercase type means you want specifically
+        that type. If the database in use doesn't support that DDL, it (the DB
+        backend) should raise an error - it means you should be using a
+        lowercased (genericized) type.
         
         """
         
@@ -30,20 +32,21 @@ class AdaptTest(TestBase):
                 mysql.dialect(), 
                 postgresql.dialect(), 
                 sqlite.dialect(), 
-                mssql.dialect()]: # TODO when dialects are complete:  engines.all_dialects():
+                mssql.dialect()]: 
             for type_, expected in (
                 (FLOAT, "FLOAT"),
                 (NUMERIC, "NUMERIC"),
                 (DECIMAL, "DECIMAL"),
                 (INTEGER, "INTEGER"),
                 (SMALLINT, "SMALLINT"),
-                (TIMESTAMP, "TIMESTAMP"),
+                (TIMESTAMP, ("TIMESTAMP", "TIMESTAMP WITHOUT TIME ZONE")),
                 (DATETIME, "DATETIME"),
                 (DATE, "DATE"),
-                (TIME, "TIME"),
+                (TIME, ("TIME", "TIME WITHOUT TIME ZONE")),
                 (CLOB, "CLOB"),
                 (VARCHAR(10), ("VARCHAR(10)","VARCHAR(10 CHAR)")),
-                (NVARCHAR(10), ("NVARCHAR(10)", "NATIONAL VARCHAR(10)", "NVARCHAR2(10)")),
+                (NVARCHAR(10), ("NVARCHAR(10)", "NATIONAL VARCHAR(10)",
+                                    "NVARCHAR2(10)")),
                 (CHAR, "CHAR"),
                 (NCHAR, ("NCHAR", "NATIONAL CHAR")),
                 (BLOB, "BLOB"),
@@ -51,14 +54,18 @@ class AdaptTest(TestBase):
             ):
                 if isinstance(expected, str):
                     expected = (expected, )
-                for exp in expected:
-                    compiled = types.to_instance(type_).compile(dialect=dialect)
-                    if exp in compiled:
-                        break
-                else:
-                    assert False, "%r matches none of %r for dialect %s" % \
-                                            (compiled, expected, dialect.name)
-            
+
+                compiled = types.to_instance(type_).\
+                            compile(dialect=dialect)
+                    
+                assert compiled in expected, \
+                    "%r matches none of %r for dialect %s" % \
+                    (compiled, expected, dialect.name)
+                
+                assert str(types.to_instance(type_)) in expected, \
+                    "default str() of type %r not expected, %r" % \
+                    (type_, expected)
+                
 class TypeAffinityTest(TestBase):
     def test_type_affinity(self):
         for type_, affin in [
diff --git a/test/zblog/test_zblog.py b/test/zblog/test_zblog.py
index 5e46c1ceb..8103cde8b 100644
--- a/test/zblog/test_zblog.py
+++ b/test/zblog/test_zblog.py
@@ -52,11 +52,10 @@ class SavePostTest(ZBlogTest):
         clear_mappers()
         super(SavePostTest, cls).teardown_class()
 
-    def testattach(self):
-        """test that a transient/pending instance has proper bi-directional behavior.
-
-        this requires that lazy loaders do not fire off for a transient/pending instance."""
-        s = create_session(bind=testing.db)
+    def test_attach_noautoflush(self):
+        """Test pending backref behavior."""
+        
+        s = create_session(bind=testing.db, autoflush=False)
 
         s.begin()
         try:
@@ -69,6 +68,21 @@ class SavePostTest(ZBlogTest):
         finally:
             s.rollback()
 
+    def test_attach_autoflush(self):
+        s = create_session(bind=testing.db, autoflush=True)
+
+        s.begin()
+        try:
+            blog = s.query(Blog).get(blog_id)
+            user = s.query(User).get(user_id)
+            post = Post(headline="asdf asdf", summary="asdfasfd", user=user)
+            s.add(post)
+            post.blog_id=blog_id
+            post.blog = blog
+            assert post in blog.posts
+        finally:
+            s.rollback()
+
     def testoptimisticorphans(self):
         """test that instances in the session with un-loaded parents will not
         get marked as "orphans" and then deleted """