finally, a really straightforward reduce() method which reduces cols

to the minimal set for every test case I can come up with, and
now replaces all the cruft in Mapper._compile_pks() as well as
Join.__init_primary_key().  mappers can now handle aliased selects
and figure out the correct PKs pretty well [ticket:933]

3 files changed, 130 insertions, 2 deletions

diff --git a/test/orm/inheritance/query.py b/test/orm/inheritance/query.py
index b3239d3b3..b9f11faa7 100644
--- a/test/orm/inheritance/query.py
+++ b/test/orm/inheritance/query.py
@@ -159,7 +159,16 @@ def make_test(select_type):
             all_employees = [e1, e2, b1, m1, e3]
             c1_employees = [e1, e2, b1, m1]
             c2_employees = [e3]
-
+            
+        def test_get(self):
+            sess = create_session()
+            
+            # for all mappers, ensure the primary key has been calculated as just the "person_id"
+            # column
+            self.assertEquals(sess.query(Person).get(e1.person_id), Engineer(name="dilbert"))
+            self.assertEquals(sess.query(Engineer).get(e1.person_id), Engineer(name="dilbert"))
+            self.assertEquals(sess.query(Manager).get(b1.person_id), Boss(name="pointy haired boss"))
+            
         def test_filter_on_subclass(self):
             sess = create_session()
             self.assertEquals(sess.query(Engineer).all()[0], Engineer(name="dilbert"))
diff --git a/test/orm/mapper.py b/test/orm/mapper.py
index 70cd81428..a8f75a31b 100644
--- a/test/orm/mapper.py
+++ b/test/orm/mapper.py
@@ -72,7 +72,7 @@ class MapperTest(MapperSuperTest):
             mapper(User, s)
             assert False
         except exceptions.ArgumentError, e:
-            assert str(e) == "Could not assemble any primary key columns for mapped table 'foo'"
+            assert "could not assemble any primary key columns for mapped table 'foo'" in str(e)
 
     def test_compileonsession(self):
         m = mapper(User, users)
diff --git a/test/sql/selectable.py b/test/sql/selectable.py
index a64697b81..45bd7d823 100755
--- a/test/sql/selectable.py
+++ b/test/sql/selectable.py
@@ -5,6 +5,7 @@ every selectable unit behaving nicely with others.."""
 import testenv; testenv.configure_for_tests()
 from sqlalchemy import *
 from testlib import *
+from sqlalchemy.sql import util as sql_util
 
 metadata = MetaData()
 table = Table('table1', metadata,
@@ -275,6 +276,124 @@ class PrimaryKeyTest(AssertMixin):
         assert str(j) == "a JOIN b ON a.id = b.id AND b.x = :b_x_1", str(j)
         assert list(j.primary_key) == [a.c.id, b.c.x]
 
+    def test_onclause_direction(self):
+        metadata = MetaData()
+
+        employee = Table( 'Employee', metadata,
+            Column('name', String(100)),
+            Column('id', Integer, primary_key= True),
+        )
+
+        engineer = Table( 'Engineer', metadata,
+            Column('id', Integer, ForeignKey( 'Employee.id', ), primary_key=True),
+        )
+
+        self.assertEquals(
+            set(employee.join(engineer, employee.c.id==engineer.c.id).primary_key),
+            set([employee.c.id])
+        )
+
+        self.assertEquals(
+            set(employee.join(engineer, engineer.c.id==employee.c.id).primary_key),
+            set([employee.c.id])
+        )
+
+
+class ReduceTest(AssertMixin):
+    def test_reduce(self):
+        meta = MetaData()
+        t1 = Table('t1', meta,
+            Column('t1id', Integer, primary_key=True),
+            Column('t1data', String(30)))
+        t2 = Table('t2', meta,
+            Column('t2id', Integer, ForeignKey('t1.t1id'), primary_key=True),
+            Column('t2data', String(30)))
+        t3 = Table('t3', meta,
+            Column('t3id', Integer, ForeignKey('t2.t2id'), primary_key=True),
+            Column('t3data', String(30)))
+        
+        
+        self.assertEquals(
+            set(sql_util.reduce_columns([t1.c.t1id, t1.c.t1data, t2.c.t2id, t2.c.t2data, t3.c.t3id, t3.c.t3data])),
+            set([t1.c.t1id, t1.c.t1data, t2.c.t2data, t3.c.t3data])
+        )
+    
+    def test_reduce_aliased_join(self):
+        metadata = MetaData()
+        people = Table('people', metadata,
+           Column('person_id', Integer, Sequence('person_id_seq', optional=True), primary_key=True),
+           Column('name', String(50)),
+           Column('type', String(30)))
+
+        engineers = Table('engineers', metadata,
+           Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
+           Column('status', String(30)),
+           Column('engineer_name', String(50)),
+           Column('primary_language', String(50)),
+          )
+     
+        managers = Table('managers', metadata,
+           Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True),
+           Column('status', String(30)),
+           Column('manager_name', String(50))
+           )
+        
+        pjoin = people.outerjoin(engineers).outerjoin(managers).select(use_labels=True).alias('pjoin')
+        self.assertEquals(
+            set(sql_util.reduce_columns([pjoin.c.people_person_id, pjoin.c.engineers_person_id, pjoin.c.managers_person_id])),
+            set([pjoin.c.people_person_id])
+        )
+        
+    def test_reduce_aliased_union(self):
+        metadata = MetaData()
+        item_table = Table(
+            'item', metadata,
+            Column('id', Integer, ForeignKey('base_item.id'), primary_key=True),
+            Column('dummy', Integer, default=0))
+
+        base_item_table = Table(
+            'base_item', metadata,
+            Column('id', Integer, primary_key=True),
+            Column('child_name', String(255), default=None))
+        
+        from sqlalchemy.orm.util import polymorphic_union
+        
+        item_join = polymorphic_union( {
+            'BaseItem':base_item_table.select(base_item_table.c.child_name=='BaseItem'),
+            'Item':base_item_table.join(item_table),
+            }, None, 'item_join')
+            
+        self.assertEquals(
+            set(sql_util.reduce_columns([item_join.c.id, item_join.c.dummy, item_join.c.child_name])),
+            set([item_join.c.id, item_join.c.dummy, item_join.c.child_name])
+        )    
+    
+    def test_reduce_aliased_union_2(self):
+        metadata = MetaData()
+
+        page_table = Table('page', metadata,
+            Column('id', Integer, primary_key=True),
+        )
+        magazine_page_table = Table('magazine_page', metadata,
+            Column('page_id', Integer, ForeignKey('page.id'), primary_key=True),
+        )
+        classified_page_table = Table('classified_page', metadata,
+            Column('magazine_page_id', Integer, ForeignKey('magazine_page.page_id'), primary_key=True),
+        )
+        
+        from sqlalchemy.orm.util import polymorphic_union
+        pjoin = polymorphic_union(
+            {
+                'm': page_table.join(magazine_page_table),
+                'c': page_table.join(magazine_page_table).join(classified_page_table),
+            }, None, 'page_join')
+            
+        self.assertEquals(
+            set(sql_util.reduce_columns([pjoin.c.id, pjoin.c.page_id, pjoin.c.magazine_page_id])),
+            set([pjoin.c.id])
+        )    
+    
+            
 class DerivedTest(AssertMixin):
     def test_table(self):
         meta = MetaData()