"""Test various algorithmic properties of selectables."""

from sqlalchemy.test.testing import eq_, assert_raises, \
    assert_raises_message
from sqlalchemy import *
from sqlalchemy.test import *
from sqlalchemy.sql import util as sql_util, visitors
from sqlalchemy import exc
from sqlalchemy.sql import table, column, null
from sqlalchemy import util

metadata = MetaData()
table1 = Table('table1', metadata,
    Column('col1', Integer, primary_key=True),
    Column('col2', String(20)),
    Column('col3', Integer),
    Column('colx', Integer),

)

table2 = Table('table2', metadata,
    Column('col1', Integer, primary_key=True),
    Column('col2', Integer, ForeignKey('table1.col1')),
    Column('col3', String(20)),
    Column('coly', Integer),
)


class SelectableTest(TestBase, AssertsExecutionResults):

    def test_distance_on_labels(self):

        # same column three times

        s = select([table1.c.col1.label('c2'), table1.c.col1,
                   table1.c.col1.label('c1')])

        # didnt do this yet...col.label().make_proxy() has same
        # "distance" as col.make_proxy() so far assert
        # s.corresponding_column(table1.c.col1) is s.c.col1

        assert s.corresponding_column(s.c.col1) is s.c.col1
        assert s.corresponding_column(s.c.c1) is s.c.c1

    def test_distance_on_aliases(self):
        a1 = table1.alias('a1')
        for s in (select([a1, table1], use_labels=True),
                  select([table1, a1], use_labels=True)):
            assert s.corresponding_column(table1.c.col1) \
                is s.c.table1_col1
            assert s.corresponding_column(a1.c.col1) is s.c.a1_col1

    def test_join_against_self(self):
        jj = select([table1.c.col1.label('bar_col1')])
        jjj = join(table1, jj, table1.c.col1 == jj.c.bar_col1)

        # test column directly agaisnt itself

        assert jjj.corresponding_column(jjj.c.table1_col1) \
            is jjj.c.table1_col1
        assert jjj.corresponding_column(jj.c.bar_col1) is jjj.c.bar_col1

        # test alias of the join

        j2 = jjj.alias('foo')
        assert j2.corresponding_column(table1.c.col1) \
            is j2.c.table1_col1
    
    def test_against_cloned_non_table(self):
        # test that corresponding column digs across
        # clone boundaries with anonymous labeled elements
        col = func.count().label('foo')
        sel = select([col])
        
        sel2 = visitors.ReplacingCloningVisitor().traverse(sel)
        assert sel2.corresponding_column(col) is sel2.c.foo

        sel3 = visitors.ReplacingCloningVisitor().traverse(sel2)
        assert sel3.corresponding_column(col) is sel3.c.foo

        
    def test_select_on_table(self):
        sel = select([table1, table2], use_labels=True)

        assert sel.corresponding_column(table1.c.col1) \
            is sel.c.table1_col1
        assert sel.corresponding_column(table1.c.col1,
                require_embedded=True) is sel.c.table1_col1
        assert table1.corresponding_column(sel.c.table1_col1) \
            is table1.c.col1
        assert table1.corresponding_column(sel.c.table1_col1,
                require_embedded=True) is None


    def test_join_against_join(self):
        j = outerjoin(table1, table2, table1.c.col1 == table2.c.col2)
        jj = select([table1.c.col1.label('bar_col1')],
                    from_obj=[j]).alias('foo')
        jjj = join(table1, jj, table1.c.col1 == jj.c.bar_col1)
        assert jjj.corresponding_column(jjj.c.table1_col1) \
            is jjj.c.table1_col1
        j2 = jjj.alias('foo')
        assert j2.corresponding_column(jjj.c.table1_col1) \
            is j2.c.table1_col1
        assert jjj.corresponding_column(jj.c.bar_col1) is jj.c.bar_col1

    def test_table_alias(self):
        a = table1.alias('a')

        j = join(a, table2)

        criterion = a.c.col1 == table2.c.col2
        self.assert_(criterion.compare(j.onclause))


    def test_union(self):

        # tests that we can correspond a column in a Select statement
        # with a certain Table, against a column in a Union where one of
        # its underlying Selects matches to that same Table

        u = select([table1.c.col1, table1.c.col2, table1.c.col3,
                   table1.c.colx, null().label('coly'
                   )]).union(select([table2.c.col1, table2.c.col2,
                             table2.c.col3, null().label('colx'),
                             table2.c.coly]))
        s1 = table1.select(use_labels=True)
        s2 = table2.select(use_labels=True)
        c = u.corresponding_column(s1.c.table1_col2)
        assert u.corresponding_column(s1.c.table1_col2) is u.c.col2
        assert u.corresponding_column(s2.c.table2_col2) is u.c.col2

    def test_union_precedence(self):
        # conflicting column correspondence should be resolved based on 
        # the order of the select()s in the union
        
        s1 = select([table1.c.col1, table1.c.col2])
        s2 = select([table1.c.col2, table1.c.col1])
        s3 = select([table1.c.col3, table1.c.colx])
        s4 = select([table1.c.colx, table1.c.col3])
        
        u1 = union(s1, s2)
        assert u1.corresponding_column(table1.c.col1) is u1.c.col1
        assert u1.corresponding_column(table1.c.col2) is u1.c.col2
        
        u1 = union(s1, s2, s3, s4)
        assert u1.corresponding_column(table1.c.col1) is u1.c.col1
        assert u1.corresponding_column(table1.c.col2) is u1.c.col2
        assert u1.corresponding_column(table1.c.colx) is u1.c.col2
        assert u1.corresponding_column(table1.c.col3) is u1.c.col1
        

    def test_singular_union(self):
        u = union(select([table1.c.col1, table1.c.col2,
                  table1.c.col3]), select([table1.c.col1,
                  table1.c.col2, table1.c.col3]))
        u = union(select([table1.c.col1, table1.c.col2, table1.c.col3]))
        assert u.c.col1 is not None
        assert u.c.col2 is not None
        assert u.c.col3 is not None

    def test_alias_union(self):

        # same as testunion, except its an alias of the union

        u = select([table1.c.col1, table1.c.col2, table1.c.col3,
                   table1.c.colx, null().label('coly'
                   )]).union(select([table2.c.col1, table2.c.col2,
                             table2.c.col3, null().label('colx'),
                             table2.c.coly])).alias('analias')
        s1 = table1.select(use_labels=True)
        s2 = table2.select(use_labels=True)
        assert u.corresponding_column(s1.c.table1_col2) is u.c.col2
        assert u.corresponding_column(s2.c.table2_col2) is u.c.col2
        assert u.corresponding_column(s2.c.table2_coly) is u.c.coly
        assert s2.corresponding_column(u.c.coly) is s2.c.table2_coly

    def test_select_union(self):

        # like testaliasunion, but off a Select off the union.

        u = select([table1.c.col1, table1.c.col2, table1.c.col3,
                   table1.c.colx, null().label('coly'
                   )]).union(select([table2.c.col1, table2.c.col2,
                             table2.c.col3, null().label('colx'),
                             table2.c.coly])).alias('analias')
        s = select([u])
        s1 = table1.select(use_labels=True)
        s2 = table2.select(use_labels=True)
        assert s.corresponding_column(s1.c.table1_col2) is s.c.col2
        assert s.corresponding_column(s2.c.table2_col2) is s.c.col2

    def test_union_against_join(self):

        # same as testunion, except its an alias of the union

        u = select([table1.c.col1, table1.c.col2, table1.c.col3,
                   table1.c.colx, null().label('coly'
                   )]).union(select([table2.c.col1, table2.c.col2,
                             table2.c.col3, null().label('colx'),
                             table2.c.coly])).alias('analias')
        j1 = table1.join(table2)
        assert u.corresponding_column(j1.c.table1_colx) is u.c.colx
        assert j1.corresponding_column(u.c.colx) is j1.c.table1_colx

    def test_join(self):
        a = join(table1, table2)
        print str(a.select(use_labels=True))
        b = table2.alias('b')
        j = join(a, b)
        print str(j)
        criterion = a.c.table1_col1 == b.c.col2
        self.assert_(criterion.compare(j.onclause))

    def test_select_alias(self):
        a = table1.select().alias('a')
        j = join(a, table2)

        criterion = a.c.col1 == table2.c.col2
        self.assert_(criterion.compare(j.onclause))

    def test_select_labels(self):
        a = table1.select(use_labels=True)
        j = join(a, table2)

        criterion = a.c.table1_col1 == table2.c.col2
        self.assert_(criterion.compare(j.onclause))


    def test_column_labels(self):
        a = select([table1.c.col1.label('acol1'),
                   table1.c.col2.label('acol2'),
                   table1.c.col3.label('acol3')])
        j = join(a, table2)
        criterion = a.c.acol1 == table2.c.col2
        self.assert_(criterion.compare(j.onclause))
    
    def test_labeled_select_correspoinding(self):
        l1 = select([func.max(table1.c.col1)]).label('foo')

        s = select([l1])
        eq_(s.corresponding_column(l1), s.c.foo)
        
        s = select([table1.c.col1, l1])
        eq_(s.corresponding_column(l1), s.c.foo)

    def test_select_alias_labels(self):
        a = table2.select(use_labels=True).alias('a')
        j = join(a, table1)

        criterion = table1.c.col1 == a.c.table2_col2
        self.assert_(criterion.compare(j.onclause))

    def test_table_joined_to_select_of_table(self):
        metadata = MetaData()
        a = Table('a', metadata,
            Column('id', Integer, primary_key=True))
        b = Table('b', metadata,
            Column('id', Integer, primary_key=True),
            Column('aid', Integer, ForeignKey('a.id')),
            )

        j1 = a.outerjoin(b)
        j2 = select([a.c.id.label('aid')]).alias('bar')

        j3 = a.join(j2, j2.c.aid==a.c.id)

        j4 = select([j3]).alias('foo')
        assert j4.corresponding_column(j2.c.aid) is j4.c.aid
        assert j4.corresponding_column(a.c.id) is j4.c.id

    def test_two_metadata_join_raises(self):
        m = MetaData()
        m2 = MetaData()

        t1 = Table('t1', m, Column('id', Integer), Column('id2', Integer))
        t2 = Table('t2', m, Column('id', Integer, ForeignKey('t1.id')))
        t3 = Table('t3', m2, Column('id', Integer, ForeignKey('t1.id2')))

        s = select([t2, t3], use_labels=True)

        assert_raises(exc.NoReferencedTableError, s.join, t1)
    

    def test_join_condition(self):
        m = MetaData()
        t1 = Table('t1', m, Column('id', Integer))
        t2 = Table('t2', m, Column('id', Integer), Column('t1id',
                   ForeignKey('t1.id')))
        t3 = Table('t3', m, Column('id', Integer), Column('t1id',
                   ForeignKey('t1.id')), Column('t2id',
                   ForeignKey('t2.id')))
        t4 = Table('t4', m, Column('id', Integer), Column('t2id',
                   ForeignKey('t2.id')))
        t1t2 = t1.join(t2)
        t2t3 = t2.join(t3)
        for (left, right, a_subset, expected) in [
            (t1, t2, None, t1.c.id == t2.c.t1id),
            (t1t2, t3, t2, t1t2.c.t2_id == t3.c.t2id),
            (t2t3, t1, t3, t1.c.id == t3.c.t1id),
            (t2t3, t4, None, t2t3.c.t2_id == t4.c.t2id),
            (t2t3, t4, t3, t2t3.c.t2_id == t4.c.t2id),
            (t2t3.join(t1), t4, None, t2t3.c.t2_id == t4.c.t2id),
            (t2t3.join(t1), t4, t1, t2t3.c.t2_id == t4.c.t2id),
            (t1t2, t2t3, t2, t1t2.c.t2_id == t2t3.c.t3_t2id),
            ]:
            assert expected.compare(sql_util.join_condition(left,
                                    right, a_subset=a_subset))
        
        # these are ambiguous, or have no joins
        for left, right, a_subset in [
            (t1t2, t3, None),
            (t2t3, t1, None),
            (t1, t4, None),
            (t1t2, t2t3, None),
        ]:
            assert_raises(
                exc.ArgumentError,
                sql_util.join_condition,
                left, right, a_subset=a_subset
            )
        
        als = t2t3.alias()
        # test join's behavior, including natural
        for left, right, expected in [
            (t1, t2, t1.c.id==t2.c.t1id),
            (t1t2, t3, t1t2.c.t2_id==t3.c.t2id),
            (t2t3, t1, t1.c.id==t3.c.t1id),
            (t2t3, t4, t2t3.c.t2_id==t4.c.t2id),
            (t2t3, t4, t2t3.c.t2_id==t4.c.t2id),
            (t2t3.join(t1), t4, t2t3.c.t2_id==t4.c.t2id),
            (t2t3.join(t1), t4, t2t3.c.t2_id==t4.c.t2id),
            (t1t2, als, t1t2.c.t2_id==als.c.t3_t2id)
        ]:
            assert expected.compare(
                left.join(right).onclause
            )



        # TODO: this raises due to right side being "grouped", and no
        # longer has FKs.  Did we want to make _FromGrouping friendlier
        # ?

        assert_raises_message(exc.ArgumentError,
                              "Perhaps you meant to convert the right "
                              "side to a subquery using alias\(\)\?",
                              t1t2.join, t2t3)
        assert_raises_message(exc.ArgumentError,
                              "Perhaps you meant to convert the right "
                              "side to a subquery using alias\(\)\?",
                              t1t2.join, t2t3.select(use_labels=True))
        

class PrimaryKeyTest(TestBase, AssertsExecutionResults):

    def test_join_pk_collapse_implicit(self):
        """test that redundant columns in a join get 'collapsed' into a
        minimal primary key, which is the root column along a chain of
        foreign key relationships."""

        meta = MetaData()
        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) == [a.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) \
            == [b.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) == [b.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 = :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))


        eq_(util.column_set(employee.join(engineer, employee.c.id
            == engineer.c.id).primary_key),
            util.column_set([employee.c.id]))
        eq_(util.column_set(employee.join(engineer, engineer.c.id
            == employee.c.id).primary_key),
            util.column_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)))
        
        eq_(util.column_set(sql_util.reduce_columns([
            t1.c.t1id,
            t1.c.t1data,
            t2.c.t2id,
            t2.c.t2data,
            t3.c.t3id,
            t3.c.t3data,
            ])), util.column_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)
        eq_(util.column_set(sql_util.reduce_columns(list(s.c), s)),
            util.column_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'
                )
        eq_(util.column_set(sql_util.reduce_columns([pjoin.c.people_person_id,
            pjoin.c.engineers_person_id, pjoin.c.managers_person_id])),
            util.column_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')
        eq_(util.column_set(sql_util.reduce_columns([item_join.c.id,
            item_join.c.dummy, item_join.c.child_name])),
            util.column_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))

        # this is essentially the union formed by the ORM's
        # polymorphic_union function. we define two versions with
        # different ordering of selects.
        #
        # the first selectable has the "real" column
        # classified_page.magazine_page_id

        pjoin = union(
                    select([
                        page_table.c.id,
                        magazine_page_table.c.page_id,
                        classified_page_table.c.magazine_page_id
                    ]).
                     select_from(
                        page_table.join(magazine_page_table).
                        join(classified_page_table)),
                        
                    select([
                        page_table.c.id,
                        magazine_page_table.c.page_id, 
                        cast(null(), Integer).label('magazine_page_id')
                    ]).
                    select_from(page_table.join(magazine_page_table))
                    ).alias('pjoin')
        eq_(util.column_set(sql_util.reduce_columns([pjoin.c.id,
            pjoin.c.page_id, pjoin.c.magazine_page_id])),
            util.column_set([pjoin.c.id]))

        # the first selectable has a CAST, which is a placeholder for
        # classified_page.magazine_page_id in the second selectable.
        # reduce_columns needs to take into account all foreign keys
        # derived from pjoin.c.magazine_page_id. the UNION construct
        # currently makes the external column look like that of the
        # first selectable only.

        pjoin = union(select([
                        page_table.c.id,
                        magazine_page_table.c.page_id, 
                        cast(null(), Integer).label('magazine_page_id')
                      ]).
                      select_from(page_table.join(magazine_page_table)),
                      
                      select([
                        page_table.c.id,
                        magazine_page_table.c.page_id,
                        classified_page_table.c.magazine_page_id
                      ]).
                      select_from(page_table.join(magazine_page_table).
                      join(classified_page_table))
                ).alias('pjoin')
        eq_(util.column_set(sql_util.reduce_columns([pjoin.c.id,
            pjoin.c.page_id, pjoin.c.magazine_page_id])),
            util.column_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)

class AnnotationsTest(TestBase):
    def test_custom_constructions(self):
        from sqlalchemy.schema import Column
        class MyColumn(Column):
            def __init__(self):
                Column.__init__(self, 'foo', Integer)
            _constructor = Column
            
        t1 = Table('t1', MetaData(), MyColumn())
        s1 = t1.select()
        assert isinstance(t1.c.foo, MyColumn)
        assert isinstance(s1.c.foo, Column)

        annot_1 = t1.c.foo._annotate({})
        s2 = select([annot_1])
        assert isinstance(s2.c.foo, Column)
        annot_2 = s1._annotate({})
        assert isinstance(annot_2.c.foo, Column)
        
    def test_annotated_corresponding_column(self):
        table1 = table('table1', column("col1"))
        
        s1 = select([table1.c.col1])
        t1 = s1._annotate({})
        t2 = s1
        
        # t1 needs to share the same _make_proxy() columns as t2, even
        # though it's annotated.  otherwise paths will diverge once they
        # are corresponded against "inner" below.

        assert t1.c is t2.c
        assert t1.c.col1 is t2.c.col1

        inner = select([s1]) 

        assert inner.corresponding_column(t2.c.col1,
                require_embedded=False) \
            is inner.corresponding_column(t2.c.col1,
                require_embedded=True) is inner.c.col1
        assert inner.corresponding_column(t1.c.col1,
                require_embedded=False) \
            is inner.corresponding_column(t1.c.col1,
                require_embedded=True) is inner.c.col1

    def test_annotated_visit(self):
        table1 = table('table1', column("col1"), column("col2"))
        
        bin = table1.c.col1 == bindparam('foo', value=None)
        assert str(bin) == "table1.col1 = :foo"
        def visit_binary(b):
            b.right = table1.c.col2
            
        b2 = visitors.cloned_traverse(bin, {}, {'binary':visit_binary})
        assert str(b2) == "table1.col1 = table1.col2"


        b3 = visitors.cloned_traverse(bin._annotate({}), {}, {'binary'
                : visit_binary})
        assert str(b3) == 'table1.col1 = table1.col2'

        def visit_binary(b):
            b.left = bindparam('bar')
        
        b4 = visitors.cloned_traverse(b2, {}, {'binary':visit_binary})
        assert str(b4) == ":bar = table1.col2"

        b5 = visitors.cloned_traverse(b3, {}, {'binary':visit_binary})
        assert str(b5) == ":bar = table1.col2"
    

    def test_annotate_expressions(self):
        table1 = table('table1', column('col1'), column('col2'))
        for expr, expected in [(table1.c.col1, 'table1.col1'),
                               (table1.c.col1 == 5,
                               'table1.col1 = :col1_1'),
                               (table1.c.col1.in_([2, 3, 4]),
                               'table1.col1 IN (:col1_1, :col1_2, '
                               ':col1_3)')]:
            eq_(str(expr), expected)
            eq_(str(expr._annotate({})), expected)
            eq_(str(sql_util._deep_annotate(expr, {})), expected)
            eq_(str(sql_util._deep_annotate(expr, {},
                exclude=[table1.c.col1])), expected)
        
    def test_deannotate(self):
        table1 = table('table1', column("col1"), column("col2"))
        
        bin = table1.c.col1 == bindparam('foo', value=None)

        b2 = sql_util._deep_annotate(bin, {'_orm_adapt':True})
        b3 = sql_util._deep_deannotate(b2)
        b4 = sql_util._deep_deannotate(bin)

        for elem in (b2._annotations, b2.left._annotations):
            assert '_orm_adapt' in elem
        
        for elem in b3._annotations, b3.left._annotations, \
            b4._annotations, b4.left._annotations:
            assert elem == {}

        assert b2.left is not bin.left 
        assert b3.left is not b2.left is not bin.left
        assert b4.left is bin.left  # since column is immutable
        assert b4.right is not bin.right is not b2.right is not b3.right