a = Table('a', meta, Column('id', Integer, primary_key=True))        b = Table('b', meta, Column('id', Integer, ForeignKey('a.id'), primary_key=True))        c = Table('c', meta, Column('id', Integer, ForeignKey('b.id'), primary_key=True))        d = Table('d', meta, Column('id', Integer, ForeignKey('c.id'), primary_key=True))        assert c.c.id.references(b.c.id)        assert not d.c.id.references(a.c.id)        assert list(a.join(b).primary_key) == [a.c.id]        assert list(b.join(c).primary_key) == [b.c.id]        assert list(a.join(b).join(c).primary_key) == [a.c.id]        assert list(b.join(c).join(d).primary_key) == [b.c.id]        assert list(d.join(c).join(b).primary_key) == [b.c.id]        assert list(a.join(b).join(c).join(d).primary_key) == [a.c.id]    def test_join_pk_collapse_explicit(self):        """test that redundant columns in a join get 'collapsed' into a minimal primary key,        which is the root column along a chain of explicit join conditions."""        meta = MetaData()        a = Table('a', meta, Column('id', Integer, primary_key=True), Column('x', Integer))        b = Table('b', meta, Column('id', Integer, ForeignKey('a.id'), primary_key=True), Column('x', Integer))        c = Table('c', meta, Column('id', Integer, ForeignKey('b.id'), primary_key=True), Column('x', Integer))        d = Table('d', meta, Column('id', Integer, ForeignKey('c.id'), primary_key=True), Column('x', Integer))        print list(a.join(b, a.c.x==b.c.id).primary_key)        assert list(a.join(b, a.c.x==b.c.id).primary_key) == [b.c.id]        assert list(b.join(c, b.c.x==c.c.id).primary_key) == [b.c.id]        assert list(a.join(b).join(c, c.c.id==b.c.x).primary_key) == [a.c.id]        assert list(b.join(c, c.c.x==b.c.id).join(d).primary_key) == [c.c.id]        assert list(b.join(c, c.c.id==b.c.x).join(d).primary_key) == [b.c.id]        assert list(d.join(b, d.c.id==b.c.id).join(c, b.c.id==c.c.x).primary_key) == [c.c.id]        assert list(a.join(b).join(c, c.c.id==b.c.x).join(d).primary_key) == [a.c.id]        assert list(a.join(b, and_(a.c.id==b.c.id, a.c.x==b.c.id)).primary_key) == [a.c.id]    def test_init_doesnt_blowitaway(self):        meta = MetaData()        a = Table('a', meta, Column('id', Integer, primary_key=True), Column('x', Integer))        b = Table('b', meta, Column('id', Integer, ForeignKey('a.id'), primary_key=True), Column('x', Integer))        j = a.join(b)        assert list(j.primary_key) == [a.c.id]        j.foreign_keys        assert list(j.primary_key) == [a.c.id]    def test_non_column_clause(self):        meta = MetaData()        a = Table('a', meta, Column('id', Integer, primary_key=True), Column('x', Integer))        b = Table('b', meta, Column('id', Integer, ForeignKey('a.id'), primary_key=True), Column('x', Integer, primary_key=True))        j = a.join(b, and_(a.c.id==b.c.id, b.c.x==5))        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(TestBase, AssertsExecutionResults):    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_selectable(self):       metadata = MetaData()       engineers = Table('engineers', metadata,           Column('engineer_id', Integer, primary_key=True),           Column('engineer_name', String(50)),          )            managers = Table('managers', metadata,           Column('manager_id', Integer, primary_key=True),           Column('manager_name', String(50))           )       s = select([engineers, managers]).where(engineers.c.engineer_name==managers.c.manager_name)              self.assertEquals(set(sql_util.reduce_columns(list(s.c), s)),        set([s.c.engineer_id, s.c.engineer_name, s.c.manager_id])        )           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(TestBase, AssertsExecutionResults):    def test_table(self):        meta = MetaData()        t1 = Table('t1', meta, Column('c1', Integer, primary_key=True), Column('c2', String(30)))        t2 = Table('t2', meta, Column('c1', Integer, primary_key=True), Column('c2', String(30)))        assert t1.is_derived_from(t1)        assert not t2.is_derived_from(t1)    def test_alias(self):        meta = MetaData()        t1 = Table('t1', meta, Column('c1', Integer, primary_key=True), Column('c2', String(30)))        t2 = Table('t2', meta, Column('c1', Integer, primary_key=True), Column('c2', String(30)))        assert t1.alias().is_derived_from(t1)        assert not t2.alias().is_derived_from(t1)        assert not t1.is_derived_from(t1.alias())        assert not t1.is_derived_from(t2.alias())    def test_select(self):        meta = MetaData()        t1 = Table('t1', meta, Column('c1', Integer, primary_key=True), Column('c2', String(30)))        t2 = Table('t2', meta, Column('c1', Integer, primary_key=True), Column('c2', String(30)))        assert t1.select().is_derived_from(t1)        assert not t2.select().is_derived_from(t1)        assert select([t1, t2]).is_derived_from(t1)        assert t1.select().alias('foo').is_derived_from(t1)        assert select([t1, t2]).alias('foo').is_derived_from(t1)        assert not t2.select().alias('foo').is_derived_from(t1)if __name__ == "__main__":    testenv.main()