diff options
Diffstat (limited to 'test/ext/test_serializer.py')
| -rw-r--r-- | test/ext/test_serializer.py | 169 |
1 files changed, 86 insertions, 83 deletions
diff --git a/test/ext/test_serializer.py b/test/ext/test_serializer.py index 4e26e5b9a..45f55e1c9 100644 --- a/test/ext/test_serializer.py +++ b/test/ext/test_serializer.py @@ -3,12 +3,13 @@ from sqlalchemy.ext import serializer from sqlalchemy import exc import sqlalchemy as sa from sqlalchemy.test import testing -from sqlalchemy import MetaData, Integer, String, ForeignKey, select, desc, func, util +from sqlalchemy import MetaData, Integer, String, ForeignKey, select, \ + desc, func, util from sqlalchemy.test.schema import Table from sqlalchemy.test.schema import Column -from sqlalchemy.orm import relationship, sessionmaker, scoped_session, class_mapper, mapper, joinedload, compile_mappers, aliased +from sqlalchemy.orm import relationship, sessionmaker, scoped_session, \ + class_mapper, mapper, joinedload, compile_mappers, aliased from sqlalchemy.test.testing import eq_ - from test.orm._base import ComparableEntity, MappedTest @@ -18,125 +19,127 @@ class User(ComparableEntity): class Address(ComparableEntity): pass + class SerializeTest(MappedTest): + run_setup_mappers = 'once' run_inserts = 'once' run_deletes = None - + @classmethod def define_tables(cls, metadata): global users, addresses - users = Table('users', metadata, - Column('id', Integer, primary_key=True), - Column('name', String(50)) - ) - addresses = Table('addresses', metadata, - Column('id', Integer, primary_key=True), - Column('email', String(50)), - Column('user_id', Integer, ForeignKey('users.id')), - ) + users = Table('users', metadata, Column('id', Integer, + primary_key=True), Column('name', String(50))) + addresses = Table('addresses', metadata, Column('id', Integer, + primary_key=True), Column('email', + String(50)), Column('user_id', Integer, + ForeignKey('users.id'))) @classmethod def setup_mappers(cls): global Session Session = scoped_session(sessionmaker()) - - mapper(User, users, properties={ - 'addresses':relationship(Address, backref='user', order_by=addresses.c.id) - }) + mapper(User, users, properties={'addresses' + : relationship(Address, backref='user', + order_by=addresses.c.id)}) mapper(Address, addresses) - compile_mappers() - + @classmethod def insert_data(cls): - params = [dict(zip(('id', 'name'), column_values)) for column_values in - [(7, 'jack'), - (8, 'ed'), - (9, 'fred'), - (10, 'chuck')] - ] + params = [dict(zip(('id', 'name'), column_values)) + for column_values in [(7, 'jack'), (8, 'ed'), (9, + 'fred'), (10, 'chuck')]] users.insert().execute(params) - - addresses.insert().execute( - [dict(zip(('id', 'user_id', 'email'), column_values)) for column_values in - [(1, 7, "jack@bean.com"), - (2, 8, "ed@wood.com"), - (3, 8, "ed@bettyboop.com"), - (4, 8, "ed@lala.com"), - (5, 9, "fred@fred.com")] - ] - ) - + addresses.insert().execute([dict(zip(('id', 'user_id', 'email' + ), column_values)) + for column_values in [(1, 7, + 'jack@bean.com'), (2, 8, + 'ed@wood.com'), (3, 8, + 'ed@bettyboop.com'), (4, 8, + 'ed@lala.com'), (5, 9, + 'fred@fred.com')]]) + def test_tables(self): - assert serializer.loads(serializer.dumps(users, -1), users.metadata, Session) is users + assert serializer.loads(serializer.dumps(users, -1), + users.metadata, Session) is users def test_columns(self): - assert serializer.loads(serializer.dumps(users.c.name, -1), users.metadata, Session) is users.c.name - + assert serializer.loads(serializer.dumps(users.c.name, -1), + users.metadata, Session) is users.c.name + def test_mapper(self): user_mapper = class_mapper(User) - assert serializer.loads(serializer.dumps(user_mapper, -1), None, None) is user_mapper - + assert serializer.loads(serializer.dumps(user_mapper, -1), + None, None) is user_mapper + def test_attribute(self): - assert serializer.loads(serializer.dumps(User.name, -1), None, None) is User.name - + assert serializer.loads(serializer.dumps(User.name, -1), None, + None) is User.name + def test_expression(self): - - expr = select([users]).select_from(users.join(addresses)).limit(5) - re_expr = serializer.loads(serializer.dumps(expr, -1), users.metadata, None) - eq_( - str(expr), - str(re_expr) - ) - + expr = \ + select([users]).select_from(users.join(addresses)).limit(5) + re_expr = serializer.loads(serializer.dumps(expr, -1), + users.metadata, None) + eq_(str(expr), str(re_expr)) assert re_expr.bind is testing.db - eq_( - re_expr.execute().fetchall(), - [(7, u'jack'), (8, u'ed'), (8, u'ed'), (8, u'ed'), (9, u'fred')] - ) - + eq_(re_expr.execute().fetchall(), [(7, u'jack'), (8, u'ed'), + (8, u'ed'), (8, u'ed'), (9, u'fred')]) + def test_query(self): - q = Session.query(User).filter(User.name=='ed').options(joinedload(User.addresses)) - eq_(q.all(), [User(name='ed', addresses=[Address(id=2), Address(id=3), Address(id=4)])]) - - q2 = serializer.loads(serializer.dumps(q, -1), users.metadata, Session) + q = Session.query(User).filter(User.name == 'ed' + ).options(joinedload(User.addresses)) + eq_(q.all(), [User(name='ed', addresses=[Address(id=2), + Address(id=3), Address(id=4)])]) + q2 = serializer.loads(serializer.dumps(q, -1), users.metadata, + Session) + def go(): - eq_(q2.all(), [User(name='ed', addresses=[Address(id=2), Address(id=3), Address(id=4)])]) - self.assert_sql_count(testing.db, go, 1) - - eq_(q2.join(User.addresses).filter(Address.email=='ed@bettyboop.com').value(func.count('*')), 1) + eq_(q2.all(), [User(name='ed', addresses=[Address(id=2), + Address(id=3), Address(id=4)])]) + self.assert_sql_count(testing.db, go, 1) + eq_(q2.join(User.addresses).filter(Address.email + == 'ed@bettyboop.com').value(func.count('*')), 1) u1 = Session.query(User).get(8) - - q = Session.query(Address).filter(Address.user==u1).order_by(desc(Address.email)) - q2 = serializer.loads(serializer.dumps(q, -1), users.metadata, Session) - - eq_(q2.all(), [Address(email='ed@wood.com'), Address(email='ed@lala.com'), Address(email='ed@bettyboop.com')]) - - q = Session.query(User).join(User.addresses).filter(Address.email.like('%fred%')) - q2 = serializer.loads(serializer.dumps(q, -1), users.metadata, Session) + q = Session.query(Address).filter(Address.user + == u1).order_by(desc(Address.email)) + q2 = serializer.loads(serializer.dumps(q, -1), users.metadata, + Session) + eq_(q2.all(), [Address(email='ed@wood.com'), + Address(email='ed@lala.com'), + Address(email='ed@bettyboop.com')]) + q = \ + Session.query(User).join(User.addresses).\ + filter(Address.email.like('%fred%')) + q2 = serializer.loads(serializer.dumps(q, -1), users.metadata, + Session) eq_(q2.all(), [User(name='fred')]) - eq_(list(q2.values(User.id, User.name)), [(9, u'fred')]) - @testing.exclude('sqlite', '<=', (3, 5, 9), 'id comparison failing on the buildbot') + @testing.exclude('sqlite', '<=', (3, 5, 9), + 'id comparison failing on the buildbot') def test_aliases(self): u7, u8, u9, u10 = Session.query(User).order_by(User.id).all() - ualias = aliased(User) - q = Session.query(User, ualias).join((ualias, User.id < ualias.id)).filter(User.id<9).order_by(User.id, ualias.id) - eq_(list(q.all()), [(u7, u8), (u7, u9), (u7, u10), (u8, u9), (u8, u10)]) - - q2 = serializer.loads(serializer.dumps(q, -1), users.metadata, Session) - - eq_(list(q2.all()), [(u7, u8), (u7, u9), (u7, u10), (u8, u9), (u8, u10)]) + q = Session.query(User, ualias).join((ualias, User.id + < ualias.id)).filter(User.id < 9).order_by(User.id, + ualias.id) + eq_(list(q.all()), [(u7, u8), (u7, u9), (u7, u10), (u8, u9), + (u8, u10)]) + q2 = serializer.loads(serializer.dumps(q, -1), users.metadata, + Session) + eq_(list(q2.all()), [(u7, u8), (u7, u9), (u7, u10), (u8, u9), + (u8, u10)]) def test_any(self): - r = User.addresses.any(Address.email=='x') + r = User.addresses.any(Address.email == 'x') ser = serializer.dumps(r, -1) x = serializer.loads(ser, users.metadata) eq_(str(r), str(x)) - + + if __name__ == '__main__': testing.main() |