diff options
Diffstat (limited to 'test/orm/test_pickled.py')
| -rw-r--r-- | test/orm/test_pickled.py | 233 |
1 files changed, 114 insertions, 119 deletions
diff --git a/test/orm/test_pickled.py b/test/orm/test_pickled.py index f4e49c518..6b925806f 100644 --- a/test/orm/test_pickled.py +++ b/test/orm/test_pickled.py @@ -8,10 +8,10 @@ from sqlalchemy.testing import assert_raises_message from sqlalchemy import Integer, String, ForeignKey, exc, MetaData from sqlalchemy.testing.schema import Table, Column from sqlalchemy.orm import mapper, relationship, create_session, \ - sessionmaker, attributes, interfaces,\ - clear_mappers, exc as orm_exc,\ - configure_mappers, Session, lazyload_all,\ - lazyload, aliased + sessionmaker, attributes, interfaces,\ + clear_mappers, exc as orm_exc,\ + configure_mappers, Session, lazyload_all,\ + lazyload, aliased from sqlalchemy.orm import state as sa_state from sqlalchemy.orm import instrumentation from sqlalchemy.orm.collections import attribute_mapped_collection, \ @@ -27,43 +27,42 @@ class PickleTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): Table('users', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), + Column('id', Integer, primary_key=True, + test_needs_autoincrement=True), Column('name', String(30), nullable=False), test_needs_acid=True, - test_needs_fk=True - ) + test_needs_fk=True) Table('addresses', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), + Column('id', Integer, primary_key=True, + test_needs_autoincrement=True), Column('user_id', None, ForeignKey('users.id')), Column('email_address', String(50), nullable=False), test_needs_acid=True, - test_needs_fk=True - ) + test_needs_fk=True) Table('orders', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), - Column('user_id', None, ForeignKey('users.id')), - Column('address_id', None, ForeignKey('addresses.id')), - Column('description', String(30)), - Column('isopen', Integer), - test_needs_acid=True, - test_needs_fk=True - ) + Column('id', Integer, primary_key=True, + test_needs_autoincrement=True), + Column('user_id', None, ForeignKey('users.id')), + Column('address_id', None, ForeignKey('addresses.id')), + Column('description', String(30)), + Column('isopen', Integer), + test_needs_acid=True, + test_needs_fk=True) Table("dingalings", metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), + Column('id', Integer, primary_key=True, + test_needs_autoincrement=True), Column('address_id', None, ForeignKey('addresses.id')), Column('data', String(30)), test_needs_acid=True, - test_needs_fk=True - ) - + test_needs_fk=True) def test_transient(self): users, addresses = (self.tables.users, - self.tables.addresses) + self.tables.addresses) mapper(User, users, properties={ - 'addresses':relationship(Address, backref="user") + 'addresses': relationship(Address, backref="user") }) mapper(Address, addresses) @@ -82,7 +81,6 @@ class PickleTest(fixtures.MappedTest): def test_no_mappers(self): users = self.tables.users - umapper = mapper(User, users) u1 = User(name='ed') u1_pickled = pickle.dumps(u1, -1) @@ -91,13 +89,13 @@ class PickleTest(fixtures.MappedTest): assert_raises_message( orm_exc.UnmappedInstanceError, - "Cannot deserialize object of type <class 'sqlalchemy.testing.pickleable.User'> - no mapper()", + "Cannot deserialize object of type " + "<class 'sqlalchemy.testing.pickleable.User'> - no mapper()", pickle.loads, u1_pickled) def test_no_instrumentation(self): users = self.tables.users - umapper = mapper(User, users) u1 = User(name='ed') u1_pickled = pickle.dumps(u1, -1) @@ -111,10 +109,9 @@ class PickleTest(fixtures.MappedTest): # compiles the mapper eq_(str(u1), "User(name='ed')") - def test_class_deferred_cols(self): addresses, users = (self.tables.addresses, - self.tables.users) + self.tables.users) mapper(User, users, properties={ 'name': sa.orm.deferred(users.c.name), @@ -137,17 +134,19 @@ class PickleTest(fixtures.MappedTest): sess2 = create_session() sess2.add(u2) eq_(u2.name, 'ed') - eq_(u2, User(name='ed', addresses=[Address(email_address='ed@bar.com')])) + eq_(u2, User(name='ed', addresses=[ + Address(email_address='ed@bar.com')])) u2 = pickle.loads(pickle.dumps(u1)) sess2 = create_session() u2 = sess2.merge(u2, load=False) eq_(u2.name, 'ed') - eq_(u2, User(name='ed', addresses=[Address(email_address='ed@bar.com')])) + eq_(u2, User(name='ed', addresses=[ + Address(email_address='ed@bar.com')])) def test_instance_lazy_relation_loaders(self): users, addresses = (self.tables.users, - self.tables.addresses) + self.tables.addresses) mapper(User, users, properties={ 'addresses': relationship(Address, lazy='noload') @@ -155,19 +154,13 @@ class PickleTest(fixtures.MappedTest): mapper(Address, addresses) sess = Session() - u1 = User(name='ed', addresses=[ - Address( - email_address='ed@bar.com', - ) - ]) + u1 = User(name='ed', addresses=[Address(email_address='ed@bar.com')]) sess.add(u1) sess.commit() sess.close() - u1 = sess.query(User).options( - lazyload(User.addresses) - ).first() + u1 = sess.query(User).options(lazyload(User.addresses)).first() u2 = pickle.loads(pickle.dumps(u1)) sess = Session() @@ -175,8 +168,7 @@ class PickleTest(fixtures.MappedTest): assert u2.addresses def test_invalidated_flag_pickle(self): - users, addresses = (self.tables.users, - self.tables.addresses) + users, addresses = (self.tables.users, self.tables.addresses) mapper(User, users, properties={ 'addresses': relationship(Address, lazy='noload') @@ -190,8 +182,7 @@ class PickleTest(fixtures.MappedTest): eq_(len(u2.addresses), 2) def test_invalidated_flag_deepcopy(self): - users, addresses = (self.tables.users, - self.tables.addresses) + users, addresses = (self.tables.users, self.tables.addresses) mapper(User, users, properties={ 'addresses': relationship(Address, lazy='noload') @@ -206,11 +197,10 @@ class PickleTest(fixtures.MappedTest): @testing.requires.non_broken_pickle def test_instance_deferred_cols(self): - users, addresses = (self.tables.users, - self.tables.addresses) + users, addresses = (self.tables.users, self.tables.addresses) mapper(User, users, properties={ - 'addresses':relationship(Address, backref="user") + 'addresses': relationship(Address, backref="user") }) mapper(Address, addresses) @@ -222,9 +212,9 @@ class PickleTest(fixtures.MappedTest): sess.expunge_all() u1 = sess.query(User).\ - options(sa.orm.defer('name'), - sa.orm.defer('addresses.email_address')).\ - get(u1.id) + options(sa.orm.defer('name'), + sa.orm.defer('addresses.email_address')).\ + get(u1.id) assert 'name' not in u1.__dict__ assert 'addresses' not in u1.__dict__ @@ -236,7 +226,8 @@ class PickleTest(fixtures.MappedTest): ad = u2.addresses[0] assert 'email_address' not in ad.__dict__ eq_(ad.email_address, 'ed@bar.com') - eq_(u2, User(name='ed', addresses=[Address(email_address='ed@bar.com')])) + eq_(u2, User(name='ed', addresses=[ + Address(email_address='ed@bar.com')])) u2 = pickle.loads(pickle.dumps(u1)) sess2 = create_session() @@ -250,14 +241,14 @@ class PickleTest(fixtures.MappedTest): assert 'email_address' not in ad.__dict__ eq_(ad.email_address, 'ed@bar.com') - eq_(u2, User(name='ed', addresses=[Address(email_address='ed@bar.com')])) + eq_(u2, User(name='ed', addresses=[ + Address(email_address='ed@bar.com')])) def test_pickle_protocols(self): - users, addresses = (self.tables.users, - self.tables.addresses) + users, addresses = (self.tables.users, self.tables.addresses) mapper(User, users, properties={ - 'addresses':relationship(Address, backref="user") + 'addresses': relationship(Address, backref="user") }) mapper(Address, addresses) @@ -307,14 +298,14 @@ class PickleTest(fixtures.MappedTest): @testing.requires.non_broken_pickle def test_options_with_descriptors(self): users, addresses, dingalings = (self.tables.users, - self.tables.addresses, - self.tables.dingalings) + self.tables.addresses, + self.tables.dingalings) mapper(User, users, properties={ - 'addresses':relationship(Address, backref="user") + 'addresses': relationship(Address, backref="user") }) mapper(Address, addresses, properties={ - 'dingaling':relationship(Dingaling) + 'dingaling': relationship(Dingaling) }) mapper(Dingaling, dingalings) sess = create_session() @@ -343,20 +334,16 @@ class PickleTest(fixtures.MappedTest): m = MetaData() c1 = Table('c1', m, - Column('parent_id', String, - ForeignKey('p.id'), primary_key=True) - ) + Column('parent_id', String, ForeignKey('p.id'), + primary_key=True)) c2 = Table('c2', m, - Column('parent_id', String, - ForeignKey('p.id'), primary_key=True) - ) - p = Table('p', m, - Column('id', String, primary_key=True) - ) + Column('parent_id', String, ForeignKey('p.id'), + primary_key=True)) + p = Table('p', m, Column('id', String, primary_key=True)) mapper(Parent, p, properties={ - 'children1':relationship(Child1), - 'children2':relationship(Child2) + 'children1': relationship(Child1), + 'children2': relationship(Child2) }) mapper(Child1, c1) mapper(Child2, c2) @@ -386,87 +373,87 @@ class PickleTest(fixtures.MappedTest): users, addresses = self.tables.users, self.tables.addresses mapper(User, users, properties={ - 'addresses':relationship( - Address, - collection_class= - attribute_mapped_collection('email_address') - ) + 'addresses': relationship( + Address, + collection_class=attribute_mapped_collection('email_address') + ) }) mapper(Address, addresses) u1 = User() - u1.addresses = {"email1":Address(email_address="email1")} + u1.addresses = {"email1": Address(email_address="email1")} for loads, dumps in picklers(): repickled = loads(dumps(u1)) eq_(u1.addresses, repickled.addresses) eq_(repickled.addresses['email1'], - Address(email_address="email1")) + Address(email_address="email1")) def test_column_mapped_collection(self): users, addresses = self.tables.users, self.tables.addresses mapper(User, users, properties={ - 'addresses':relationship( - Address, - collection_class= - column_mapped_collection( - addresses.c.email_address) - ) + 'addresses': relationship( + Address, + collection_class=column_mapped_collection( + addresses.c.email_address) + ) }) mapper(Address, addresses) u1 = User() u1.addresses = { - "email1":Address(email_address="email1"), - "email2":Address(email_address="email2") + "email1": Address(email_address="email1"), + "email2": Address(email_address="email2") } for loads, dumps in picklers(): repickled = loads(dumps(u1)) eq_(u1.addresses, repickled.addresses) eq_(repickled.addresses['email1'], - Address(email_address="email1")) + Address(email_address="email1")) def test_composite_column_mapped_collection(self): users, addresses = self.tables.users, self.tables.addresses mapper(User, users, properties={ - 'addresses':relationship( - Address, - collection_class= - column_mapped_collection([ - addresses.c.id, - addresses.c.email_address]) - ) + 'addresses': relationship( + Address, + collection_class=column_mapped_collection([ + addresses.c.id, + addresses.c.email_address]) + ) }) mapper(Address, addresses) u1 = User() u1.addresses = { - (1, "email1"):Address(id=1, email_address="email1"), - (2, "email2"):Address(id=2, email_address="email2") + (1, "email1"): Address(id=1, email_address="email1"), + (2, "email2"): Address(id=2, email_address="email2") } for loads, dumps in picklers(): repickled = loads(dumps(u1)) eq_(u1.addresses, repickled.addresses) eq_(repickled.addresses[(1, 'email1')], - Address(id=1, email_address="email1")) + Address(id=1, email_address="email1")) + class PolymorphicDeferredTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): Table('users', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), - Column('name', String(30)), - Column('type', String(30))) + Column('id', Integer, primary_key=True, + test_needs_autoincrement=True), + Column('name', String(30)), + Column('type', String(30))) Table('email_users', metadata, - Column('id', Integer, ForeignKey('users.id'), primary_key=True), - Column('email_address', String(30))) - + Column('id', Integer, ForeignKey('users.id'), primary_key=True), + Column('email_address', String(30))) def test_polymorphic_deferred(self): email_users, users = (self.tables.email_users, - self.tables.users, - ) + self.tables.users, + ) - mapper(User, users, polymorphic_identity='user', polymorphic_on=users.c.type) - mapper(EmailUser, email_users, inherits=User, polymorphic_identity='emailuser') + mapper(User, users, polymorphic_identity='user', + polymorphic_on=users.c.type) + mapper(EmailUser, email_users, inherits=User, + polymorphic_identity='emailuser') eu = EmailUser(name="user1", email_address='foo@bar.com') sess = create_session() @@ -481,6 +468,7 @@ class PolymorphicDeferredTest(fixtures.MappedTest): assert 'email_address' not in eu2.__dict__ eq_(eu2.email_address, 'foo@bar.com') + class TupleLabelTest(_fixtures.FixtureTest): @classmethod def setup_classes(cls): @@ -488,14 +476,18 @@ class TupleLabelTest(_fixtures.FixtureTest): @classmethod def setup_mappers(cls): - users, addresses, orders = cls.tables.users, cls.tables.addresses, cls.tables.orders + users, addresses, orders = (cls.tables.users, cls.tables.addresses, + cls.tables.orders) mapper(User, users, properties={ - 'addresses':relationship(Address, backref='user', order_by=addresses.c.id), - 'orders':relationship(Order, backref='user', order_by=orders.c.id), # o2m, m2o + 'addresses': relationship(Address, backref='user', + order_by=addresses.c.id), + # o2m, m2o + 'orders': relationship(Order, backref='user', + order_by=orders.c.id), }) mapper(Address, addresses) mapper(Order, orders, properties={ - 'address':relationship(Address), # m2o + 'address': relationship(Address), # m2o }) def test_tuple_labeling(self): @@ -519,7 +511,8 @@ class TupleLabelTest(_fixtures.FixtureTest): eq_(row.name, row[0]) eq_(row.foobar, row[1]) - for row in sess.query(User).values(User.name, User.id.label('foobar')): + for row in sess.query(User).values(User.name, + User.id.label('foobar')): if pickled is not False: row = pickle.loads(pickle.dumps(row, pickled)) eq_(list(row.keys()), ['name', 'foobar']) @@ -534,7 +527,8 @@ class TupleLabelTest(_fixtures.FixtureTest): eq_(row.User, row[0]) oalias = aliased(Order, name='orders') - for row in sess.query(User, oalias).join(oalias, User.orders).all(): + for row in sess.query(User, oalias).join(oalias, User.orders) \ + .all(): if pickled is not False: row = pickle.loads(pickle.dumps(row, pickled)) eq_(list(row.keys()), ['User', 'orders']) @@ -551,23 +545,25 @@ class TupleLabelTest(_fixtures.FixtureTest): ret = sess.query(User, Address).join(User.addresses).all() pickle.loads(pickle.dumps(ret, pickled)) + class CustomSetupTeardownTest(fixtures.MappedTest): @classmethod def define_tables(cls, metadata): Table('users', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), + Column('id', Integer, primary_key=True, + test_needs_autoincrement=True), Column('name', String(30), nullable=False), test_needs_acid=True, - test_needs_fk=True - ) + test_needs_fk=True) Table('addresses', metadata, - Column('id', Integer, primary_key=True, test_needs_autoincrement=True), + Column('id', Integer, primary_key=True, + test_needs_autoincrement=True), Column('user_id', None, ForeignKey('users.id')), Column('email_address', String(50), nullable=False), test_needs_acid=True, - test_needs_fk=True - ) + test_needs_fk=True) + def test_rebuild_state(self): """not much of a 'test', but illustrate how to remove instance-level state before pickling. @@ -584,4 +580,3 @@ class CustomSetupTeardownTest(fixtures.MappedTest): u2 = pickle.loads(pickle.dumps(u1)) attributes.manager_of_class(User).setup_instance(u2) assert attributes.instance_state(u2) - |