1
from sqlalchemy import Integer, ForeignKey
2
from sqlalchemy.types import PickleType, TypeDecorator, VARCHAR
3
from sqlalchemy.orm import mapper, Session, composite
4
from sqlalchemy.orm.mapper import Mapper
5
from sqlalchemy.orm.instrumentation import ClassManager
6
from test.lib.schema import Table, Column
7
from test.lib.testing import eq_, assert_raises_message
8
from test.lib.util import picklers
9
from test.lib import testing
10
from test.lib import fixtures
14
class Foo(fixtures.BasicEntity):
20
class _MutableDictTestBase(object):
21
run_define_tables = 'each'
24
def _type_fixture(cls):
25
from sqlalchemy.ext.mutable import Mutable
27
# needed for pickle support
30
class MutationDict(Mutable, dict):
32
def coerce(cls, key, value):
33
if not isinstance(value, MutationDict):
34
if isinstance(value, dict):
35
return MutationDict(value)
36
return Mutable.coerce(key, value)
40
def __getstate__(self):
43
def __setstate__(self, state):
46
def __setitem__(self, key, value):
47
dict.__setitem__(self, key, value)
50
def __delitem__(self, key):
51
dict.__delitem__(self, key)
55
def setup_mappers(cls):
61
# clear out mapper events
62
Mapper.dispatch._clear()
63
ClassManager.dispatch._clear()
64
super(_MutableDictTestBase, self).teardown()
66
def test_coerce_none(self):
73
def test_coerce_raise(self):
74
assert_raises_message(
76
"Attribute 'data' does not accept objects of type",
77
Foo, data=set([1,2,3])
80
def test_in_place_mutation(self):
83
f1 = Foo(data={'a':'b'})
90
eq_(f1.data, {'a':'c'})
92
def test_replace(self):
94
f1 = Foo(data={'a':'b'})
100
eq_(f1.data, {'b':'c'})
102
def test_pickle_parent(self):
105
f1 = Foo(data={'a':'b'})
111
for loads, dumps in picklers():
113
f2 = loads(dumps(f1))
116
assert f2 in sess.dirty
118
def _test_non_mutable(self):
121
f1 = Foo(non_mutable_data={'a':'b'})
125
f1.non_mutable_data['a'] = 'c'
128
eq_(f1.non_mutable_data, {'a':'b'})
130
class MutableWithScalarPickleTest(_MutableDictTestBase, fixtures.MappedTest):
132
def define_tables(cls, metadata):
133
MutationDict = cls._type_fixture()
135
mutable_pickle = MutationDict.as_mutable(PickleType)
136
Table('foo', metadata,
137
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
138
Column('skip', mutable_pickle),
139
Column('data', mutable_pickle),
140
Column('non_mutable_data', PickleType)
143
def test_non_mutable(self):
144
self._test_non_mutable()
146
class MutableWithScalarJSONTest(_MutableDictTestBase, fixtures.MappedTest):
147
# json introduced in 2.6
148
__skip_if__ = lambda : sys.version_info < (2, 6),
151
def define_tables(cls, metadata):
154
class JSONEncodedDict(TypeDecorator):
157
def process_bind_param(self, value, dialect):
158
if value is not None:
159
value = json.dumps(value)
163
def process_result_value(self, value, dialect):
164
if value is not None:
165
value = json.loads(value)
168
MutationDict = cls._type_fixture()
170
Table('foo', metadata,
171
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
172
Column('data', MutationDict.as_mutable(JSONEncodedDict)),
173
Column('non_mutable_data', JSONEncodedDict)
176
def test_non_mutable(self):
177
self._test_non_mutable()
179
class MutableAssocWithAttrInheritTest(_MutableDictTestBase, fixtures.MappedTest):
181
def define_tables(cls, metadata):
182
MutationDict = cls._type_fixture()
184
Table('foo', metadata,
185
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
186
Column('data', PickleType),
187
Column('non_mutable_data', PickleType)
190
Table('subfoo', metadata,
191
Column('id', Integer, ForeignKey('foo.id'), primary_key=True),
194
def setup_mappers(cls):
196
subfoo = cls.tables.subfoo
199
mapper(SubFoo, subfoo, inherits=Foo)
200
MutationDict.associate_with_attribute(Foo.data)
202
def test_in_place_mutation(self):
205
f1 = SubFoo(data={'a':'b'})
212
eq_(f1.data, {'a':'c'})
214
def test_replace(self):
216
f1 = SubFoo(data={'a':'b'})
222
eq_(f1.data, {'b':'c'})
224
class MutableAssociationScalarPickleTest(_MutableDictTestBase, fixtures.MappedTest):
226
def define_tables(cls, metadata):
227
MutationDict = cls._type_fixture()
228
MutationDict.associate_with(PickleType)
230
Table('foo', metadata,
231
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
232
Column('skip', PickleType),
233
Column('data', PickleType)
236
class MutableAssociationScalarJSONTest(_MutableDictTestBase, fixtures.MappedTest):
237
# json introduced in 2.6
238
__skip_if__ = lambda : sys.version_info < (2, 6),
241
def define_tables(cls, metadata):
244
class JSONEncodedDict(TypeDecorator):
247
def process_bind_param(self, value, dialect):
248
if value is not None:
249
value = json.dumps(value)
253
def process_result_value(self, value, dialect):
254
if value is not None:
255
value = json.loads(value)
258
MutationDict = cls._type_fixture()
259
MutationDict.associate_with(JSONEncodedDict)
261
Table('foo', metadata,
262
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
263
Column('data', JSONEncodedDict)
267
class _CompositeTestBase(object):
269
def define_tables(cls, metadata):
270
Table('foo', metadata,
271
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
272
Column('x', Integer),
277
# clear out mapper events
278
Mapper.dispatch._clear()
279
ClassManager.dispatch._clear()
280
super(_CompositeTestBase, self).teardown()
283
def _type_fixture(cls):
285
from sqlalchemy.ext.mutable import Mutable
286
from sqlalchemy.ext.mutable import MutableComposite
290
class Point(MutableComposite):
291
def __init__(self, x, y):
295
def __setattr__(self, key, value):
296
object.__setattr__(self, key, value)
299
def __composite_values__(self):
300
return self.x, self.y
302
def __getstate__(self):
303
d = dict(self.__dict__)
304
d.pop('_parents', None)
307
#def __setstate__(self, state):
308
# self.x, self.y = state
310
def __eq__(self, other):
311
return isinstance(other, Point) and \
312
other.x == self.x and \
316
class MutableCompositesTest(_CompositeTestBase, fixtures.MappedTest):
319
def setup_mappers(cls):
322
Point = cls._type_fixture()
324
mapper(Foo, foo, properties={
325
'data':composite(Point, foo.c.x, foo.c.y)
328
def test_in_place_mutation(self):
338
eq_(f1.data, Point(3, 5))
340
def test_pickle_of_parent(self):
348
assert 'data' in f1.__dict__
351
for loads, dumps in picklers():
353
f2 = loads(dumps(f1))
356
assert f2 in sess.dirty
358
def test_set_none(self):
363
eq_(f1.data, Point(None, None))
367
eq_(f1.data, Point(None, 5))
369
def test_set_illegal(self):
371
assert_raises_message(
373
"Attribute 'data' does not accept objects",
374
setattr, f1, 'data', 'foo'
377
class MutableInheritedCompositesTest(_CompositeTestBase, fixtures.MappedTest):
379
def define_tables(cls, metadata):
380
Table('foo', metadata,
381
Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
382
Column('x', Integer),
385
Table('subfoo', metadata,
386
Column('id', Integer, ForeignKey('foo.id'), primary_key=True),
390
def setup_mappers(cls):
392
subfoo = cls.tables.subfoo
394
Point = cls._type_fixture()
396
mapper(Foo, foo, properties={
397
'data':composite(Point, foo.c.x, foo.c.y)
399
mapper(SubFoo, subfoo, inherits=Foo)
401
def test_in_place_mutation_subclass(self):
411
eq_(f1.data, Point(3, 5))
413
def test_pickle_of_parent_subclass(self):
421
assert 'data' in f1.__dict__
424
for loads, dumps in picklers():
426
f2 = loads(dumps(f1))
429
assert f2 in sess.dirty
added
removed
test/ext/test_mutable.py
