diff options
Diffstat (limited to 'test/orm/inheritance/polymorph2.py')
| -rw-r--r-- | test/orm/inheritance/polymorph2.py | 196 |
1 files changed, 101 insertions, 95 deletions
diff --git a/test/orm/inheritance/polymorph2.py b/test/orm/inheritance/polymorph2.py index 6454ab860..f87ea9357 100644 --- a/test/orm/inheritance/polymorph2.py +++ b/test/orm/inheritance/polymorph2.py @@ -17,13 +17,13 @@ class RelationTest1(ORMTest): def define_tables(self, metadata): global people, managers - people = Table('people', metadata, + 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, + managers = Table('managers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), Column('status', String(30)), Column('manager_name', String(50)) @@ -32,7 +32,7 @@ class RelationTest1(ORMTest): def tearDown(self): people.update(values={people.c.manager_id:None}).execute() super(RelationTest1, self).tearDown() - + def testparentrefsdescendant(self): class Person(AttrSettable): pass @@ -51,9 +51,13 @@ class RelationTest1(ORMTest): clear_mappers() mapper(Person, people, properties={ - 'manager':relation(Manager, primaryjoin=people.c.manager_id==managers.c.person_id, foreignkey=people.c.manager_id, uselist=False, post_update=True) + 'manager':relation(Manager, primaryjoin=(people.c.manager_id == + managers.c.person_id), + foreign_keys=[people.c.manager_id], + uselist=False, post_update=True) }) - mapper(Manager, managers, inherits=Person, inherit_condition=people.c.person_id==managers.c.person_id) + mapper(Manager, managers, inherits=Person, + inherit_condition=people.c.person_id==managers.c.person_id) session = create_session() p = Person(name='some person') @@ -76,7 +80,10 @@ class RelationTest1(ORMTest): mapper(Person, people) mapper(Manager, managers, inherits=Person, inherit_condition=people.c.person_id==managers.c.person_id, properties={ - 'employee':relation(Person, primaryjoin=people.c.manager_id==managers.c.person_id, foreignkey=people.c.manager_id, uselist=False, post_update=True) + 'employee':relation(Person, primaryjoin=(people.c.manager_id == + managers.c.person_id), + foreign_keys=[people.c.manager_id], + uselist=False, post_update=True) }) session = create_session() @@ -91,27 +98,27 @@ class RelationTest1(ORMTest): m = session.query(Manager).get(m.person_id) print p, m, m.employee assert m.employee is p - + class RelationTest2(ORMTest): """test self-referential relationships on polymorphic mappers""" def define_tables(self, metadata): global people, managers, data - people = Table('people', 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))) - managers = Table('managers', metadata, + 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 testrelationonsubclass_j1_nodata(self): self.do_test("join1", False) def testrelationonsubclass_j2_nodata(self): @@ -124,7 +131,7 @@ class RelationTest2(ORMTest): self.do_test("join3", False) def testrelationonsubclass_j3_data(self): self.do_test("join3", True) - + def do_test(self, jointype="join1", usedata=False): class Person(AttrSettable): pass @@ -146,13 +153,13 @@ class RelationTest2(ORMTest): 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, select_table=poly_union, polymorphic_identity='person', polymorphic_on=polymorphic_on) if usedata: @@ -177,7 +184,7 @@ class RelationTest2(ORMTest): m.data = Data('ms data') sess.save(m) sess.flush() - + sess.clear() p = sess.query(Person).get(p.person_id) m = sess.query(Manager).get(m.person_id) @@ -191,13 +198,13 @@ class RelationTest3(ORMTest): """test self-referential relationships on polymorphic mappers""" def define_tables(self, metadata): global people, managers, data - people = Table('people', metadata, + people = Table('people', metadata, Column('person_id', Integer, Sequence('person_id_seq', optional=True), primary_key=True), Column('colleague_id', Integer, ForeignKey('people.person_id')), Column('name', String(50)), Column('type', String(30))) - managers = Table('managers', metadata, + managers = Table('managers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), Column('status', String(30)), ) @@ -233,23 +240,23 @@ def generate_test(jointype="join1", usedata=False): poly_union = people.outerjoin(managers) elif jointype == "join4": poly_union=None - + if usedata: mapper(Data, data) - + if usedata: mapper(Person, people, select_table=poly_union, polymorphic_identity='person', polymorphic_on=people.c.type, properties={ 'colleagues':relation(Person, primaryjoin=people.c.colleague_id==people.c.person_id, remote_side=people.c.colleague_id, uselist=True), 'data':relation(Data, uselist=False) - } + } ) else: mapper(Person, people, select_table=poly_union, polymorphic_identity='person', polymorphic_on=people.c.type, properties={ - 'colleagues':relation(Person, primaryjoin=people.c.colleague_id==people.c.person_id, + 'colleagues':relation(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') @@ -268,7 +275,7 @@ def generate_test(jointype="join1", usedata=False): sess.save(m) sess.save(p) sess.flush() - + sess.clear() p = sess.query(Person).get(p.person_id) p2 = sess.query(Person).get(p2.person_id) @@ -281,41 +288,41 @@ def generate_test(jointype="join1", usedata=False): if usedata: assert p.data.data == 'ps data' assert m.data.data == 'ms data' - - do_test.__name__ = 'test_relationonbaseclass_%s_%s' % (jointype, data and "nodata" or "data") + + do_test.__name__ = 'test_relationonbaseclass_%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(RelationTest3, func.__name__, func) - - + + class RelationTest4(ORMTest): def define_tables(self, metadata): global people, engineers, managers, cars - people = Table('people', metadata, + people = Table('people', metadata, Column('person_id', Integer, primary_key=True), Column('name', String(50))) - engineers = Table('engineers', metadata, + engineers = Table('engineers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), Column('status', String(30))) - managers = Table('managers', metadata, + managers = Table('managers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), Column('longer_status', String(70))) - cars = Table('cars', metadata, + cars = Table('cars', metadata, Column('car_id', Integer, primary_key=True), Column('owner', Integer, ForeignKey('people.person_id'))) - + def testmanytoonepolymorphic(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): @@ -336,21 +343,21 @@ class RelationTest4(ORMTest): def __repr__(self): return "Car number %d" % self.car_id - # create a union that represents both types of joins. + # 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, select_table=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':relation(person_mapper)}) - + print class_mapper(Person).primary_key print person_mapper.get_select_mapper().primary_key - + session = create_session() # creating 5 managers named from M1 to E5 @@ -364,7 +371,7 @@ class RelationTest4(ORMTest): engineer4 = session.query(Engineer).filter(Engineer.name=="E4").first() manager3 = session.query(Manager).filter(Manager.name=="M3").first() - + car1 = Car(employee=engineer4) session.save(car1) car2 = Car(employee=manager3) @@ -372,7 +379,7 @@ class RelationTest4(ORMTest): session.flush() session.clear() - + print "----------------------------" car1 = session.query(Car).get(car1.car_id) print "----------------------------" @@ -395,7 +402,7 @@ class RelationTest4(ORMTest): testcar = session.query(Car).options(eagerload('employee')).get(car1.car_id) assert str(testcar.employee) == "Engineer E4, status X" self.assert_sql_count(testbase.db, go, 1) - + session.clear() s = session.query(Car) c = s.join("employee").filter(Person.name=="E4")[0] @@ -404,23 +411,23 @@ class RelationTest4(ORMTest): class RelationTest5(ORMTest): def define_tables(self, metadata): global people, engineers, managers, cars - people = Table('people', metadata, + people = Table('people', metadata, Column('person_id', Integer, primary_key=True), Column('name', String(50)), Column('type', String(50))) - engineers = Table('engineers', metadata, + engineers = Table('engineers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), Column('status', String(30))) - managers = Table('managers', metadata, + managers = Table('managers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), Column('longer_status', String(70))) - cars = Table('cars', metadata, + cars = Table('cars', metadata, Column('car_id', Integer, primary_key=True), Column('owner', Integer, ForeignKey('people.person_id'))) - + def testeagerempty(self): """an easy one...test parent object with child relation to an inheriting mapper, using eager loads, works when there are no child objects present""" @@ -456,8 +463,8 @@ class RelationTest5(ORMTest): sess.save(car2) sess.flush() sess.clear() - - carlist = sess.query(Car).select() + + carlist = sess.query(Car).all() assert carlist[0].manager is None assert carlist[1].manager.person_id == car2.manager.person_id @@ -465,12 +472,12 @@ class RelationTest6(ORMTest): """test self-referential relationships on a single joined-table inheritance mapper""" def define_tables(self, metadata): global people, managers, data - people = Table('people', metadata, + people = Table('people', metadata, Column('person_id', Integer, Sequence('person_id_seq', optional=True), primary_key=True), Column('name', String(50)), ) - managers = Table('managers', metadata, + 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)), @@ -562,7 +569,7 @@ class RelationTest7(ORMTest): employee_join = polymorphic_union( { 'engineer':people.join(engineers), - 'manager':people.join(managers), + 'manager':people.join(managers), }, "type", 'employee_join') car_join = polymorphic_union( @@ -578,7 +585,7 @@ class RelationTest7(ORMTest): offroad_car_mapper = mapper(Offraod_Car, offroad_cars, inherits=car_mapper, polymorphic_identity='offroad') person_mapper = mapper(Person, people, select_table=employee_join,polymorphic_on=employee_join.c.type, - polymorphic_identity='person', + polymorphic_identity='person', properties={ 'car':relation(car_mapper) }) @@ -599,10 +606,10 @@ class RelationTest7(ORMTest): session.flush() session.clear() - r = session.query(Person).select() + r = session.query(Person).all() for p in r: assert p.car_id == p.car.car_id - + class GenerativeTest(AssertMixin): def setUpAll(self): # cars---owned by--- people (abstract) --- has a --- status @@ -615,24 +622,24 @@ class GenerativeTest(AssertMixin): global metadata, status, people, engineers, managers, cars metadata = MetaData(testbase.db) # table definitions - status = Table('status', metadata, + status = Table('status', metadata, Column('status_id', Integer, primary_key=True), Column('name', String(20))) - people = Table('people', metadata, + people = Table('people', metadata, Column('person_id', Integer, primary_key=True), Column('status_id', Integer, ForeignKey('status.status_id'), nullable=False), Column('name', String(50))) - engineers = Table('engineers', metadata, + engineers = Table('engineers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), Column('field', String(30))) - managers = Table('managers', metadata, + managers = Table('managers', metadata, Column('person_id', Integer, ForeignKey('people.person_id'), primary_key=True), Column('category', String(70))) - cars = Table('cars', metadata, + cars = Table('cars', metadata, Column('car_id', Integer, primary_key=True), Column('status_id', Integer, ForeignKey('status.status_id'), nullable=False), Column('owner', Integer, ForeignKey('people.person_id'), nullable=False)) @@ -645,7 +652,7 @@ class GenerativeTest(AssertMixin): clear_mappers() for t in metadata.table_iterator(reverse=True): t.delete().execute() - + def testjointo(self): # class definitions class PersistentObject(object): @@ -668,7 +675,7 @@ class GenerativeTest(AssertMixin): def __repr__(self): return "Car number %d" % self.car_id - # create a union that represents both types of joins. + # create a union that represents both types of joins. employee_join = polymorphic_union( { 'engineer':people.join(engineers), @@ -676,8 +683,8 @@ class GenerativeTest(AssertMixin): }, "type", 'employee_join') status_mapper = mapper(Status, status) - person_mapper = mapper(Person, people, - select_table=employee_join,polymorphic_on=employee_join.c.type, + person_mapper = mapper(Person, people, + select_table=employee_join,polymorphic_on=employee_join.c.type, polymorphic_identity='person', properties={'status':relation(status_mapper)}) engineer_mapper = mapper(Engineer, engineers, inherits=person_mapper, polymorphic_identity='engineer') manager_mapper = mapper(Manager, managers, inherits=person_mapper, polymorphic_identity='manager') @@ -693,7 +700,7 @@ class GenerativeTest(AssertMixin): session.flush() # TODO: we haven't created assertions for all the data combinations created here - + # creating 5 managers named from M1 to M5 and 5 engineers named from E1 to E5 # M4, M5, E4 and E5 are dead for i in range(1,5): @@ -707,7 +714,7 @@ class GenerativeTest(AssertMixin): session.flush() # get E4 - engineer4 = session.query(engineer_mapper).get_by(name="E4") + engineer4 = session.query(engineer_mapper).filter_by(name="E4").one() # create 2 cars for E4, one active and one dead car1 = Car(employee=engineer4,status=active) @@ -722,13 +729,13 @@ class GenerativeTest(AssertMixin): assert str(list(r)) == "[Manager M2, category YYYYYYYYY, status Status active, Engineer E2, field X, status Status active]" r = session.query(Engineer).join('status').filter(people.c.name.in_(['E2', 'E3', 'E4', 'M4', 'M2', 'M1']) & (status.c.name=="active")) assert str(list(r)) == "[Engineer E2, field X, status Status active, Engineer E3, field X, status Status active]" - # this test embeds the original polymorphic union (employee_join) fully - # into the WHERE criterion, using a correlated select. ticket #577 tracks - # that Query's adaptation of the WHERE clause does not dig into the + # this test embeds the original polymorphic union (employee_join) fully + # into the WHERE criterion, using a correlated select. ticket #577 tracks + # that Query's adaptation of the WHERE clause does not dig into the # mapped selectable itself, which permanently breaks the mapped selectable. r = session.query(Person).filter(exists([Car.c.owner], Car.c.owner==employee_join.c.person_id)) assert str(list(r)) == "[Engineer E4, field X, status Status dead]" - + class MultiLevelTest(ORMTest): def define_tables(self, metadata): global table_Employee, table_Engineer, table_Manager @@ -768,7 +775,7 @@ class MultiLevelTest(ORMTest): # 'Engineer': table_Employee.join(table_Engineer).select(table_Employee.c.atype == 'Engineer'), # 'Employee': table_Employee.select( table_Employee.c.atype == 'Employee'), # }, None, 'pu_employee', ) - + mapper_Employee = mapper( Employee, table_Employee, polymorphic_identity= 'Employee', polymorphic_on= pu_Employee.c.atype, @@ -802,9 +809,9 @@ class MultiLevelTest(ORMTest): session.save(b) session.save(c) session.flush() - assert set(session.query(Employee).select()) == set([a,b,c]) - assert set(session.query( Engineer).select()) == set([b,c]) - assert session.query( Manager).select() == [c] + assert set(session.query(Employee).all()) == set([a,b,c]) + assert set(session.query( Engineer).all()) == set([b,c]) + assert session.query( Manager).all() == [c] class ManyToManyPolyTest(ORMTest): def define_tables(self, metadata): @@ -828,9 +835,9 @@ class ManyToManyPolyTest(ORMTest): 'collection', metadata, Column('id', Integer, primary_key=True), Column('name', Unicode(255))) - + def test_pjoin_compile(self): - """test that remote_side columns in the secondary join table arent attempted to be + """test that remote_side columns in the secondary join table arent attempted to be matched to the target polymorphic selectable""" class BaseItem(object): pass class Item(BaseItem): pass @@ -853,13 +860,13 @@ class ManyToManyPolyTest(ORMTest): polymorphic_identity='Item') mapper(Collection, collection_table) - + class_mapper(BaseItem) class CustomPKTest(ORMTest): def define_tables(self, metadata): global t1, t2 - t1 = Table('t1', metadata, + t1 = Table('t1', metadata, Column('id', Integer, primary_key=True), Column('type', String(30), nullable=False), Column('data', String(30))) @@ -870,19 +877,19 @@ class CustomPKTest(ORMTest): def test_custompk(self): """test that the primary_key attribute is propigated to the polymorphic mapper""" - + class T1(object):pass class T2(T1):pass - + # create a polymorphic union with the select against the base table first. - # with the join being second, the alias of the union will + # with the join being second, the alias of the union will # pick up two "primary key" columns. technically the alias should have a # 2-col pk in any case but the leading select has a NULL for the "t2id" column d = util.OrderedDict() d['t1'] = t1.select(t1.c.type=='t1') d['t2'] = t1.join(t2) pjoin = polymorphic_union(d, None, 'pjoin') - + mapper(T1, t1, polymorphic_on=t1.c.type, polymorphic_identity='t1', select_table=pjoin, primary_key=[pjoin.c.id]) mapper(T2, t2, inherits=T1, polymorphic_identity='t2') print [str(c) for c in class_mapper(T1).primary_key] @@ -893,24 +900,24 @@ class CustomPKTest(ORMTest): sess.save(ot2) sess.flush() sess.clear() - + # query using get(), using only one value. this requires the select_table mapper # has the same single-col primary key. assert sess.query(T1).get(ot1.id).id == ot1.id - + ot1 = sess.query(T1).get(ot1.id) ot1.data = 'hi' sess.flush() def test_pk_collapses(self): - """test that a composite primary key attribute formed by a join is "collapsed" into its + """test that a composite primary key attribute formed by a join is "collapsed" into its minimal columns""" class T1(object):pass class T2(T1):pass # create a polymorphic union with the select against the base table first. - # with the join being second, the alias of the union will + # with the join being second, the alias of the union will # pick up two "primary key" columns. technically the alias should have a # 2-col pk in any case but the leading select has a NULL for the "t2id" column d = util.OrderedDict() @@ -922,7 +929,7 @@ class CustomPKTest(ORMTest): mapper(T2, t2, inherits=T1, polymorphic_identity='t2') assert len(class_mapper(T1).primary_key) == 1 assert len(class_mapper(T1).get_select_mapper().compile().primary_key) == 1 - + print [str(c) for c in class_mapper(T1).primary_key] ot1 = T1() ot2 = T2() @@ -943,7 +950,7 @@ class CustomPKTest(ORMTest): class InheritingEagerTest(ORMTest): def define_tables(self, metadata): global people, employees, tags, peopleTags - + people = Table('people', metadata, Column('id', Integer, primary_key=True), Column('_type', String(30), nullable=False), @@ -956,20 +963,20 @@ class InheritingEagerTest(ORMTest): tags = Table('tags', metadata, Column('id', Integer, primary_key=True), - Column('label', String, nullable=False), + Column('label', String(50), nullable=False), ) peopleTags = Table('peopleTags', metadata, Column('person_id', Integer,ForeignKey('people.id')), Column('tag_id', Integer,ForeignKey('tags.id')), ) - + def test_basic(self): """test that Query uses the full set of mapper._eager_loaders when generating SQL""" - + class Person(fixtures.Base): pass - + class Employee(Person): def __init__(self, name='bob'): self.name = name @@ -988,7 +995,7 @@ class InheritingEagerTest(ORMTest): bob = Employee() session.save(bob) - + tag = Tag('crazy') bob.tags.append(tag) @@ -1000,8 +1007,7 @@ class InheritingEagerTest(ORMTest): # query from Employee with limit, query needs to apply eager limiting subquery instance = session.query(Employee).filter_by(id=1).limit(1).first() assert len(instance.tags) == 2 - - -if __name__ == "__main__": + + +if __name__ == "__main__": testbase.main() - |