path: root/test/ext/test_mutable.py

from sqlalchemy import Integer
from sqlalchemy.types import PickleType, TypeDecorator, VARCHAR
from sqlalchemy.orm import mapper, Session, composite
from sqlalchemy.orm.mapper import Mapper
from sqlalchemy.orm.instrumentation import ClassManager
from test.lib.schema import Table, Column
from test.lib.testing import eq_
from test.lib.util import picklers
from test.lib import testing
from test.orm import _base
import sys
import pickle

class Foo(_base.BasicEntity):
    pass

class _MutableDictTestBase(object):
    run_define_tables = 'each'

    @classmethod
    def _type_fixture(cls):
        from sqlalchemy.ext.mutable import Mutable

        # needed for pickle support
        global MutationDict

        class MutationDict(Mutable, dict):
            @classmethod
            def coerce(cls, key, value):
                if not isinstance(value, MutationDict):
                    if isinstance(value, dict):
                        return MutationDict(value)
                    return Mutable.coerce(key, value)
                else:
                    return value

            def __getstate__(self):
                return dict(self)

            def __setstate__(self, state):
                self.update(state)

            def __setitem__(self, key, value):
                dict.__setitem__(self, key, value)
                self.changed()

            def __delitem__(self, key):
                dict.__delitem__(self, key)
                self.changed()
        return MutationDict

    @testing.resolve_artifact_names
    def setup_mappers(cls):
        mapper(Foo, foo)

    def teardown(self):
        # clear out mapper events
        Mapper.dispatch._clear()
        ClassManager.dispatch._clear()
        super(_MutableDictTestBase, self).teardown()

    @testing.resolve_artifact_names
    def test_in_place_mutation(self):
        sess = Session()

        f1 = Foo(data={'a':'b'})
        sess.add(f1)
        sess.commit()

        f1.data['a'] = 'c'
        sess.commit()

        eq_(f1.data, {'a':'c'})

    @testing.resolve_artifact_names
    def test_pickle_parent(self):
        sess = Session()

        f1 = Foo(data={'a':'b'})
        sess.add(f1)
        sess.commit()
        f1.data
        sess.close()

        for loads, dumps in picklers():
            sess = Session()
            f2 = loads(dumps(f1))
            sess.add(f2)
            f2.data['a'] = 'c'
            assert f2 in sess.dirty

    @testing.resolve_artifact_names
    def _test_non_mutable(self):
        sess = Session()

        f1 = Foo(non_mutable_data={'a':'b'})
        sess.add(f1)
        sess.commit()

        f1.non_mutable_data['a'] = 'c'
        sess.commit()

        eq_(f1.non_mutable_data, {'a':'b'})

class MutableWithScalarPickleTest(_MutableDictTestBase, _base.MappedTest):
    @classmethod
    def define_tables(cls, metadata):
        MutationDict = cls._type_fixture()

        Table('foo', metadata,
            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
            Column('data', MutationDict.as_mutable(PickleType)),
            Column('non_mutable_data', PickleType)
        )

    def test_non_mutable(self):
        self._test_non_mutable()

class MutableWithScalarJSONTest(_MutableDictTestBase, _base.MappedTest):
    # json introduced in 2.6
    __skip_if__ = lambda : sys.version_info < (2, 6),

    @classmethod
    def define_tables(cls, metadata):
        import json

        class JSONEncodedDict(TypeDecorator):
            impl = VARCHAR(50)

            def process_bind_param(self, value, dialect):
                if value is not None:
                    value = json.dumps(value)

                return value

            def process_result_value(self, value, dialect):
                if value is not None:
                    value = json.loads(value)
                return value

        MutationDict = cls._type_fixture()

        Table('foo', metadata,
            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
            Column('data', MutationDict.as_mutable(JSONEncodedDict)),
            Column('non_mutable_data', JSONEncodedDict)
        )

    def test_non_mutable(self):
        self._test_non_mutable()

class MutableAssociationScalarPickleTest(_MutableDictTestBase, _base.MappedTest):
    @classmethod
    def define_tables(cls, metadata):
        MutationDict = cls._type_fixture()
        MutationDict.associate_with(PickleType)

        Table('foo', metadata,
            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
            Column('data', PickleType)
        )

class MutableAssociationScalarJSONTest(_MutableDictTestBase, _base.MappedTest):
    # json introduced in 2.6
    __skip_if__ = lambda : sys.version_info < (2, 6),

    @classmethod
    def define_tables(cls, metadata):
        import json

        class JSONEncodedDict(TypeDecorator):
            impl = VARCHAR(50)

            def process_bind_param(self, value, dialect):
                if value is not None:
                    value = json.dumps(value)

                return value

            def process_result_value(self, value, dialect):
                if value is not None:
                    value = json.loads(value)
                return value

        MutationDict = cls._type_fixture()
        MutationDict.associate_with(JSONEncodedDict)

        Table('foo', metadata,
            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
            Column('data', JSONEncodedDict)
        )


class _CompositeTestBase(object):
    @classmethod
    def define_tables(cls, metadata):
        Table('foo', metadata,
            Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
            Column('x', Integer),
            Column('y', Integer)
        )

    def teardown(self):
        # clear out mapper events
        Mapper.dispatch._clear()
        ClassManager.dispatch._clear()
        super(_CompositeTestBase, self).teardown()

class MutableCompositesTest(_CompositeTestBase, _base.MappedTest):

    @classmethod
    def _type_fixture(cls):

        from sqlalchemy.ext.mutable import Mutable
        from sqlalchemy.ext.mutable import MutableComposite

        global Point

        class Point(MutableComposite):
            def __init__(self, x, y):
                self.x = x
                self.y = y

            def __setattr__(self, key, value):
                object.__setattr__(self, key, value)
                self.changed()

            def __composite_values__(self):
                return self.x, self.y

            def __getstate__(self):
                d = dict(self.__dict__)
                d.pop('_parents', None)
                return d

            #def __setstate__(self, state):
            #    self.x, self.y = state

            def __eq__(self, other):
                return isinstance(other, Point) and \
                    other.x == self.x and \
                    other.y == self.y
        return Point

    @classmethod
    @testing.resolve_artifact_names
    def setup_mappers(cls):
        Point = cls._type_fixture()

        mapper(Foo, foo, properties={
            'data':composite(Point, foo.c.x, foo.c.y)
        })

    @testing.resolve_artifact_names
    def test_in_place_mutation(self):
        sess = Session()
        d = Point(3, 4)
        f1 = Foo(data=d)
        sess.add(f1)
        sess.commit()

        f1.data.y = 5
        sess.commit()

        eq_(f1.data, Point(3, 5))

    @testing.resolve_artifact_names
    def test_pickle_of_parent(self):
        sess = Session()
        d = Point(3, 4)
        f1 = Foo(data=d)
        sess.add(f1)
        sess.commit()

        f1.data
        assert 'data' in f1.__dict__
        sess.close()

        for loads, dumps in picklers():
            sess = Session()
            f2 = loads(dumps(f1))
            sess.add(f2)
            f2.data.y = 12
            assert f2 in sess.dirty