From: Mike Bayer Date: Sat, 30 Jun 2007 02:04:05 +0000 (+0000) Subject: - finished moving all EagerLoader tests from mapper to eager_relations X-Git-Tag: rel_0_4_6~152 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=f62b68a9f7ec817b80aaea62743c6662224473b8;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git - finished moving all EagerLoader tests from mapper to eager_relations - consolidated other eager tests into assorted_eager - moved assertion objects into fixtures module --- diff --git a/test/orm/alltests.py b/test/orm/alltests.py index 5b3038fd94..e0d8163f25 100644 --- a/test/orm/alltests.py +++ b/test/orm/alltests.py @@ -12,9 +12,7 @@ def suite(): 'orm.mapper', 'orm.generative', 'orm.lazytest1', - 'orm.eagertest1', - 'orm.eagertest2', - 'orm.eagertest3', + 'orm.assorted_eager', 'orm.sessioncontext', 'orm.unitofwork', diff --git a/test/orm/eagertest3.py b/test/orm/assorted_eager.py similarity index 62% rename from test/orm/eagertest3.py rename to test/orm/assorted_eager.py index bfcd166434..77f5526610 100644 --- a/test/orm/eagertest3.py +++ b/test/orm/assorted_eager.py @@ -1,10 +1,12 @@ -from testbase import PersistTest, AssertMixin +"""eager loading unittests derived from mailing list-reported problems and trac tickets.""" + +from testbase import PersistTest, AssertMixin, ORMTest import testbase from sqlalchemy import * from sqlalchemy.orm import * -from sqlalchemy.ext.selectresults import SelectResults +from sqlalchemy.ext.sessioncontext import SessionContext from testbase import Table, Column -import random +import random, datetime class EagerTest(AssertMixin): def setUpAll(self): @@ -121,7 +123,7 @@ class EagerTest(AssertMixin): assert result == [u'1 Some Category', u'3 Some Category'] def test_dslish(self): - """test the same as witheagerload except building the query via SelectResults""" + """test the same as witheagerload except using generative""" s = create_session() q=s.query(Test).options(eagerload('category')) l=q.filter ( @@ -172,6 +174,7 @@ class EagerTest2(AssertMixin): def tearDown(self): for t in metadata.table_iterator(reverse=True): t.delete().execute() + def testeagerterminate(self): """test that eager query generation does not include the same mapper's table twice. @@ -411,6 +414,266 @@ class EagerTest5(testbase.ORMTest): # object is not in the session; therefore the lazy load cant trigger here, # eager load had to succeed assert len([c for c in d2.comments]) == 1 + +class EagerTest6(ORMTest): + def define_tables(self, metadata): + global designType, design, part, inheritedPart + designType = Table('design_types', metadata, + Column('design_type_id', Integer, primary_key=True), + ) + + design =Table('design', metadata, + Column('design_id', Integer, primary_key=True), + Column('design_type_id', Integer, ForeignKey('design_types.design_type_id'))) + + part = Table('parts', metadata, + Column('part_id', Integer, primary_key=True), + Column('design_id', Integer, ForeignKey('design.design_id')), + Column('design_type_id', Integer, ForeignKey('design_types.design_type_id'))) + + inheritedPart = Table('inherited_part', metadata, + Column('ip_id', Integer, primary_key=True), + Column('part_id', Integer, ForeignKey('parts.part_id')), + Column('design_id', Integer, ForeignKey('design.design_id')), + ) + + def testone(self): + class Part(object):pass + class Design(object):pass + class DesignType(object):pass + class InheritedPart(object):pass + + mapper(Part, part) + + mapper(InheritedPart, inheritedPart, properties=dict( + part=relation(Part, lazy=False) + )) + + mapper(Design, design, properties=dict( + parts=relation(Part, private=True, backref="design"), + inheritedParts=relation(InheritedPart, private=True, backref="design"), + )) + + mapper(DesignType, designType, properties=dict( + # designs=relation(Design, private=True, backref="type"), + )) + + class_mapper(Design).add_property("type", relation(DesignType, lazy=False, backref="designs")) + class_mapper(Part).add_property("design", relation(Design, lazy=False, backref="parts")) + #Part.mapper.add_property("designType", relation(DesignType)) + + d = Design() + sess = create_session() + sess.save(d) + sess.flush() + sess.clear() + x = sess.query(Design).get(1) + x.inheritedParts + +class EagerTest7(ORMTest): + def define_tables(self, metadata): + global companies_table, addresses_table, invoice_table, phones_table, items_table, ctx + global Company, Address, Phone, Item,Invoice + + ctx = SessionContext(create_session) + + companies_table = Table('companies', metadata, + Column('company_id', Integer, Sequence('company_id_seq', optional=True), primary_key = True), + Column('company_name', String(40)), + + ) + + addresses_table = Table('addresses', metadata, + Column('address_id', Integer, Sequence('address_id_seq', optional=True), primary_key = True), + Column('company_id', Integer, ForeignKey("companies.company_id")), + Column('address', String(40)), + ) + + phones_table = Table('phone_numbers', metadata, + Column('phone_id', Integer, Sequence('phone_id_seq', optional=True), primary_key = True), + Column('address_id', Integer, ForeignKey('addresses.address_id')), + Column('type', String(20)), + Column('number', String(10)), + ) + + invoice_table = Table('invoices', metadata, + Column('invoice_id', Integer, Sequence('invoice_id_seq', optional=True), primary_key = True), + Column('company_id', Integer, ForeignKey("companies.company_id")), + Column('date', DateTime), + ) + + items_table = Table('items', metadata, + Column('item_id', Integer, Sequence('item_id_seq', optional=True), primary_key = True), + Column('invoice_id', Integer, ForeignKey('invoices.invoice_id')), + Column('code', String(20)), + Column('qty', Integer), + ) + + class Company(object): + def __init__(self): + self.company_id = None + def __repr__(self): + return "Company:" + repr(getattr(self, 'company_id', None)) + " " + repr(getattr(self, 'company_name', None)) + " " + str([repr(addr) for addr in self.addresses]) + + class Address(object): + def __repr__(self): + return "Address: " + repr(getattr(self, 'address_id', None)) + " " + repr(getattr(self, 'company_id', None)) + " " + repr(self.address) + str([repr(ph) for ph in getattr(self, 'phones', [])]) + + class Phone(object): + def __repr__(self): + return "Phone: " + repr(getattr(self, 'phone_id', None)) + " " + repr(getattr(self, 'address_id', None)) + " " + repr(self.type) + " " + repr(self.number) + + class Invoice(object): + def __init__(self): + self.invoice_id = None + def __repr__(self): + return "Invoice:" + repr(getattr(self, 'invoice_id', None)) + " " + repr(getattr(self, 'date', None)) + " " + repr(self.company) + " " + str([repr(item) for item in self.items]) + + class Item(object): + def __repr__(self): + return "Item: " + repr(getattr(self, 'item_id', None)) + " " + repr(getattr(self, 'invoice_id', None)) + " " + repr(self.code) + " " + repr(self.qty) + + def testone(self): + """tests eager load of a many-to-one attached to a one-to-many. this testcase illustrated + the bug, which is that when the single Company is loaded, no further processing of the rows + occurred in order to load the Company's second Address object.""" + + mapper(Address, addresses_table, properties={ + }, extension=ctx.mapper_extension) + mapper(Company, companies_table, properties={ + 'addresses' : relation(Address, lazy=False), + }, extension=ctx.mapper_extension) + mapper(Invoice, invoice_table, properties={ + 'company': relation(Company, lazy=False, ) + }, extension=ctx.mapper_extension) + + c1 = Company() + c1.company_name = 'company 1' + a1 = Address() + a1.address = 'a1 address' + c1.addresses.append(a1) + a2 = Address() + a2.address = 'a2 address' + c1.addresses.append(a2) + i1 = Invoice() + i1.date = datetime.datetime.now() + i1.company = c1 + + ctx.current.flush() + + company_id = c1.company_id + invoice_id = i1.invoice_id + + ctx.current.clear() + + c = ctx.current.query(Company).get(company_id) + + ctx.current.clear() + + i = ctx.current.query(Invoice).get(invoice_id) + + self.echo(repr(c)) + self.echo(repr(i.company)) + self.assert_(repr(c) == repr(i.company)) + + def testtwo(self): + """this is the original testcase that includes various complicating factors""" + + mapper(Phone, phones_table, extension=ctx.mapper_extension) + + mapper(Address, addresses_table, properties={ + 'phones': relation(Phone, lazy=False, backref='address') + }, extension=ctx.mapper_extension) + + mapper(Company, companies_table, properties={ + 'addresses' : relation(Address, lazy=False, backref='company'), + }, extension=ctx.mapper_extension) + + mapper(Item, items_table, extension=ctx.mapper_extension) + + mapper(Invoice, invoice_table, properties={ + 'items': relation(Item, lazy=False, backref='invoice'), + 'company': relation(Company, lazy=False, backref='invoices') + }, extension=ctx.mapper_extension) + + ctx.current.clear() + c1 = Company() + c1.company_name = 'company 1' + + a1 = Address() + a1.address = 'a1 address' + + p1 = Phone() + p1.type = 'home' + p1.number = '1111' + + a1.phones.append(p1) + + p2 = Phone() + p2.type = 'work' + p2.number = '22222' + a1.phones.append(p2) + + c1.addresses.append(a1) + + a2 = Address() + a2.address = 'a2 address' + + p3 = Phone() + p3.type = 'home' + p3.number = '3333' + a2.phones.append(p3) + + p4 = Phone() + p4.type = 'work' + p4.number = '44444' + a2.phones.append(p4) + + c1.addresses.append(a2) + + ctx.current.flush() + + company_id = c1.company_id + + ctx.current.clear() + + a = ctx.current.query(Company).get(company_id) + self.echo(repr(a)) + + # set up an invoice + i1 = Invoice() + i1.date = datetime.datetime.now() + i1.company = c1 + + item1 = Item() + item1.code = 'aaaa' + item1.qty = 1 + item1.invoice = i1 + + item2 = Item() + item2.code = 'bbbb' + item2.qty = 2 + item2.invoice = i1 + + item3 = Item() + item3.code = 'cccc' + item3.qty = 3 + item3.invoice = i1 + + ctx.current.flush() + + invoice_id = i1.invoice_id + + ctx.current.clear() + + c = ctx.current.query(Company).get(company_id) + self.echo(repr(c)) + + ctx.current.clear() + + i = ctx.current.query(Invoice).get(invoice_id) + + assert repr(i.company) == repr(c), repr(i.company) + " does not match " + repr(c) if __name__ == "__main__": diff --git a/test/orm/eager_relations.py b/test/orm/eager_relations.py index d8fdffdbef..e81e96858f 100644 --- a/test/orm/eager_relations.py +++ b/test/orm/eager_relations.py @@ -21,7 +21,7 @@ class EagerTest(QueryTest): q = sess.query(User) assert [User(id=7, addresses=[Address(id=1, email_address='jack@bean.com')])] == q.filter(users.c.id == 7).all() - assert self.user_address_result == q.all() + assert fixtures.user_address_result == q.all() def test_no_orphan(self): """test that an eagerly loaded child object is not marked as an orphan""" @@ -113,11 +113,11 @@ class EagerTest(QueryTest): q = create_session().query(Item) def go(): - assert self.item_keyword_result == q.all() + assert fixtures.item_keyword_result == q.all() self.assert_sql_count(testbase.db, go, 1) def go(): - assert self.item_keyword_result[0:2] == q.join('keywords').filter(keywords.c.name == 'red').all() + assert fixtures.item_keyword_result[0:2] == q.join('keywords').filter(keywords.c.name == 'red').all() self.assert_sql_count(testbase.db, go, 1) @@ -130,7 +130,7 @@ class EagerTest(QueryTest): q = create_session().query(Item) def go(): - assert self.item_keyword_result[0:2] == q.options(eagerload('keywords')).join('keywords').filter(keywords.c.name == 'red').all() + assert fixtures.item_keyword_result[0:2] == q.options(eagerload('keywords')).join('keywords').filter(keywords.c.name == 'red').all() self.assert_sql_count(testbase.db, go, 1) @@ -145,7 +145,7 @@ class EagerTest(QueryTest): assert class_mapper(Address).props['user'].lazy is False sess = create_session() - assert self.user_address_result == sess.query(User).all() + assert fixtures.user_address_result == sess.query(User).all() def test_double(self): """tests lazy loading with two relations simulatneously, from the same table, using aliases. """ @@ -203,12 +203,10 @@ class EagerTest(QueryTest): if testbase.db.engine.name == 'mssql': l = q.limit(2).all() - assert self.user_all_result[:2] == l + assert fixtures.user_all_result[:2] == l else: l = q.limit(2).offset(1).all() - print l - print self.user_all_result[1:3] - assert self.user_all_result[1:3] == l + assert fixtures.user_all_result[1:3] == l def test_distinct(self): # this is an involved 3x union of the users table to get a lot of rows. @@ -226,7 +224,7 @@ class EagerTest(QueryTest): def go(): l = q.filter(s.c.u2_id==User.c.id).distinct().all() - assert self.user_address_result == l + assert fixtures.user_address_result == l self.assert_sql_count(testbase.db, go, 1) def test_limit_2(self): @@ -240,7 +238,7 @@ class EagerTest(QueryTest): l = q.filter((Item.c.description=='item 2') | (Item.c.description=='item 5') | (Item.c.description=='item 3')).\ order_by(Item.c.id).limit(2).all() - assert self.item_keyword_result[1:3] == l + assert fixtures.item_keyword_result[1:3] == l def test_limit_3(self): """test that the ORDER BY is propigated from the inner select to the outer select, when using the @@ -324,7 +322,7 @@ class EagerTest(QueryTest): l = q.filter("users.id in (7, 8, 9)") def go(): - assert self.user_order_result[0:3] == l.all() + assert fixtures.user_order_result[0:3] == l.all() self.assert_sql_count(testbase.db, go, 1) def test_double_with_aggregate(self): @@ -357,5 +355,47 @@ class EagerTest(QueryTest): ] == q.all() self.assert_sql_count(testbase.db, go, 1) + def test_wide(self): + mapper(Order, orders, properties={'items':relation(Item, secondary=order_items, lazy=False)}) + mapper(Item, items) + mapper(User, users, properties = dict( + addresses = relation(mapper(Address, addresses), lazy = False), + orders = relation(Order, lazy = False), + )) + q = create_session().query(User) + l = q.select() + assert fixtures.user_all_result == q.all() + + def test_against_select(self): + """test eager loading of a mapper which is against a select""" + + s = select([orders], orders.c.isopen==1).alias('openorders') + + mapper(Order, s, properties={ + 'user':relation(User, lazy=False) + }) + mapper(User, users) + + q = create_session().query(Order) + assert [ + Order(id=3, user=User(id=7)), + Order(id=4, user=User(id=9)) + ] == q.all() + + q = q.select_from(s.join(order_items).join(items)).filter(~items.c.id.in_(1, 2, 5)) + assert [ + Order(id=3, user=User(id=7)), + ] == q.all() + + def test_aliasing(self): + """test that eager loading uses aliases to insulate the eager load from regular criterion against those tables.""" + + mapper(User, users, properties = dict( + addresses = relation(mapper(Address, addresses), lazy=False) + )) + q = create_session().query(User) + l = q.filter(addresses.c.email_address == 'ed@lala.com').filter(addresses.c.user_id==users.c.id) + assert fixtures.user_address_result[1:2] == l.all() + if __name__ == '__main__': testbase.main() diff --git a/test/orm/eagertest1.py b/test/orm/eagertest1.py deleted file mode 100644 index dd3ad92e57..0000000000 --- a/test/orm/eagertest1.py +++ /dev/null @@ -1,71 +0,0 @@ -from testbase import PersistTest, AssertMixin -import testbase -import unittest, sys, os -from sqlalchemy import * -from sqlalchemy.orm import * -from testbase import Table, Column -import datetime - -class EagerTest(AssertMixin): - def setUpAll(self): - global designType, design, part, inheritedPart - designType = Table('design_types', testbase.metadata, - Column('design_type_id', Integer, primary_key=True), - ) - - design =Table('design', testbase.metadata, - Column('design_id', Integer, primary_key=True), - Column('design_type_id', Integer, ForeignKey('design_types.design_type_id'))) - - part = Table('parts', testbase.metadata, - Column('part_id', Integer, primary_key=True), - Column('design_id', Integer, ForeignKey('design.design_id')), - Column('design_type_id', Integer, ForeignKey('design_types.design_type_id'))) - - inheritedPart = Table('inherited_part', testbase.metadata, - Column('ip_id', Integer, primary_key=True), - Column('part_id', Integer, ForeignKey('parts.part_id')), - Column('design_id', Integer, ForeignKey('design.design_id')), - ) - - testbase.metadata.create_all() - def tearDownAll(self): - testbase.metadata.drop_all() - testbase.metadata.clear() - def testone(self): - class Part(object):pass - class Design(object):pass - class DesignType(object):pass - class InheritedPart(object):pass - - mapper(Part, part) - - mapper(InheritedPart, inheritedPart, properties=dict( - part=relation(Part, lazy=False) - )) - - mapper(Design, design, properties=dict( - parts=relation(Part, private=True, backref="design"), - inheritedParts=relation(InheritedPart, private=True, backref="design"), - )) - - mapper(DesignType, designType, properties=dict( - # designs=relation(Design, private=True, backref="type"), - )) - - class_mapper(Design).add_property("type", relation(DesignType, lazy=False, backref="designs")) - class_mapper(Part).add_property("design", relation(Design, lazy=False, backref="parts")) - #Part.mapper.add_property("designType", relation(DesignType)) - - d = Design() - sess = create_session() - sess.save(d) - sess.flush() - sess.clear() - x = sess.query(Design).get(1) - x.inheritedParts - -if __name__ == "__main__": - testbase.main() - - diff --git a/test/orm/eagertest2.py b/test/orm/eagertest2.py deleted file mode 100644 index 0b51ab9dbf..0000000000 --- a/test/orm/eagertest2.py +++ /dev/null @@ -1,240 +0,0 @@ -from testbase import PersistTest, AssertMixin -import testbase -import unittest, sys, os -from sqlalchemy import * -from sqlalchemy.orm import * -from sqlalchemy.ext.sessioncontext import SessionContext -from testbase import Table, Column -import datetime - -class EagerTest(AssertMixin): - def setUpAll(self): - global companies_table, addresses_table, invoice_table, phones_table, items_table, ctx, metadata - - metadata = BoundMetaData(testbase.db) - ctx = SessionContext(create_session) - - companies_table = Table('companies', metadata, - Column('company_id', Integer, Sequence('company_id_seq', optional=True), primary_key = True), - Column('company_name', String(40)), - - ) - - addresses_table = Table('addresses', metadata, - Column('address_id', Integer, Sequence('address_id_seq', optional=True), primary_key = True), - Column('company_id', Integer, ForeignKey("companies.company_id")), - Column('address', String(40)), - ) - - phones_table = Table('phone_numbers', metadata, - Column('phone_id', Integer, Sequence('phone_id_seq', optional=True), primary_key = True), - Column('address_id', Integer, ForeignKey('addresses.address_id')), - Column('type', String(20)), - Column('number', String(10)), - ) - - invoice_table = Table('invoices', metadata, - Column('invoice_id', Integer, Sequence('invoice_id_seq', optional=True), primary_key = True), - Column('company_id', Integer, ForeignKey("companies.company_id")), - Column('date', DateTime), - ) - - items_table = Table('items', metadata, - Column('item_id', Integer, Sequence('item_id_seq', optional=True), primary_key = True), - Column('invoice_id', Integer, ForeignKey('invoices.invoice_id')), - Column('code', String(20)), - Column('qty', Integer), - ) - - metadata.create_all() - - def tearDownAll(self): - metadata.drop_all() - - def tearDown(self): - clear_mappers() - for t in metadata.table_iterator(reverse=True): - t.delete().execute() - - def testone(self): - """tests eager load of a many-to-one attached to a one-to-many. this testcase illustrated - the bug, which is that when the single Company is loaded, no further processing of the rows - occurred in order to load the Company's second Address object.""" - class Company(object): - def __init__(self): - self.company_id = None - def __repr__(self): - return "Company:" + repr(getattr(self, 'company_id', None)) + " " + repr(getattr(self, 'company_name', None)) + " " + str([repr(addr) for addr in self.addresses]) - - class Address(object): - def __repr__(self): - return "Address: " + repr(getattr(self, 'address_id', None)) + " " + repr(getattr(self, 'company_id', None)) + " " + repr(self.address) - - class Invoice(object): - def __init__(self): - self.invoice_id = None - def __repr__(self): - return "Invoice:" + repr(getattr(self, 'invoice_id', None)) + " " + repr(getattr(self, 'date', None)) + " " + repr(self.company) - - mapper(Address, addresses_table, properties={ - }, extension=ctx.mapper_extension) - mapper(Company, companies_table, properties={ - 'addresses' : relation(Address, lazy=False), - }, extension=ctx.mapper_extension) - mapper(Invoice, invoice_table, properties={ - 'company': relation(Company, lazy=False, ) - }, extension=ctx.mapper_extension) - - c1 = Company() - c1.company_name = 'company 1' - a1 = Address() - a1.address = 'a1 address' - c1.addresses.append(a1) - a2 = Address() - a2.address = 'a2 address' - c1.addresses.append(a2) - i1 = Invoice() - i1.date = datetime.datetime.now() - i1.company = c1 - - ctx.current.flush() - - company_id = c1.company_id - invoice_id = i1.invoice_id - - ctx.current.clear() - - c = ctx.current.query(Company).get(company_id) - - ctx.current.clear() - - i = ctx.current.query(Invoice).get(invoice_id) - - self.echo(repr(c)) - self.echo(repr(i.company)) - self.assert_(repr(c) == repr(i.company)) - - def testtwo(self): - """this is the original testcase that includes various complicating factors""" - class Company(object): - def __init__(self): - self.company_id = None - def __repr__(self): - return "Company:" + repr(getattr(self, 'company_id', None)) + " " + repr(getattr(self, 'company_name', None)) + " " + str([repr(addr) for addr in self.addresses]) - - class Address(object): - def __repr__(self): - return "Address: " + repr(getattr(self, 'address_id', None)) + " " + repr(getattr(self, 'company_id', None)) + " " + repr(self.address) + str([repr(ph) for ph in self.phones]) - - class Phone(object): - def __repr__(self): - return "Phone: " + repr(getattr(self, 'phone_id', None)) + " " + repr(getattr(self, 'address_id', None)) + " " + repr(self.type) + " " + repr(self.number) - - class Invoice(object): - def __init__(self): - self.invoice_id = None - def __repr__(self): - return "Invoice:" + repr(getattr(self, 'invoice_id', None)) + " " + repr(getattr(self, 'date', None)) + " " + repr(self.company) + " " + str([repr(item) for item in self.items]) - - class Item(object): - def __repr__(self): - return "Item: " + repr(getattr(self, 'item_id', None)) + " " + repr(getattr(self, 'invoice_id', None)) + " " + repr(self.code) + " " + repr(self.qty) - - mapper(Phone, phones_table, extension=ctx.mapper_extension) - - mapper(Address, addresses_table, properties={ - 'phones': relation(Phone, lazy=False, backref='address') - }, extension=ctx.mapper_extension) - - mapper(Company, companies_table, properties={ - 'addresses' : relation(Address, lazy=False, backref='company'), - }, extension=ctx.mapper_extension) - - mapper(Item, items_table, extension=ctx.mapper_extension) - - mapper(Invoice, invoice_table, properties={ - 'items': relation(Item, lazy=False, backref='invoice'), - 'company': relation(Company, lazy=False, backref='invoices') - }, extension=ctx.mapper_extension) - - ctx.current.clear() - c1 = Company() - c1.company_name = 'company 1' - - a1 = Address() - a1.address = 'a1 address' - - p1 = Phone() - p1.type = 'home' - p1.number = '1111' - - a1.phones.append(p1) - - p2 = Phone() - p2.type = 'work' - p2.number = '22222' - a1.phones.append(p2) - - c1.addresses.append(a1) - - a2 = Address() - a2.address = 'a2 address' - - p3 = Phone() - p3.type = 'home' - p3.number = '3333' - a2.phones.append(p3) - - p4 = Phone() - p4.type = 'work' - p4.number = '44444' - a2.phones.append(p4) - - c1.addresses.append(a2) - - ctx.current.flush() - - company_id = c1.company_id - - ctx.current.clear() - - a = ctx.current.query(Company).get(company_id) - self.echo(repr(a)) - - # set up an invoice - i1 = Invoice() - i1.date = datetime.datetime.now() - i1.company = c1 - - item1 = Item() - item1.code = 'aaaa' - item1.qty = 1 - item1.invoice = i1 - - item2 = Item() - item2.code = 'bbbb' - item2.qty = 2 - item2.invoice = i1 - - item3 = Item() - item3.code = 'cccc' - item3.qty = 3 - item3.invoice = i1 - - ctx.current.flush() - - invoice_id = i1.invoice_id - - ctx.current.clear() - - c = ctx.current.query(Company).get(company_id) - self.echo(repr(c)) - - ctx.current.clear() - - i = ctx.current.query(Invoice).get(invoice_id) - - assert repr(i.company) == repr(c), repr(i.company) + " does not match " + repr(c) - -if __name__ == "__main__": - testbase.main() diff --git a/test/orm/fixtures.py b/test/orm/fixtures.py index 49652243a6..ddec0a2812 100644 --- a/test/orm/fixtures.py +++ b/test/orm/fixtures.py @@ -159,3 +159,72 @@ def install_fixture_data(): dict(keyword_id=7, item_id=2), dict(keyword_id=6, item_id=3) ) + +class Fixtures(object): + @property + def user_address_result(self): + return [ + User(id=7, addresses=[ + Address(id=1) + ]), + User(id=8, addresses=[ + Address(id=2, email_address='ed@wood.com'), + Address(id=3, email_address='ed@bettyboop.com'), + Address(id=4, email_address='ed@lala.com'), + ]), + User(id=9, addresses=[ + Address(id=5) + ]), + User(id=10, addresses=[]) + ] + + @property + def user_all_result(self): + return [ + User(id=7, addresses=[ + Address(id=1) + ], orders=[ + Order(description='order 1', items=[Item(description='item 1'), Item(description='item 2'), Item(description='item 3')]), + Order(description='order 3'), + Order(description='order 5'), + ]), + User(id=8, addresses=[ + Address(id=2), + Address(id=3), + Address(id=4) + ]), + User(id=9, addresses=[ + Address(id=5) + ], orders=[ + Order(description='order 2', items=[Item(description='item 1'), Item(description='item 2'), Item(description='item 3')]), + Order(description='order 4', items=[Item(description='item 1'), Item(description='item 5')]), + ]), + User(id=10, addresses=[]) + ] + + @property + def user_order_result(self): + return [ + User(id=7, orders=[ + Order(id=1, items=[Item(id=1), Item(id=2), Item(id=3)]), + Order(id=3, items=[Item(id=3), Item(id=4), Item(id=5)]), + Order(id=5, items=[Item(id=5)]), + ]), + User(id=8, orders=[]), + User(id=9, orders=[ + Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)]), + Order(id=4, items=[Item(id=1), Item(id=5)]), + ]), + User(id=10) + ] + + @property + def item_keyword_result(self): + return [ + Item(id=1, keywords=[Keyword(name='red'), Keyword(name='big'), Keyword(name='round')]), + Item(id=2, keywords=[Keyword(name='red'), Keyword(name='small'), Keyword(name='square')]), + Item(id=3, keywords=[Keyword(name='green'), Keyword(name='big'), Keyword(name='round')]), + Item(id=4, keywords=[]), + Item(id=5, keywords=[]), + ] +fixtures = Fixtures() diff --git a/test/orm/lazy_relations.py b/test/orm/lazy_relations.py index 81ec00ac67..5accb74713 100644 --- a/test/orm/lazy_relations.py +++ b/test/orm/lazy_relations.py @@ -145,12 +145,10 @@ class LazyTest(QueryTest): if testbase.db.engine.name == 'mssql': l = q.limit(2).all() - assert self.user_all_result[:2] == l + assert fixtures.user_all_result[:2] == l else: l = q.limit(2).offset(1).all() - print l - print self.user_all_result[1:3] - assert self.user_all_result[1:3] == l + assert fixtures.user_all_result[1:3] == l def test_distinct(self): mapper(Item, items) @@ -170,7 +168,7 @@ class LazyTest(QueryTest): s = union_all(u2.select(use_labels=True), u2.select(use_labels=True), u2.select(use_labels=True)).alias('u') print [key for key in s.c.keys()] l = q.filter(s.c.u2_id==User.c.id).distinct().all() - assert self.user_all_result == l + assert fixtures.user_all_result == l def test_one_to_many_scalar(self): mapper(User, users, properties = dict( @@ -225,9 +223,9 @@ class LazyTest(QueryTest): )) q = create_session().query(Item) - assert self.item_keyword_result == q.all() + assert fixtures.item_keyword_result == q.all() - assert self.item_keyword_result[0:2] == q.join('keywords').filter(keywords.c.name == 'red').all() + assert fixtures.item_keyword_result[0:2] == q.join('keywords').filter(keywords.c.name == 'red').all() def test_uses_get(self): """test that a simple many-to-one lazyload optimizes to use query.get().""" diff --git a/test/orm/mapper.py b/test/orm/mapper.py index 0dd40f6657..19a2e0a3c0 100644 --- a/test/orm/mapper.py +++ b/test/orm/mapper.py @@ -1,3 +1,5 @@ +"""tests general mapper operations with an emphasis on selecting/loading""" + from testbase import PersistTest, AssertMixin import testbase import unittest, sys, os @@ -8,8 +10,6 @@ from sqlalchemy.ext.sessioncontext import SessionContext, SessionContextExt from tables import * import tables -"""tests general mapper operations with an emphasis on selecting/loading""" - class MapperSuperTest(AssertMixin): def setUpAll(self): tables.create() @@ -853,7 +853,6 @@ class NoLoadTest(MapperSuperTest): {'user_id' : 7, 'addresses' : (Address, [{'address_id' : 1}])}, ) - class MapperExtensionTest(MapperSuperTest): def testcreateinstance(self): class Ext(MapperExtension): @@ -868,82 +867,6 @@ class MapperExtensionTest(MapperSuperTest): l = q.select(); self.assert_result(l, User, *user_address_result) -class EagerTest(MapperSuperTest): - - def testwithrepeat(self): - """tests a one-to-many eager load where we also query on joined criterion, where the joined - criterion is using the same tables that are used within the eager load. the mapper must insure that the - criterion doesnt interfere with the eager load criterion.""" - m = mapper(User, users, properties = dict( - addresses = relation(mapper(Address, addresses), primaryjoin = users.c.user_id==addresses.c.user_id, lazy = False) - )) - q = create_session().query(m) - l = q.select(and_(addresses.c.email_address == 'ed@lala.com', addresses.c.user_id==users.c.user_id)) - self.assert_result(l, User, - {'user_id' : 8, 'addresses' : (Address, [{'address_id' : 2, 'email_address':'ed@wood.com'}, {'address_id':3, 'email_address':'ed@bettyboop.com'}, {'address_id':4, 'email_address':'ed@lala.com'}])}, - ) - - - def testonselect(self): - """test eager loading of a mapper which is against a select""" - - s = select([orders], orders.c.isopen==1).alias('openorders') - print "SELECT:", id(s), str(s) - mapper(Order, s, properties={ - 'user':relation(User, lazy=False) - }) - mapper(User, users) - - q = create_session().query(Order) - self.assert_result(q.list(), Order, - {'order_id':3, 'user' : (User, {'user_id':7})}, - {'order_id':4, 'user' : (User, {'user_id':9})}, - ) - - q = q.select_from(s.outerjoin(orderitems)).filter(orderitems.c.item_name != 'item 2') - self.assert_result(q.list(), Order, - {'order_id':3, 'user' : (User, {'user_id':7})}, - ) - - - def testmulti(self): - """tests eager loading with two relations simultaneously""" - m = mapper(User, users, properties = dict( - addresses = relation(mapper(Address, addresses), primaryjoin = users.c.user_id==addresses.c.user_id, lazy = False), - orders = relation(mapper(Order, orders), lazy = False), - )) - q = create_session().query(m) - l = q.select() - self.assert_result(l, User, - {'user_id' : 7, - 'addresses' : (Address, [{'address_id' : 1}]), - 'orders' : (Order, [{'order_id' : 1}, {'order_id' : 3},{'order_id' : 5},]) - }, - {'user_id' : 8, - 'addresses' : (Address, [{'address_id' : 2}, {'address_id' : 3}, {'address_id' : 4}]), - 'orders' : (Order, []) - }, - {'user_id' : 9, - 'addresses' : (Address, []), - 'orders' : (Order, [{'order_id' : 2},{'order_id' : 4}]) - } - ) - - def testnested(self): - """tests eager loading of a parent item with two types of child items, - where one of those child items eager loads its own child items.""" - ordermapper = mapper(Order, orders, properties = dict( - items = relation(mapper(Item, orderitems), lazy = False) - )) - - m = mapper(User, users, properties = dict( - addresses = relation(mapper(Address, addresses), lazy = False), - orders = relation(ordermapper, primaryjoin = users.c.user_id==orders.c.user_id, lazy = False), - )) - q = create_session().query(m) - l = q.select() - self.assert_result(l, User, *user_all_result) - if __name__ == "__main__": diff --git a/test/orm/query.py b/test/orm/query.py index c500167417..773344dae1 100644 --- a/test/orm/query.py +++ b/test/orm/query.py @@ -71,72 +71,6 @@ class QueryTest(testbase.ORMTest): }) mapper(Keyword, keywords) - @property - def user_address_result(self): - return [ - User(id=7, addresses=[ - Address(id=1) - ]), - User(id=8, addresses=[ - Address(id=2, email_address='ed@wood.com'), - Address(id=3, email_address='ed@bettyboop.com'), - Address(id=4, email_address='ed@lala.com'), - ]), - User(id=9, addresses=[ - Address(id=5) - ]), - User(id=10, addresses=[]) - ] - - @property - def user_all_result(self): - return [ - User(id=7, addresses=[ - Address(id=1) - ], orders=[ - Order(description='order 1', items=[Item(description='item 1'), Item(description='item 2'), Item(description='item 3')]), - Order(description='order 3'), - Order(description='order 5'), - ]), - User(id=8, addresses=[ - Address(id=2), - Address(id=3), - Address(id=4) - ]), - User(id=9, addresses=[ - Address(id=5) - ], orders=[ - Order(description='order 2', items=[Item(description='item 1'), Item(description='item 2'), Item(description='item 3')]), - Order(description='order 4', items=[Item(description='item 1'), Item(description='item 5')]), - ]), - User(id=10, addresses=[]) - ] - - @property - def user_order_result(self): - return [ - User(id=7, orders=[ - Order(id=1, items=[Item(id=1), Item(id=2), Item(id=3)]), - Order(id=3, items=[Item(id=3), Item(id=4), Item(id=5)]), - Order(id=5, items=[Item(id=5)]), - ]), - User(id=8, orders=[]), - User(id=9, orders=[ - Order(id=2, items=[Item(id=1), Item(id=2), Item(id=3)]), - Order(id=4, items=[Item(id=1), Item(id=5)]), - ]), - User(id=10) - ] - - @property - def item_keyword_result(self): - return [ - Item(id=1, keywords=[Keyword(name='red'), Keyword(name='big'), Keyword(name='round')]), - Item(id=2, keywords=[Keyword(name='red'), Keyword(name='small'), Keyword(name='square')]), - Item(id=3, keywords=[Keyword(name='green'), Keyword(name='big'), Keyword(name='round')]), - Item(id=4, keywords=[]), - Item(id=5, keywords=[]), - ] class GetTest(QueryTest): def test_get(self): @@ -261,7 +195,7 @@ class InstancesTest(QueryTest): def go(): l = q.options(contains_alias('ulist'), contains_eager('addresses')).instances(query.execute()) - assert self.user_address_result == l + assert fixtures.user_address_result == l self.assert_sql_count(testbase.db, go, 1) def test_contains_eager(self): @@ -271,7 +205,7 @@ class InstancesTest(QueryTest): def go(): l = q.options(contains_eager('addresses')).instances(selectquery.execute()) - assert self.user_address_result == l + assert fixtures.user_address_result == l self.assert_sql_count(testbase.db, go, 1) def test_contains_eager_alias(self): @@ -282,13 +216,13 @@ class InstancesTest(QueryTest): def go(): # test using a string alias name l = q.options(contains_eager('addresses', alias="adalias")).instances(selectquery.execute()) - assert self.user_address_result == l + assert fixtures.user_address_result == l self.assert_sql_count(testbase.db, go, 1) def go(): # test using the Alias object itself l = q.options(contains_eager('addresses', alias=adalias)).instances(selectquery.execute()) - assert self.user_address_result == l + assert fixtures.user_address_result == l self.assert_sql_count(testbase.db, go, 1) def decorate(row): @@ -300,7 +234,7 @@ class InstancesTest(QueryTest): def go(): # test using a custom 'decorate' function l = q.options(contains_eager('addresses', decorator=decorate)).instances(selectquery.execute()) - assert self.user_address_result == l + assert fixtures.user_address_result == l self.assert_sql_count(testbase.db, go, 1) def test_multi_mappers(self):