diff options
Diffstat (limited to 'test/orm/test_naturalpks.py')
| -rw-r--r-- | test/orm/test_naturalpks.py | 103 |
1 files changed, 70 insertions, 33 deletions
diff --git a/test/orm/test_naturalpks.py b/test/orm/test_naturalpks.py index f0f3ebb21..bb7b8a71d 100644 --- a/test/orm/test_naturalpks.py +++ b/test/orm/test_naturalpks.py @@ -33,20 +33,20 @@ class NaturalPKTest(fixtures.MappedTest): addresses = Table('addresses', metadata, Column('email', String(50), primary_key=True), - Column('username', String(50), + Column('username', String(50), ForeignKey('users.username', **fk_args)), test_needs_fk=True) items = Table('items', metadata, Column('itemname', String(50), primary_key=True), - Column('description', String(100)), + Column('description', String(100)), test_needs_fk=True) users_to_items = Table('users_to_items', metadata, - Column('username', String(50), + Column('username', String(50), ForeignKey('users.username', **fk_args), primary_key=True), - Column('itemname', String(50), + Column('itemname', String(50), ForeignKey('items.itemname', **fk_args), primary_key=True), test_needs_fk=True) @@ -168,15 +168,15 @@ class NaturalPKTest(fixtures.MappedTest): def go(): sess.flush() if not passive_updates: - # test passive_updates=False; + # test passive_updates=False; #load addresses, update user, update 2 addresses - self.assert_sql_count(testing.db, go, 4) + self.assert_sql_count(testing.db, go, 4) else: # test passive_updates=True; update user - self.assert_sql_count(testing.db, go, 1) + self.assert_sql_count(testing.db, go, 1) sess.expunge_all() assert User(username='jack', addresses=[ - Address(username='jack'), + Address(username='jack'), Address(username='jack')]) == \ sess.query(User).get('jack') @@ -189,7 +189,6 @@ class NaturalPKTest(fixtures.MappedTest): u1 = sess.query(User).get('fred') eq_(User(username='fred', fullname='jack'), u1) - @testing.fails_on('sqlite', 'sqlite doesnt support ON UPDATE CASCADE') @testing.fails_on('oracle', 'oracle doesnt support ON UPDATE CASCADE') def test_manytoone_passive(self): @@ -198,6 +197,43 @@ class NaturalPKTest(fixtures.MappedTest): def test_manytoone_nonpassive(self): self._test_manytoone(False) + def test_manytoone_nonpassive_cold_mapping(self): + """test that the mapper-level m2o dependency processor + is set up even if the opposite side relationship + hasn't yet been part of a flush. + + """ + users, Address, addresses, User = (self.tables.users, + self.classes.Address, + self.tables.addresses, + self.classes.User) + + with testing.db.begin() as conn: + conn.execute(users.insert(), + username='jack', fullname='jack' + ) + conn.execute(addresses.insert(), + email='jack1', username='jack' + ) + conn.execute(addresses.insert(), + email='jack2', username='jack' + ) + + mapper(User, users) + mapper(Address, addresses, properties={ + 'user': relationship(User, + passive_updates=False) + }) + + sess = create_session() + u1 = sess.query(User).first() + a1, a2 = sess.query(Address).all() + u1.username = 'ed' + + def go(): + sess.flush() + self.assert_sql_count(testing.db, go, 3) + def _test_manytoone(self, passive_updates): users, Address, addresses, User = (self.tables.users, self.classes.Address, @@ -206,7 +242,7 @@ class NaturalPKTest(fixtures.MappedTest): mapper(User, users) mapper(Address, addresses, properties={ - 'user':relationship(User, passive_updates=passive_updates) + 'user': relationship(User, passive_updates=passive_updates) }) sess = create_session() @@ -238,6 +274,7 @@ class NaturalPKTest(fixtures.MappedTest): eq_([Address(username='ed'), Address(username='ed')], sess.query(Address).all()) + @testing.fails_on('sqlite', 'sqlite doesnt support ON UPDATE CASCADE') @testing.fails_on('oracle', 'oracle doesnt support ON UPDATE CASCADE') def test_onetoone_passive(self): @@ -349,9 +386,9 @@ class NaturalPKTest(fixtures.MappedTest): def test_manytomany_passive(self): self._test_manytomany(True) - # mysqldb executemany() of the association table fails to + # mysqldb executemany() of the association table fails to # report the correct row count - @testing.fails_if(lambda: testing.against('mysql') + @testing.fails_if(lambda: testing.against('mysql') and not testing.against('+zxjdbc')) def test_manytomany_nonpassive(self): self._test_manytomany(False) @@ -489,7 +526,7 @@ class ReversePKsTest(fixtures.MappedTest): session.add(a_editable) session.commit() - # do the switch in both directions - + # do the switch in both directions - # one or the other should raise the error # based on platform dictionary ordering a_published.status = ARCHIVED @@ -509,9 +546,9 @@ class ReversePKsTest(fixtures.MappedTest): class SelfReferentialTest(fixtures.MappedTest): - # mssql, mysql don't allow + # mssql, mysql don't allow # ON UPDATE on self-referential keys - __unsupported_on__ = ('mssql','mysql') + __unsupported_on__ = ('mssql','mysql') @classmethod def define_tables(cls, metadata): @@ -596,8 +633,8 @@ class SelfReferentialTest(fixtures.MappedTest): Node, nodes = self.classes.Node, self.tables.nodes mapper(Node, nodes, properties={ - 'parentnode':relationship(Node, - remote_side=nodes.c.name, + 'parentnode':relationship(Node, + remote_side=nodes.c.name, passive_updates=passive) } ) @@ -686,7 +723,7 @@ class NonPKCascadeTest(fixtures.MappedTest): u1.username = 'ed' sess.flush() assert u1.addresses[0].username == 'ed' - eq_(sa.select([addresses.c.username]).execute().fetchall(), + eq_(sa.select([addresses.c.username]).execute().fetchall(), [('ed',), ('ed',)]) sess.expunge_all() @@ -698,14 +735,14 @@ class NonPKCascadeTest(fixtures.MappedTest): def go(): sess.flush() if not passive_updates: - # test passive_updates=False; load addresses, + # test passive_updates=False; load addresses, # update user, update 2 addresses - self.assert_sql_count(testing.db, go, 4) + self.assert_sql_count(testing.db, go, 4) else: # test passive_updates=True; update user self.assert_sql_count(testing.db, go, 1) sess.expunge_all() - assert User(username='jack', + assert User(username='jack', addresses=[Address(username='jack'), Address(username='jack')]) == \ sess.query(User).get(u1.id) @@ -719,7 +756,7 @@ class NonPKCascadeTest(fixtures.MappedTest): a1 = sess.query(Address).get(a1.id) eq_(a1.username, None) - eq_(sa.select([addresses.c.username]).execute().fetchall(), + eq_(sa.select([addresses.c.username]).execute().fetchall(), [(None,), (None,)]) u1 = sess.query(User).get(u1.id) @@ -742,7 +779,7 @@ class CascadeToFKPKTest(fixtures.MappedTest, testing.AssertsCompiledSQL): test_needs_fk=True) Table('addresses', metadata, - Column('username', String(50), + Column('username', String(50), ForeignKey('users.username', **fk_args), primary_key=True ), @@ -777,7 +814,7 @@ class CascadeToFKPKTest(fixtures.MappedTest, testing.AssertsCompiledSQL): def _test_o2m_change(self, passive_updates): """Change the PK of a related entity to another. - "on update cascade" is not involved here, so the mapper has + "on update cascade" is not involved here, so the mapper has to do the UPDATE itself. """ @@ -945,7 +982,7 @@ class CascadeToFKPKTest(fixtures.MappedTest, testing.AssertsCompiledSQL): def _test_onetomany(self, passive_updates): """Change the PK of a related entity via foreign key cascade. - For databases that require "on update cascade", the mapper + For databases that require "on update cascade", the mapper has to identify the row by the new value, not the old, when it does the update. @@ -969,7 +1006,7 @@ class CascadeToFKPKTest(fixtures.MappedTest, testing.AssertsCompiledSQL): sess.flush() eq_(a1.username, 'ed') eq_(a2.username, 'ed') - eq_(sa.select([addresses.c.username]).execute().fetchall(), + eq_(sa.select([addresses.c.username]).execute().fetchall(), [('ed',), ('ed',)]) u1.username = 'jack' @@ -986,7 +1023,7 @@ class JoinedInheritanceTest(fixtures.MappedTest): """Test cascades of pk->pk/fk on joined table inh.""" # mssql doesn't allow ON UPDATE on self-referential keys - __unsupported_on__ = ('mssql',) + __unsupported_on__ = ('mssql',) __requires__ = 'skip_mysql_on_windows', @@ -1006,13 +1043,13 @@ class JoinedInheritanceTest(fixtures.MappedTest): Column('name', String(50), ForeignKey('person.name', **fk_args), primary_key=True), Column('primary_language', String(50)), - Column('boss_name', String(50), + Column('boss_name', String(50), ForeignKey('manager.name', **fk_args)), test_needs_fk=True ) Table('manager', metadata, - Column('name', String(50), + Column('name', String(50), ForeignKey('person.name', **fk_args), primary_key=True), Column('paperwork', String(50)), @@ -1057,12 +1094,12 @@ class JoinedInheritanceTest(fixtures.MappedTest): self.classes.Engineer, self.tables.engineer) - mapper(Person, person, polymorphic_on=person.c.type, + mapper(Person, person, polymorphic_on=person.c.type, polymorphic_identity='person', passive_updates=passive_updates) mapper(Engineer, engineer, inherits=Person, polymorphic_identity='engineer', properties={ - 'boss':relationship(Manager, + 'boss':relationship(Manager, primaryjoin=manager.c.name==engineer.c.boss_name, passive_updates=passive_updates ) @@ -1087,12 +1124,12 @@ class JoinedInheritanceTest(fixtures.MappedTest): self.classes.Engineer, self.tables.engineer) - mapper(Person, person, polymorphic_on=person.c.type, + mapper(Person, person, polymorphic_on=person.c.type, polymorphic_identity='person', passive_updates=passive_updates) mapper(Engineer, engineer, inherits=Person, polymorphic_identity='engineer', properties={ - 'boss':relationship(Manager, + 'boss':relationship(Manager, primaryjoin=manager.c.name==engineer.c.boss_name, passive_updates=passive_updates ) |