diff options
Diffstat (limited to 'test/orm/unitofwork.py')
| -rw-r--r-- | test/orm/unitofwork.py | 353 |
1 files changed, 178 insertions, 175 deletions
diff --git a/test/orm/unitofwork.py b/test/orm/unitofwork.py index 443c5638a..87a565f3e 100644 --- a/test/orm/unitofwork.py +++ b/test/orm/unitofwork.py @@ -1,4 +1,7 @@ # coding: utf-8 + +"""Tests unitofwork operations.""" + import testbase import pickleable from sqlalchemy import * @@ -6,9 +9,8 @@ from sqlalchemy import exceptions, sql from sqlalchemy.orm import * from testlib import * from testlib.tables import * -from testlib import tables, fixtures +from testlib import engines, tables, fixtures -"""tests unitofwork operations""" # TODO: convert suite to not use Session.mapper, use fixtures.Base # with explicit session.save() @@ -17,12 +19,12 @@ mapper = Session.mapper class UnitOfWorkTest(object): pass - + class HistoryTest(ORMTest): metadata = tables.metadata def define_tables(self, metadata): pass - + def test_backref(self): s = Session() class User(object):pass @@ -31,7 +33,7 @@ class HistoryTest(ORMTest): m = mapper(User, users, properties = dict( addresses = relation(am, backref='user', lazy=False)) ) - + u = User(_sa_session=s) a = Address(_sa_session=s) a.user = u @@ -42,7 +44,7 @@ class HistoryTest(ORMTest): s.close() u = s.query(m).select()[0] print u.addresses[0].user - + class VersioningTest(ORMTest): def define_tables(self, metadata): global version_table @@ -61,7 +63,7 @@ class VersioningTest(ORMTest): f1 = Foo(value='f1', _sa_session=s) f2 = Foo(value='f2', _sa_session=s) s.commit() - + f1.value='f1rev2' s.commit() s2 = Session() @@ -82,11 +84,11 @@ class VersioningTest(ORMTest): # Only dialects with a sane rowcount can detect the ConcurrentModificationError if testbase.db.dialect.supports_sane_rowcount: assert success - + s.close() f1 = s.query(Foo).get(f1.id) f2 = s.query(Foo).get(f2.id) - + f1_s.value='f1rev4' s2.commit() @@ -100,7 +102,7 @@ class VersioningTest(ORMTest): success = True if testbase.db.dialect.supports_sane_multi_rowcount: assert success - + @engines.close_open_connections def test_versioncheck(self): """test that query.with_lockmode performs a 'version check' on an already loaded instance""" @@ -123,11 +125,11 @@ class VersioningTest(ORMTest): s1.query(Foo).load(f1s1.id) # now assert version OK s1.query(Foo).with_lockmode('read').get(f1s1.id) - + # assert brand new load is OK too s1.close() s1.query(Foo).with_lockmode('read').get(f1s1.id) - + @engines.close_open_connections def test_noversioncheck(self): """test that query.with_lockmode works OK when the mapper has no version id col""" @@ -141,7 +143,7 @@ class VersioningTest(ORMTest): f1s2 = s2.query(Foo).with_lockmode('read').get(f1s1.id) assert f1s2.id == f1s1.id assert f1s2.value == f1s1.value - + class UnicodeTest(ORMTest): def define_tables(self, metadata): global uni_table, uni_table2 @@ -164,18 +166,18 @@ class UnicodeTest(ORMTest): self.assert_(t1.txt == txt) Session.commit() self.assert_(t1.txt == txt) - + def test_relation(self): class Test(object): def __init__(self, txt): self.txt = txt class Test2(object):pass - + mapper(Test, uni_table, properties={ 't2s':relation(Test2) }) mapper(Test2, uni_table2) - + txt = u"\u0160\u0110\u0106\u010c\u017d" t1 = Test(txt=txt) t1.t2s.append(Test2()) @@ -186,11 +188,14 @@ class UnicodeTest(ORMTest): assert len(t1.t2s) == 2 class UnicodeSchemaTest(ORMTest): - @testing.supported('sqlite', 'postgres') - def define_tables(self, metadata): - global t1, t2, t3 + __unsupported_on__ = ('oracle', 'mssql', 'firebird', 'sybase', + 'access', 'maxdb') + __excluded_on__ = (('mysql', '<', (4, 1, 1)),) + + metadata = MetaData(engines.utf8_engine()) - #unicode_bind = utf8_engine() + def define_tables(self, metadata): + global t1, t2 t1 = Table('unitable1', metadata, Column(u'méil', Integer, primary_key=True, key='a'), @@ -204,8 +209,7 @@ class UnicodeSchemaTest(ORMTest): Column(u'\u6e2c\u8a66_2', Integer, key="e"), test_needs_fk=True, ) - - @testing.supported('sqlite', 'postgres') + def test_mapping(self): class A(fixtures.Base):pass class B(fixtures.Base):pass @@ -234,7 +238,6 @@ class UnicodeSchemaTest(ORMTest): assert new_a1.t2s[0].d == b1.d Session.clear() - @testing.supported('sqlite', 'postgres') def test_inheritance_mapping(self): class A(fixtures.Base):pass class B(A):pass @@ -249,7 +252,7 @@ class UnicodeSchemaTest(ORMTest): # breaks the comparison ????? l = Session.query(A).all() assert [A(b=5), B(e=7)] == l - + class MutableTypesTest(ORMTest): def define_tables(self, metadata): global table @@ -302,7 +305,7 @@ class MutableTypesTest(ORMTest): {'mutabletest_id': f1.id, 'val': u'hi', 'data':f1.data} ), ]) - + def test_nocomparison(self): """test that types marked as MutableType get changes detected on them when the type has no __eq__ method""" class Foo(object):pass @@ -310,11 +313,11 @@ class MutableTypesTest(ORMTest): f1 = Foo() f1.data = pickleable.BarWithoutCompare(4,5) Session.commit() - + def go(): Session.commit() self.assert_sql_count(testbase.db, go, 0) - + Session.close() f2 = Session.query(Foo).get_by(id=f1.id) @@ -327,7 +330,7 @@ class MutableTypesTest(ORMTest): def go(): Session.commit() self.assert_sql_count(testbase.db, go, 1) - + Session.close() f3 = Session.query(Foo).get_by(id=f1.id) print f2.data, f3.data @@ -336,10 +339,10 @@ class MutableTypesTest(ORMTest): def go(): Session.commit() self.assert_sql_count(testbase.db, go, 0) - + def test_unicode(self): """test that two equivalent unicode values dont get flagged as changed. - + apparently two equal unicode objects dont compare via "is" in all cases, so this tests the compare_values() call on types.String and its usage via types.Unicode.""" class Foo(object):pass @@ -362,7 +365,7 @@ class MutableTypesTest2(ORMTest): Column('id', Integer, Sequence('mutableidseq', optional=True), primary_key=True), Column('data', PickleType(comparator=operator.eq)), ) - + def test_dicts(self): """dictionaries dont pickle the same way twice, sigh.""" @@ -390,25 +393,25 @@ class MutableTypesTest2(ORMTest): def go(): Session.commit() self.assert_sql_count(testbase.db, go, 1) - + Session.clear() f = Session.query(Foo).get(f1.id) assert f.data == [{'personne': {'nom': u'Smith', 'pers_id': 1, 'prenom': u'john', 'civilite': u'Mr', \ 'int_3': False, 'int_2': False, 'int_1': u'23', 'VenSoir': False, 'str_1': u'Test', \ 'SamMidi': False, 'str_2': u'chien', 'DimMidi': False, 'SamSoir': True, 'SamAcc': False}}] - + class PKTest(ORMTest): def define_tables(self, metadata): global table, table2, table3 table = Table( - 'multipk', metadata, + 'multipk', metadata, Column('multi_id', Integer, Sequence("multi_id_seq", optional=True), primary_key=True), Column('multi_rev', Integer, primary_key=True), Column('name', String(50), nullable=False), Column('value', String(100)) ) - + table2 = Table('multipk2', metadata, Column('pk_col_1', String(30), primary_key=True), Column('pk_col_2', String(30), primary_key=True), @@ -422,8 +425,8 @@ class PKTest(ORMTest): ) # not supported on sqlite since sqlite's auto-pk generation only works with - # single column primary keys - @testing.unsupported('sqlite') + # single column primary keys + @testing.fails_on('sqlite') def test_primarykey(self): class Entry(object): pass @@ -436,7 +439,7 @@ class PKTest(ORMTest): Session.close() e2 = Query(Entry).get((e.multi_id, 2)) self.assert_(e is not e2 and e._instance_key == e2._instance_key) - + # this one works with sqlite since we are manually setting up pk values def test_manualpk(self): class Entry(object): @@ -447,7 +450,7 @@ class PKTest(ORMTest): e.pk_col_2 = 'pk1_related' e.data = 'im the data' Session.commit() - + def test_keypks(self): import datetime class Entity(object): @@ -463,7 +466,7 @@ class PKTest(ORMTest): class ForeignPKTest(ORMTest): """tests mapper detection of the relationship direction when parent/child tables are joined on their primary keys""" - + def define_tables(self, metadata): global people, peoplesites @@ -472,13 +475,13 @@ class ForeignPKTest(ORMTest): Column('firstname', String(10)), Column('lastname', String(10)), ) - + peoplesites = Table("peoplesites", metadata, - Column('person', String(10), ForeignKey("people.person"), + Column('person', String(10), ForeignKey("people.person"), primary_key=True), Column('site', String(10)), ) - + def test_basic(self): class PersonSite(object):pass class Person(object):pass @@ -486,7 +489,7 @@ class ForeignPKTest(ORMTest): m2 = mapper(Person, people, properties = { - 'sites' : relation(PersonSite), + 'sites' : relation(PersonSite), }, ) compile_mappers() @@ -539,12 +542,12 @@ class ClauseAttributesTest(ORMTest): assert u.name == 'test2' assert u.counter == 2 self.assert_sql_count(testbase.db, go, 1) - + sess.clear() u = sess.query(User).get(u.id) assert u.name == 'test2' assert u.counter == 2 - + @testing.unsupported('mssql') def test_insert(self): class User(object): @@ -581,7 +584,7 @@ class PassiveDeletesTest(ORMTest): pass class MyOtherClass(object): pass - + mapper(MyOtherClass, myothertable) mapper(MyClass, mytable, properties={ @@ -621,15 +624,15 @@ class ExtraPassiveDeletesTest(ORMTest): ForeignKeyConstraint(['parent_id'],['mytable.id']), # no CASCADE, the same as ON DELETE RESTRICT test_needs_fk=True, ) - + def test_assertions(self): class MyClass(object): pass class MyOtherClass(object): pass - + mapper(MyOtherClass, myothertable) - + try: mapper(MyClass, mytable, properties={ 'children':relation(MyOtherClass, passive_deletes='all', cascade="all") @@ -637,14 +640,14 @@ class ExtraPassiveDeletesTest(ORMTest): assert False except exceptions.ArgumentError, e: assert str(e) == "Can't set passive_deletes='all' in conjunction with 'delete' or 'delete-orphan' cascade" - + @testing.unsupported('sqlite') def test_extra_passive(self): class MyClass(object): pass class MyOtherClass(object): pass - + mapper(MyOtherClass, myothertable) mapper(MyClass, mytable, properties={ @@ -669,12 +672,12 @@ class ExtraPassiveDeletesTest(ORMTest): except exceptions.DBAPIError: assert True - + class DefaultTest(ORMTest): """tests that when saving objects whose table contains DefaultGenerators, either python-side, preexec or database-side, - the newly saved instances receive all the default values either through a post-fetch or getting the pre-exec'ed + the newly saved instances receive all the default values either through a post-fetch or getting the pre-exec'ed defaults back from the engine.""" - + def define_tables(self, metadata): db = testbase.db use_string_defaults = db.engine.__module__.endswith('postgres') or db.engine.__module__.endswith('oracle') or db.engine.__module__.endswith('sqlite') @@ -687,7 +690,7 @@ class DefaultTest(ORMTest): hohotype = Integer self.hohoval = 9 self.althohoval = 15 - + global default_table default_table = Table('default_test', metadata, Column('id', Integer, Sequence("dt_seq", optional=True), primary_key=True), @@ -696,42 +699,42 @@ class DefaultTest(ORMTest): Column('foober', String(30), default="im foober", onupdate="im the update") ) - + def test_insert(self): class Hoho(object):pass mapper(Hoho, default_table) - + h1 = Hoho(hoho=self.althohoval) h2 = Hoho(counter=12) h3 = Hoho(hoho=self.althohoval, counter=12) h4 = Hoho() h5 = Hoho(foober='im the new foober') Session.commit() - + self.assert_(h1.hoho==self.althohoval) self.assert_(h3.hoho==self.althohoval) - + def go(): # test deferred load of attribues, one select per instance self.assert_(h2.hoho==h4.hoho==h5.hoho==self.hohoval) self.assert_sql_count(testbase.db, go, 3) - + def go(): self.assert_(h1.counter == h4.counter==h5.counter==7) self.assert_sql_count(testbase.db, go, 1) - + def go(): self.assert_(h3.counter == h2.counter == 12) self.assert_(h2.foober == h3.foober == h4.foober == 'im foober') self.assert_(h5.foober=='im the new foober') self.assert_sql_count(testbase.db, go, 0) - + Session.close() - + l = Hoho.query.all() - + (h1, h2, h3, h4, h5) = l - + self.assert_(h1.hoho==self.althohoval) self.assert_(h3.hoho==self.althohoval) self.assert_(h2.hoho==h4.hoho==h5.hoho==self.hohoval) @@ -739,21 +742,21 @@ class DefaultTest(ORMTest): self.assert_(h1.counter == h4.counter==h5.counter==7) self.assert_(h2.foober == h3.foober == h4.foober == 'im foober') self.assert_(h5.foober=='im the new foober') - + def test_insert_nopostfetch(self): # populates the PassiveDefaults explicitly so there is no "post-update" class Hoho(object):pass mapper(Hoho, default_table) - + h1 = Hoho(hoho="15", counter="15") - + Session.commit() def go(): self.assert_(h1.hoho=="15") self.assert_(h1.counter=="15") self.assert_(h1.foober=="im foober") self.assert_sql_count(testbase.db, go, 0) - + def test_update(self): class Hoho(object):pass mapper(Hoho, default_table) @@ -766,7 +769,7 @@ class DefaultTest(ORMTest): class OneToManyTest(ORMTest): metadata = tables.metadata - + def define_tables(self, metadata): pass @@ -834,7 +837,7 @@ class OneToManyTest(ORMTest): u2.user_name = 'user2modified' u1.addresses.append(a3) del u1.addresses[0] - self.assert_sql(testbase.db, lambda: Session.commit(), + self.assert_sql(testbase.db, lambda: Session.commit(), [ ( "UPDATE users SET user_name=:user_name WHERE users.user_id = :users_user_id", @@ -951,10 +954,10 @@ class OneToManyTest(ORMTest): m2 = mapper(Address, addresses) m = mapper(User, users, properties={ 'boston_addresses' : relation(m2, primaryjoin= - and_(users.c.user_id==addresses.c.user_id, + and_(users.c.user_id==addresses.c.user_id, addresses.c.email_address.like('%boston%'))), 'newyork_addresses' : relation(m2, primaryjoin= - and_(users.c.user_id==addresses.c.user_id, + and_(users.c.user_id==addresses.c.user_id, addresses.c.email_address.like('%newyork%'))), }) u = User() @@ -971,7 +974,7 @@ class SaveTest(ORMTest): metadata = tables.metadata def define_tables(self, metadata): pass - + def setUp(self): super(SaveTest, self).setUp() keywords.insert().execute( @@ -993,7 +996,7 @@ class SaveTest(ORMTest): u2.user_name = 'savetester2' Session.save(u) - + Session.flush([u]) Session.commit() @@ -1001,7 +1004,7 @@ class SaveTest(ORMTest): nu = Session.get(m, u.user_id) print "U: " + repr(u) + "NU: " + repr(nu) self.assert_(u is nu) - + # clear out the identity map, so next get forces a SELECT Session.close() @@ -1009,7 +1012,7 @@ class SaveTest(ORMTest): nu = Session.get(m, u.user_id) self.assert_(u is not nu and u.user_id == nu.user_id and nu.user_name == 'savetester') Session.close() - + # change first users name and save Session.update(u) u.user_name = 'modifiedname' @@ -1022,7 +1025,7 @@ class SaveTest(ORMTest): print repr(u.user_id), repr(userlist[0].user_id), repr(userlist[0].user_name) self.assert_(u.user_id == userlist[0].user_id and userlist[0].user_name == 'modifiedname') self.assert_(u2.user_id == userlist[1].user_id and userlist[1].user_name == 'savetester2') - + def test_synonym(self): class User(object): def _get_name(self): @@ -1030,11 +1033,11 @@ class SaveTest(ORMTest): def _set_name(self, name): self.user_name = name + ":User" name = property(_get_name, _set_name) - + mapper(User, users, properties={ 'name':synonym('user_name') }) - + u = User() u.name = "some name" assert u.name == 'User:some name:User' @@ -1043,15 +1046,15 @@ class SaveTest(ORMTest): Session.clear() u = Session.query(User).first() assert u.name == 'User:some name:User' - + def test_lazyattr_commit(self): """tests that when a lazy-loaded list is unloaded, and a commit occurs, that the 'passive' call on that list does not blow away its value""" - + m1 = mapper(User, users, properties = { 'addresses': relation(mapper(Address, addresses)) }) - + u = User() u.addresses.append(Address()) u.addresses.append(Address()) @@ -1064,10 +1067,10 @@ class SaveTest(ORMTest): u1.user_name = 'newname' Session.commit() self.assert_(len(u1.addresses) == 4) - + def test_inherits(self): m1 = mapper(User, users) - + class AddressUser(User): """a user object that also has the users mailing address.""" pass @@ -1077,20 +1080,20 @@ class SaveTest(ORMTest): AddressUser, addresses, inherits=m1 ) - + au = AddressUser() Session.commit() Session.close() l = Session.query(AddressUser).selectone() self.assert_(l.user_id == au.user_id and l.address_id == au.address_id) - + def test_deferred(self): """test deferred column operations""" - + mapper(User, users, properties={ 'user_name':deferred(users.c.user_name) }) - + # dont set deferred attribute, commit session u = User() u.user_id=42 @@ -1101,7 +1104,7 @@ class SaveTest(ORMTest): Session.commit() assert list(Session.execute(users.select(), mapper=User)) == [(42, 'some name')] Session.clear() - + # assert that a set operation doesn't trigger a load operation u = Session.query(User).filter(User.user_name=='some name').one() def go(): @@ -1109,19 +1112,19 @@ class SaveTest(ORMTest): self.assert_sql_count(testbase.db, go, 0) Session.flush() assert list(Session.execute(users.select(), mapper=User)) == [(42, 'some other name')] - + Session.clear() - + # test assigning None to an unloaded deferred also works u = Session.query(User).filter(User.user_name=='some other name').one() u.user_name = None Session.flush() assert list(Session.execute(users.select(), mapper=User)) == [(42, None)] - - + + # why no support on oracle ? because oracle doesn't save - # "blank" strings; it saves a single space character. - @testing.unsupported('oracle') + # "blank" strings; it saves a single space character. + @testing.unsupported('oracle') def test_dont_update_blanks(self): mapper(User, users) u = User() @@ -1138,13 +1141,13 @@ class SaveTest(ORMTest): """tests a save of an object where each instance spans two tables. also tests redefinition of the keynames for the column properties.""" usersaddresses = sql.join(users, addresses, users.c.user_id == addresses.c.user_id) - m = mapper(User, usersaddresses, + m = mapper(User, usersaddresses, properties = dict( - email = addresses.c.email_address, + email = addresses.c.email_address, foo_id = [users.c.user_id, addresses.c.user_id], ) ) - + u = User() u.user_name = 'multitester' u.email = 'multi@test.org' @@ -1153,10 +1156,10 @@ class SaveTest(ORMTest): id = m.primary_key_from_instance(u) Session.close() - + u = Session.get(User, id) assert u.user_name == 'multitester' - + usertable = users.select(users.c.user_id.in_([u.foo_id])).execute().fetchall() self.assertEqual(usertable[0].values(), [u.foo_id, 'multitester']) addresstable = addresses.select(addresses.c.address_id.in_([u.address_id])).execute().fetchall() @@ -1174,14 +1177,14 @@ class SaveTest(ORMTest): Session.close() u = Session.get(User, id) assert u.user_name == 'imnew' - + def test_history_get(self): """tests that the history properly lazy-fetches data when it wasnt otherwise loaded""" mapper(User, users, properties={ 'addresses':relation(Address, cascade="all, delete-orphan") }) mapper(Address, addresses) - + u = User() u.addresses.append(Address()) u.addresses.append(Address()) @@ -1192,12 +1195,12 @@ class SaveTest(ORMTest): Session.commit() assert users.count().scalar() == 0 assert addresses.count().scalar() == 0 - - - + + + def test_batchmode(self): """test the 'batch=False' flag on mapper()""" - + class TestExtension(MapperExtension): def before_insert(self, mapper, connection, instance): self.current_instance = instance @@ -1209,9 +1212,9 @@ class SaveTest(ORMTest): u2 = User() u2.username = 'user2' Session.commit() - + clear_mappers() - + m = mapper(User, users, extension=TestExtension()) u1 = User() u1.username = 'user1' @@ -1222,14 +1225,14 @@ class SaveTest(ORMTest): assert False except AssertionError: assert True - - + + class ManyToOneTest(ORMTest): metadata = tables.metadata - + def define_tables(self, metadata): pass - + def test_m2o_onetoone(self): # TODO: put assertion in here !!! m = mapper(Address, addresses, properties = dict( @@ -1249,7 +1252,7 @@ class ManyToOneTest(ORMTest): a.user = User() a.user.user_name = elem['user_name'] objects.append(a) - + Session.commit() objects[2].email_address = 'imnew@foo.bar' objects[3].user = User() @@ -1263,11 +1266,11 @@ class ManyToOneTest(ORMTest): "UPDATE email_addresses SET email_address=:email_address WHERE email_addresses.address_id = :email_addresses_address_id": lambda ctx: {'email_address': 'imnew@foo.bar', 'email_addresses_address_id': objects[2].address_id} , - + "UPDATE email_addresses SET user_id=:user_id WHERE email_addresses.address_id = :email_addresses_address_id": lambda ctx: {'user_id': objects[3].user.user_id, 'email_addresses_address_id': objects[3].address_id} }, - + ], with_sequences=[ ( @@ -1278,11 +1281,11 @@ class ManyToOneTest(ORMTest): "UPDATE email_addresses SET email_address=:email_address WHERE email_addresses.address_id = :email_addresses_address_id": lambda ctx: {'email_address': 'imnew@foo.bar', 'email_addresses_address_id': objects[2].address_id} , - + "UPDATE email_addresses SET user_id=:user_id WHERE email_addresses.address_id = :email_addresses_address_id": lambda ctx: {'user_id': objects[3].user.user_id, 'email_addresses_address_id': objects[3].address_id} }, - + ]) l = sql.select([users, addresses], sql.and_(users.c.user_id==addresses.c.user_id, addresses.c.address_id==a.address_id)).execute() assert l.fetchone().values() == [a.user.user_id, 'asdf8d', a.address_id, a.user_id, 'theater@foo.com'] @@ -1296,7 +1299,7 @@ class ManyToOneTest(ORMTest): a1.email_address = 'emailaddress1' u1 = User() u1.user_name='user1' - + a1.user = u1 Session.commit() Session.close() @@ -1357,7 +1360,7 @@ class ManyToOneTest(ORMTest): u1 = Session.query(User).get(u1.user_id) u2 = Session.query(User).get(u2.user_id) assert a1.user is u1 - + a1.user = u2 Session.commit() Session.close() @@ -1390,13 +1393,13 @@ class ManyToOneTest(ORMTest): assert sess.query(Address).get(a1.address_id).user is None assert sess.query(User).get(u1.user_id).addresses == [] - + class ManyToManyTest(ORMTest): metadata = tables.metadata - + def define_tables(self, metadata): pass - + def test_manytomany(self): items = orderitems @@ -1436,7 +1439,7 @@ class ManyToManyTest(ORMTest): item.keywords.append(k) Session.commit() - + l = Session.query(m).select(items.c.item_name.in_([e['item_name'] for e in data[1:]]), order_by=[items.c.item_name]) self.assert_result(l, *data) @@ -1456,7 +1459,7 @@ class ManyToManyTest(ORMTest): lambda ctx: [{'item_id': objects[5].item_id, 'keyword_id': k.keyword_id}] ) ], - + with_sequences = [ { "UPDATE items SET item_name=:item_name WHERE items.item_id = :items_item_id": @@ -1478,12 +1481,12 @@ class ManyToManyTest(ORMTest): "DELETE FROM itemkeywords WHERE itemkeywords.item_id = :item_id AND itemkeywords.keyword_id = :keyword_id", [{'item_id': objects[5].item_id, 'keyword_id': dkid}] ), - ( + ( "INSERT INTO itemkeywords (item_id, keyword_id) VALUES (:item_id, :keyword_id)", lambda ctx: [{'item_id': objects[2].item_id, 'keyword_id': k.keyword_id}] ) ]) - + Session.delete(objects[3]) Session.commit() @@ -1502,7 +1505,7 @@ class ManyToManyTest(ORMTest): i.keywords.append(k1) i.keywords.append(k2) Session.commit() - + assert itemkeywords.count().scalar() == 2 i.keywords = [] Session.commit() @@ -1510,19 +1513,19 @@ class ManyToManyTest(ORMTest): def test_scalar(self): """test that dependency.py doesnt try to delete an m2m relation referencing None.""" - + mapper(Keyword, keywords) mapper(Item, orderitems, properties = dict( keyword = relation(Keyword, secondary=itemkeywords, uselist=False), )) - + i = Item() Session.commit() Session.delete(i) Session.commit() - - + + def test_manytomany_update(self): """tests some history operations on a many to many""" @@ -1533,7 +1536,7 @@ class ManyToManyTest(ORMTest): return other.__class__ == Keyword and other.name == self.name def __repr__(self): return "Keyword(%s, %s)" % (getattr(self, 'keyword_id', 'None'), self.name) - + mapper(Keyword, keywords) mapper(Item, orderitems, properties = dict( keywords = relation(Keyword, secondary=itemkeywords, lazy=False, order_by=keywords.c.name), @@ -1546,18 +1549,18 @@ class ManyToManyTest(ORMTest): item.keywords.append(k2) item.keywords.append(k3) Session.commit() - + item.keywords = [] item.keywords.append(k1) item.keywords.append(k2) Session.commit() - + Session.close() item = Session.query(Item).get(item.item_id) print [k1, k2] print item.keywords assert item.keywords == [k1, k2] - + def test_association(self): """basic test of an association object""" class IKAssociation(object): @@ -1578,29 +1581,29 @@ class ManyToManyTest(ORMTest): )) data = [Item, - {'item_name': 'a_item1', 'keywords' : (IKAssociation, + {'item_name': 'a_item1', 'keywords' : (IKAssociation, [ {'keyword' : (Keyword, {'name': 'big'})}, - {'keyword' : (Keyword, {'name': 'green'})}, + {'keyword' : (Keyword, {'name': 'green'})}, {'keyword' : (Keyword, {'name': 'purple'})}, {'keyword' : (Keyword, {'name': 'round'})} ] - ) + ) }, - {'item_name': 'a_item2', 'keywords' : (IKAssociation, + {'item_name': 'a_item2', 'keywords' : (IKAssociation, [ {'keyword' : (Keyword, {'name': 'huge'})}, - {'keyword' : (Keyword, {'name': 'violet'})}, + {'keyword' : (Keyword, {'name': 'violet'})}, {'keyword' : (Keyword, {'name': 'yellow'})} ] - ) + ) }, - {'item_name': 'a_item3', 'keywords' : (IKAssociation, + {'item_name': 'a_item3', 'keywords' : (IKAssociation, [ {'keyword' : (Keyword, {'name': 'big'})}, - {'keyword' : (Keyword, {'name': 'blue'})}, + {'keyword' : (Keyword, {'name': 'blue'})}, ] - ) + ) } ] for elem in data[1:]: @@ -1621,9 +1624,9 @@ class ManyToManyTest(ORMTest): Session.close() l = Item.query.filter(items.c.item_name.in_([e['item_name'] for e in data[1:]])).order_by(items.c.item_name).all() self.assert_result(l, *data) - + class SaveTest2(ORMTest): - + def define_tables(self, metadata): global users, addresses users = Table('users', metadata, @@ -1636,7 +1639,7 @@ class SaveTest2(ORMTest): Column('rel_user_id', Integer, ForeignKey(users.c.user_id)), Column('email_address', String(20)), ) - + def test_m2o_nonmatch(self): m = mapper(Address, addresses, properties = dict( user = relation(mapper(User, users), lazy = True, uselist = False) @@ -1670,7 +1673,7 @@ class SaveTest2(ORMTest): {'rel_user_id': 2, 'email_address': 'thesdf@asdf.com'} ) ], - + with_sequences = [ ( "INSERT INTO users (user_id, user_name) VALUES (:user_id, :user_name)", @@ -1737,19 +1740,19 @@ class SaveTest3(ORMTest): class BooleanColTest(ORMTest): def define_tables(self, metadata): global t - t =Table('t1', metadata, + t =Table('t1', metadata, Column('id', Integer, primary_key=True), Column('name', String(30)), Column('value', Boolean)) - + def test_boolean(self): # use the regular mapper from sqlalchemy.orm import mapper - + class T(fixtures.Base): pass mapper(T, t) - + sess = create_session() t1 = T(value=True, name="t1") t2 = T(value=False, name="t2") @@ -1757,9 +1760,9 @@ class BooleanColTest(ORMTest): sess.save(t1) sess.save(t2) sess.save(t3) - + sess.flush() - + for clear in (False, True): if clear: sess.clear() @@ -1770,7 +1773,7 @@ class BooleanColTest(ORMTest): if clear: sess.clear() self.assertEquals(sess.query(T).filter(T.value==False).all(), [T(value=False, name="t2")]) - + t2 = sess.query(T).get(t2.id) t2.value = True sess.flush() @@ -1778,16 +1781,16 @@ class BooleanColTest(ORMTest): t2.value = False sess.flush() self.assertEquals(sess.query(T).filter(T.value==True).all(), [T(value=True, name="t1"),T(value=True, name="t3")]) - - + + class RowSwitchTest(ORMTest): def define_tables(self, metadata): global t1, t2, t3, t1t3 - + global T1, T2, T3 - + Session.remove() - + # parent t1 = Table('t1', metadata, Column('id', Integer, primary_key=True), @@ -1811,7 +1814,7 @@ class RowSwitchTest(ORMTest): Column('t1id', Integer, ForeignKey('t1.id'),nullable=False), Column('t3id', Integer, ForeignKey('t3.id'),nullable=False), ) - + class T1(fixtures.Base): pass @@ -1820,29 +1823,29 @@ class RowSwitchTest(ORMTest): class T3(fixtures.Base): pass - + def tearDown(self): Session.remove() super(RowSwitchTest, self).tearDown() - + def test_onetomany(self): mapper(T1, t1, properties={ 't2s':relation(T2, cascade="all, delete-orphan") }) mapper(T2, t2) - + sess = Session(autoflush=False) - + o1 = T1(data='some t1', id=1) o1.t2s.append(T2(data='some t2', id=1)) o1.t2s.append(T2(data='some other t2', id=2)) - + sess.save(o1) sess.flush() - + assert list(sess.execute(t1.select(), mapper=T1)) == [(1, 'some t1')] assert list(sess.execute(t2.select(), mapper=T1)) == [(1, 'some t2', 1), (2, 'some other t2', 1)] - + o2 = T1(data='some other t1', id=o1.id, t2s=[ T2(data='third t2', id=3), T2(data='fourth t2', id=4), @@ -1885,7 +1888,7 @@ class RowSwitchTest(ORMTest): assert list(sess.execute(t3.select(), mapper=T1)) == [(3, 'third t3'), (4, 'fourth t3')] def test_manytoone(self): - + mapper(T2, t2, properties={ 't1':relation(T1) }) @@ -1910,8 +1913,8 @@ class RowSwitchTest(ORMTest): assert list(sess.execute(t1.select(), mapper=T1)) == [(2, 'some other t1')] assert list(sess.execute(t2.select(), mapper=T1)) == [(1, 'some other t2', 2)] - - - + + + if __name__ == "__main__": - testbase.main() + testbase.main() |