sqlalchemy/test/ext/test_serializer.py

# coding: utf-8

from sqlalchemy.ext import serializer
from sqlalchemy import testing
from sqlalchemy import Integer, String, ForeignKey, select, \
    desc, func, util, MetaData
from sqlalchemy.testing.schema import Table
from sqlalchemy.testing.schema import Column
from sqlalchemy.orm import relationship, sessionmaker, scoped_session, \
    class_mapper, mapper, joinedload, configure_mappers, aliased
from sqlalchemy.testing import eq_, AssertsCompiledSQL
from sqlalchemy.util import u, ue

from sqlalchemy.testing import fixtures

class User(fixtures.ComparableEntity):
    pass

class Address(fixtures.ComparableEntity):
    pass

users = addresses = Session = None

class SerializeTest(AssertsCompiledSQL, fixtures.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')))

    @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(Address, addresses)
        configure_mappers()

    @classmethod
    def insert_data(cls):
        params = [dict(list(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(list(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

    def test_columns(self):
        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

    def test_attribute(self):
        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))
        assert re_expr.bind is testing.db
        eq_(re_expr.execute().fetchall(), [(7, 'jack'), (8, 'ed'),
            (8, 'ed'), (8, 'ed'), (9, 'fred')])

    def test_query_one(self):
        q = Session.query(User).\
                filter(User.name == 'ed').\
                    options(joinedload(User.addresses))

        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)
        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')])

    @testing.requires.non_broken_pickle
    def test_query_two(self):
        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, 'fred')])

    # fails too often/randomly
    #@testing.requires.non_broken_pickle
    #def test_query_three(self):
    #    ua = aliased(User)
    #    q = \
    #        Session.query(ua).join(ua.addresses).\
    #           filter(Address.email.like('%fred%'))
    #    q2 = serializer.loads(serializer.dumps(q, -1), users.metadata,
    #                          Session)
    #    eq_(q2.all(), [User(name='fred')])
    #
        # try to pull out the aliased entity here...
    #    ua_2 = q2._entities[0].entity_zero.entity
    #    eq_(list(q2.values(ua_2.id, ua_2.name)), [(9, 'fred')])

    @testing.requires.non_broken_pickle
    def test_orm_join(self):
        from sqlalchemy.orm.util import join

        j = join(User, Address, User.addresses)

        j2 = serializer.loads(serializer.dumps(j, -1), users.metadata)
        assert j2.left is j.left
        assert j2.right is j.right
        assert j2._target_adapter._next

    @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)])

    @testing.requires.non_broken_pickle
    def test_any(self):
        r = User.addresses.any(Address.email == 'x')
        ser = serializer.dumps(r, -1)
        x = serializer.loads(ser, users.metadata)
        eq_(str(r), str(x))

    def test_unicode(self):
        m = MetaData()
        t = Table(ue('\u6e2c\u8a66'), m,
                Column(ue('\u6e2c\u8a66_id'), Integer))

        expr = select([t]).where(t.c[ue('\u6e2c\u8a66_id')] == 5)

        expr2 = serializer.loads(serializer.dumps(expr, -1), m)

        self.assert_compile(
            expr2,
            ue('SELECT "\u6e2c\u8a66"."\u6e2c\u8a66_id" FROM "\u6e2c\u8a66" '
                'WHERE "\u6e2c\u8a66"."\u6e2c\u8a66_id" = :\u6e2c\u8a66_id_1'),
            dialect="default"
        )


if __name__ == '__main__':
    testing.main()