"""Miscellaneous inheritance-related tests, many very old.
These are generally tests derived from specific user issues.

"""

from sqlalchemy.testing import eq_
from sqlalchemy import *
from sqlalchemy import util
from sqlalchemy.orm import *
from sqlalchemy.orm.interfaces import MANYTOONE
from sqlalchemy.testing import AssertsExecutionResults
from sqlalchemy import testing
from sqlalchemy.testing.util import function_named
from sqlalchemy.testing import fixtures
from test.orm import _fixtures
from sqlalchemy.testing import eq_
from sqlalchemy.testing.schema import Table, Column


class AttrSettable(object):
    def __init__(self, **kwargs):
        [setattr(self, k, v) for k, v in kwargs.items()]
    def __repr__(self):
        return self.__class__.__name__ + "(%s)" % (hex(id(self)))


class RelationshipTest1(fixtures.MappedTest):
    """test self-referential relationships on polymorphic mappers"""
    @classmethod
    def define_tables(cls, metadata):
        global people, managers

        people = Table('people', metadata,
           Column('person_id', Integer, Sequence('person_id_seq',
                                        optional=True),
                                        primary_key=True),
           Column('manager_id', Integer,
                                ForeignKey('managers.person_id',
                                use_alter=True, name="mpid_fq")),
           Column('name', String(50)),
           Column('type', String(30)))

        managers = Table('managers', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'),
                                    primary_key=True),
           Column('status', String(30)),
           Column('manager_name', String(50))
           )

    def teardown(self):
        people.update(values={people.c.manager_id:None}).execute()
        super(RelationshipTest1, self).teardown()

    def test_parent_refs_descendant(self):
        class Person(AttrSettable):
            pass
        class Manager(Person):
            pass

        mapper(Person, people, properties={
            'manager':relationship(Manager, primaryjoin=(
                                people.c.manager_id ==
                                        managers.c.person_id),
                               uselist=False, post_update=True)
        })
        mapper(Manager, managers, inherits=Person,
               inherit_condition=people.c.person_id==managers.c.person_id)

        eq_(class_mapper(Person).get_property('manager').synchronize_pairs,
                [(managers.c.person_id,people.c.manager_id)])

        session = create_session()
        p = Person(name='some person')
        m = Manager(name='some manager')
        p.manager = m
        session.add(p)
        session.flush()
        session.expunge_all()

        p = session.query(Person).get(p.person_id)
        m = session.query(Manager).get(m.person_id)
        assert p.manager is m

    def test_descendant_refs_parent(self):
        class Person(AttrSettable):
            pass
        class Manager(Person):
            pass

        mapper(Person, people)
        mapper(Manager, managers, inherits=Person,
                            inherit_condition=people.c.person_id==
                                        managers.c.person_id,
        properties={
            'employee':relationship(Person, primaryjoin=(
                                            people.c.manager_id ==
                                                     managers.c.person_id),
                                foreign_keys=[people.c.manager_id],
                                uselist=False, post_update=True)
        })

        session = create_session()
        p = Person(name='some person')
        m = Manager(name='some manager')
        m.employee = p
        session.add(m)
        session.flush()
        session.expunge_all()

        p = session.query(Person).get(p.person_id)
        m = session.query(Manager).get(m.person_id)
        assert m.employee is p

class RelationshipTest2(fixtures.MappedTest):
    """test self-referential relationships on polymorphic mappers"""
    @classmethod
    def define_tables(cls, metadata):
        global people, managers, data
        people = Table('people', metadata,
           Column('person_id', Integer, primary_key=True,
                                        test_needs_autoincrement=True),
           Column('name', String(50)),
           Column('type', String(30)))

        managers = Table('managers', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'),
                                                primary_key=True),
           Column('manager_id', Integer, ForeignKey('people.person_id')),
           Column('status', String(30)),
           )

        data = Table('data', metadata,
            Column('person_id', Integer, ForeignKey('managers.person_id'),
                                                primary_key=True),
            Column('data', String(30))
            )

    def testrelationshiponsubclass_j1_nodata(self):
        self.do_test("join1", False)
    def testrelationshiponsubclass_j2_nodata(self):
        self.do_test("join2", False)
    def testrelationshiponsubclass_j1_data(self):
        self.do_test("join1", True)
    def testrelationshiponsubclass_j2_data(self):
        self.do_test("join2", True)
    def testrelationshiponsubclass_j3_nodata(self):
        self.do_test("join3", False)
    def testrelationshiponsubclass_j3_data(self):
        self.do_test("join3", True)

    def do_test(self, jointype="join1", usedata=False):
        class Person(AttrSettable):
            pass
        class Manager(Person):
            pass

        if jointype == "join1":
            poly_union = polymorphic_union({
                'person':people.select(people.c.type=='person'),
                'manager':join(people, managers,
                        people.c.person_id==managers.c.person_id)
            }, None)
            polymorphic_on=poly_union.c.type
        elif jointype == "join2":
            poly_union = polymorphic_union({
                'person':people.select(people.c.type=='person'),
                'manager':managers.join(people,
                        people.c.person_id==managers.c.person_id)
            }, None)
            polymorphic_on=poly_union.c.type
        elif jointype == "join3":
            poly_union = None
            polymorphic_on = people.c.type

        if usedata:
            class Data(object):
                def __init__(self, data):
                    self.data = data
            mapper(Data, data)

        mapper(Person, people,
                            with_polymorphic=('*', poly_union),
                            polymorphic_identity='person',
                            polymorphic_on=polymorphic_on)

        if usedata:
            mapper(Manager, managers,
                            inherits=Person,
                            inherit_condition=people.c.person_id==
                                    managers.c.person_id,
                            polymorphic_identity='manager',
                  properties={
                    'colleague':relationship(
                                Person,
                                primaryjoin=managers.c.manager_id==
                                    people.c.person_id,
                                lazy='select', uselist=False),
                    'data':relationship(Data, uselist=False)
                 }
            )
        else:
            mapper(Manager, managers, inherits=Person,
                                inherit_condition=people.c.person_id==
                                    managers.c.person_id,
                                    polymorphic_identity='manager',
                  properties={
                    'colleague':relationship(Person,
                                primaryjoin=managers.c.manager_id==
                                        people.c.person_id,
                                        lazy='select', uselist=False)
                 }
            )

        sess = create_session()
        p = Person(name='person1')
        m = Manager(name='manager1')
        m.colleague = p
        if usedata:
            m.data = Data('ms data')
        sess.add(m)
        sess.flush()

        sess.expunge_all()
        p = sess.query(Person).get(p.person_id)
        m = sess.query(Manager).get(m.person_id)
        assert m.colleague is p
        if usedata:
            assert m.data.data == 'ms data'

class RelationshipTest3(fixtures.MappedTest):
    """test self-referential relationships on polymorphic mappers"""
    @classmethod
    def define_tables(cls, metadata):
        global people, managers, data
        people = Table('people', metadata,
           Column('person_id', Integer, primary_key=True,
                                            test_needs_autoincrement=True),
           Column('colleague_id', Integer, ForeignKey('people.person_id')),
           Column('name', String(50)),
           Column('type', String(30)))

        managers = Table('managers', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'),
                                            primary_key=True),
           Column('status', String(30)),
           )

        data = Table('data', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'),
                                            primary_key=True),
           Column('data', String(30))
           )

def _generate_test(jointype="join1", usedata=False):
    def do_test(self):
        class Person(AttrSettable):
            pass
        class Manager(Person):
            pass

        if usedata:
            class Data(object):
                def __init__(self, data):
                    self.data = data

        if jointype == "join1":
            poly_union = polymorphic_union({
                'manager':managers.join(people,
                                    people.c.person_id==managers.c.person_id),
                'person':people.select(people.c.type=='person')
            }, None)
        elif jointype =="join2":
            poly_union = polymorphic_union({
                'manager':join(people, managers,
                                    people.c.person_id==managers.c.person_id),
                'person':people.select(people.c.type=='person')
            }, None)
        elif jointype == 'join3':
            poly_union = people.outerjoin(managers)
        elif jointype == "join4":
            poly_union=None

        if usedata:
            mapper(Data, data)

        if usedata:
            mapper(Person, people,
                                with_polymorphic=('*', poly_union),
                                polymorphic_identity='person',
                                polymorphic_on=people.c.type,
                  properties={
                    'colleagues':relationship(Person,
                                primaryjoin=people.c.colleague_id==
                                        people.c.person_id,
                                remote_side=people.c.colleague_id,
                                uselist=True),
                    'data':relationship(Data, uselist=False)
                    }
            )
        else:
            mapper(Person, people,
                    with_polymorphic=('*', poly_union),
                    polymorphic_identity='person',
                    polymorphic_on=people.c.type,
                  properties={
                    'colleagues':relationship(Person,
                        primaryjoin=people.c.colleague_id==people.c.person_id,
                        remote_side=people.c.colleague_id, uselist=True)
                    }
            )

        mapper(Manager, managers, inherits=Person,
                    inherit_condition=people.c.person_id==
                        managers.c.person_id,
                        polymorphic_identity='manager')

        sess = create_session()
        p = Person(name='person1')
        p2 = Person(name='person2')
        p3 = Person(name='person3')
        m = Manager(name='manager1')
        p.colleagues.append(p2)
        m.colleagues.append(p3)
        if usedata:
            p.data = Data('ps data')
            m.data = Data('ms data')

        sess.add(m)
        sess.add(p)
        sess.flush()

        sess.expunge_all()
        p = sess.query(Person).get(p.person_id)
        p2 = sess.query(Person).get(p2.person_id)
        p3 = sess.query(Person).get(p3.person_id)
        m = sess.query(Person).get(m.person_id)
        assert len(p.colleagues) == 1
        assert p.colleagues == [p2]
        assert m.colleagues == [p3]
        if usedata:
            assert p.data.data == 'ps data'
            assert m.data.data == 'ms data'

    do_test = function_named(
        do_test, 'test_relationship_on_base_class_%s_%s' % (
        jointype, data and "nodata" or "data"))
    return do_test

for jointype in ["join1", "join2", "join3", "join4"]:
    for data in (True, False):
        func = _generate_test(jointype, data)
        setattr(RelationshipTest3, func.__name__, func)
del func

class RelationshipTest4(fixtures.MappedTest):
    @classmethod
    def define_tables(cls, metadata):
        global people, engineers, managers, cars
        people = Table('people', metadata,
           Column('person_id', Integer, primary_key=True,
                                    test_needs_autoincrement=True),
           Column('name', String(50)))

        engineers = Table('engineers', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'),
                                    primary_key=True),
           Column('status', String(30)))

        managers = Table('managers', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'),
                                    primary_key=True),
           Column('longer_status', String(70)))

        cars = Table('cars', metadata,
           Column('car_id', Integer, primary_key=True,
                                    test_needs_autoincrement=True),
           Column('owner', Integer, ForeignKey('people.person_id')))

    def test_many_to_one_polymorphic(self):
        """in this test, the polymorphic union is between two subclasses, but
        does not include the base table by itself in the union. however, the
        primaryjoin condition is going to be against the base table, and its a
        many-to-one relationship (unlike the test in polymorph.py) so the
        column in the base table is explicit. Can the ClauseAdapter figure out
        how to alias the primaryjoin to the polymorphic union ?"""

        # class definitions
        class Person(object):
            def __init__(self, **kwargs):
                for key, value in kwargs.items():
                    setattr(self, key, value)
            def __repr__(self):
                return "Ordinary person %s" % self.name
        class Engineer(Person):
            def __repr__(self):
                return "Engineer %s, status %s" % \
                        (self.name, self.status)
        class Manager(Person):
            def __repr__(self):
                return "Manager %s, status %s" % \
                        (self.name, self.longer_status)
        class Car(object):
            def __init__(self, **kwargs):
                for key, value in kwargs.items():
                    setattr(self, key, value)
            def __repr__(self):
                return "Car number %d" % self.car_id

        # create a union that represents both types of joins.
        employee_join = polymorphic_union(
            {
                'engineer':people.join(engineers),
                'manager':people.join(managers),
            }, "type", 'employee_join')

        person_mapper   = mapper(Person, people,
                                    with_polymorphic=('*', employee_join),
                                    polymorphic_on=employee_join.c.type,
                                    polymorphic_identity='person')
        engineer_mapper = mapper(Engineer, engineers,
                                    inherits=person_mapper,
                                    polymorphic_identity='engineer')
        manager_mapper  = mapper(Manager, managers,
                                    inherits=person_mapper,
                                    polymorphic_identity='manager')
        car_mapper      = mapper(Car, cars,
                                    properties= {'employee':
                                            relationship(person_mapper)})

        session = create_session()

        # creating 5 managers named from M1 to E5
        for i in range(1,5):
            session.add(Manager(name="M%d" % i,
                                longer_status="YYYYYYYYY"))
        # creating 5 engineers named from E1 to E5
        for i in range(1,5):
            session.add(Engineer(name="E%d" % i,status="X"))

        session.flush()

        engineer4 = session.query(Engineer).\
                                filter(Engineer.name=="E4").first()
        manager3 = session.query(Manager).\
                                filter(Manager.name=="M3").first()

        car1 = Car(employee=engineer4)
        session.add(car1)
        car2 = Car(employee=manager3)
        session.add(car2)
        session.flush()

        session.expunge_all()

        def go():
            testcar = session.query(Car).options(
                                    joinedload('employee')
                                    ).get(car1.car_id)
            assert str(testcar.employee) == "Engineer E4, status X"
        self.assert_sql_count(testing.db, go, 1)

        car1 = session.query(Car).get(car1.car_id)
        usingGet = session.query(person_mapper).get(car1.owner)
        usingProperty = car1.employee

        assert str(engineer4) == "Engineer E4, status X"
        assert str(usingGet) == "Engineer E4, status X"
        assert str(usingProperty) == "Engineer E4, status X"

        session.expunge_all()
        # and now for the lightning round, eager !

        def go():
            testcar = session.query(Car).options(
                                    joinedload('employee')
                                    ).get(car1.car_id)
            assert str(testcar.employee) == "Engineer E4, status X"
        self.assert_sql_count(testing.db, go, 1)

        session.expunge_all()
        s = session.query(Car)
        c = s.join("employee").filter(Person.name=="E4")[0]
        assert c.car_id==car1.car_id

class RelationshipTest5(fixtures.MappedTest):
    @classmethod
    def define_tables(cls, metadata):
        global people, engineers, managers, cars
        people = Table('people', metadata,
           Column('person_id', Integer, primary_key=True,
                                test_needs_autoincrement=True),
           Column('name', String(50)),
           Column('type', String(50)))

        engineers = Table('engineers', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'),
                                primary_key=True),
           Column('status', String(30)))

        managers = Table('managers', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'),
                                primary_key=True),
           Column('longer_status', String(70)))

        cars = Table('cars', metadata,
           Column('car_id', Integer, primary_key=True,
                                test_needs_autoincrement=True),
           Column('owner', Integer, ForeignKey('people.person_id')))

    def test_eager_empty(self):
        """test parent object with child relationship to an inheriting mapper,
        using eager loads, works when there are no child objects present"""

        class Person(object):
            def __init__(self, **kwargs):
                for key, value in kwargs.items():
                    setattr(self, key, value)
            def __repr__(self):
                return "Ordinary person %s" % self.name
        class Engineer(Person):
            def __repr__(self):
                return "Engineer %s, status %s" % \
                        (self.name, self.status)
        class Manager(Person):
            def __repr__(self):
                return "Manager %s, status %s" % \
                        (self.name, self.longer_status)
        class Car(object):
            def __init__(self, **kwargs):
                for key, value in kwargs.items():
                    setattr(self, key, value)
            def __repr__(self):
                return "Car number %d" % self.car_id

        person_mapper   = mapper(Person, people,
                                    polymorphic_on=people.c.type,
                                    polymorphic_identity='person')
        engineer_mapper = mapper(Engineer, engineers,
                                    inherits=person_mapper,
                                    polymorphic_identity='engineer')
        manager_mapper  = mapper(Manager, managers,
                                    inherits=person_mapper,
                                    polymorphic_identity='manager')
        car_mapper      = mapper(Car, cars, properties= {
                                    'manager':relationship(
                                        manager_mapper, lazy='joined')})

        sess = create_session()
        car1 = Car()
        car2 = Car()
        car2.manager = Manager()
        sess.add(car1)
        sess.add(car2)
        sess.flush()
        sess.expunge_all()

        carlist = sess.query(Car).all()
        assert carlist[0].manager is None
        assert carlist[1].manager.person_id == car2.manager.person_id

class RelationshipTest6(fixtures.MappedTest):
    """test self-referential relationships on a single joined-table
    inheritance mapper"""

    @classmethod
    def define_tables(cls, metadata):
        global people, managers, data
        people = Table('people', metadata,
           Column('person_id', Integer, primary_key=True,
                                test_needs_autoincrement=True),
           Column('name', String(50)),
           )

        managers = Table('managers', metadata,
           Column('person_id', Integer, ForeignKey('people.person_id'),
                                primary_key=True),
           Column('colleague_id', Integer,
                            ForeignKey('managers.person_id')),
           Column('status', String(30)),
           )

    def test_basic(self):
        class Person(AttrSettable):
            pass
        class Manager(Person):
            pass

        mapper(Person, people)

        mapper(Manager, managers, inherits=Person,
                            inherit_condition=people.c.person_id==\
                                    managers.c.person_id,
              properties={
                'colleague':relationship(Manager,
                                primaryjoin=managers.c.colleague_id==\
                                    managers.c.person_id,
                                    lazy='select', uselist=False)
             }
        )

        sess = create_session()
        m = Manager(name='manager1')
        m2 =Manager(name='manager2')
        m.colleague = m2
        sess.add(m)
        sess.flush()

        sess.expunge_all()
        m = sess.query(Manager).get(m.person_id)
        m2 = sess.query(Manager).get(m2.person_id)
        assert m.colleague is m2

class RelationshipTest7(fixtures.MappedTest):
    @classmethod
    def define_tables(cls, metadata):
        global people, engineers, managers, cars, offroad_cars
        cars = Table('cars', metadata,
                Column('car_id', Integer, primary_key=True,
                                    test_needs_autoincrement=True),
                Column('name', String(30)))

        offroad_cars = Table('offroad_cars', metadata,
                Column('car_id',Integer, ForeignKey('cars.car_id'),
                                    nullable=False,primary_key=True))

        people = Table('people', metadata,
                Column('person_id', Integer, primary_key=True,
                                    test_needs_autoincrement=True),
                Column('car_id', Integer, ForeignKey('cars.car_id'),
                                    nullable=False),
                Column('name', String(50)))

        engineers = Table('engineers', metadata,
                Column('person_id', Integer, ForeignKey('people.person_id'),
                                    primary_key=True),
                Column('field', String(30)))


        managers = Table('managers', metadata,
                Column('person_id', Integer, ForeignKey('people.person_id'),
                                    primary_key=True),
                Column('category', String(70)))

    def test_manytoone_lazyload(self):
        """test that lazy load clause to a polymorphic child mapper generates
        correctly [ticket:493]"""

        class PersistentObject(object):
            def __init__(self, **kwargs):
                for key, value in kwargs.items():
                    setattr(self, key, value)

        class Status(PersistentObject):
            def __repr__(self):
                return "Status %s" % self.name

        class Person(PersistentObject):
            def __repr__(self):
                return "Ordinary person %s" % self.name

        class Engineer(Person):
            def __repr__(self):
                return "Engineer %s, field %s" % (self.name,
                                                self.field)

        class Manager(Person):
            def __repr__(self):
                return "Manager %s, category %s" % (self.name,
                                                self.category)

        class Car(PersistentObject):
            def __repr__(self):
                return "Car number %d, name %s" % \
                                        (self.car_id, self.name)

        class Offraod_Car(Car):
            def __repr__(self):
                return "Offroad Car number %d, name %s" % \
                                        (self.car_id,self.name)

        employee_join = polymorphic_union(
                {
                    'engineer':people.join(engineers),
                    'manager':people.join(managers),
                }, "type", 'employee_join')

        car_join = polymorphic_union(
            {
                'car' : cars.outerjoin(offroad_cars).\
                        select(offroad_cars.c.car_id == None).reduce_columns(),
                'offroad' : cars.join(offroad_cars)
            }, "type", 'car_join')

        car_mapper  = mapper(Car, cars,
                with_polymorphic=('*', car_join) ,polymorphic_on=car_join.c.type,
                polymorphic_identity='car',
                )
        offroad_car_mapper = mapper(Offraod_Car, offroad_cars,
                                    inherits=car_mapper, polymorphic_identity='offroad')
        person_mapper = mapper(Person, people,
                with_polymorphic=('*', employee_join),
                polymorphic_on=employee_join.c.type,
                polymorphic_identity='person',
                properties={
                    'car':relationship(car_mapper)
                    })
        engineer_mapper = mapper(Engineer, engineers,
                                        inherits=person_mapper,
                                        polymorphic_identity='engineer')
        manager_mapper  = mapper(Manager, managers,
                                        inherits=person_mapper,
                                        polymorphic_identity='manager')

        session = create_session()
        basic_car=Car(name="basic")
        offroad_car=Offraod_Car(name="offroad")

        for i in range(1,4):
            if i%2: