1
"""Test the interaction of :class:`.MutableType` as well as the
2
``mutable=True`` flag with the ORM.
4
For new mutablity functionality, see test.ext.test_mutable.
7
from test.lib.testing import eq_
9
from sqlalchemy.orm import mapper as orm_mapper
11
import sqlalchemy as sa
12
from sqlalchemy import Integer, String, ForeignKey
13
from test.lib import testing, pickleable
14
from test.lib.schema import Table, Column
15
from sqlalchemy.orm import mapper, create_session, Session, attributes
16
from test.lib.testing import eq_, ne_
17
from test.lib.util import gc_collect
18
from test.lib import fixtures
19
from test.orm import _fixtures
21
class MutableTypesTest(fixtures.MappedTest):
24
def define_tables(cls, metadata):
25
Table('mutable_t', metadata,
26
Column('id', Integer, primary_key=True,
27
test_needs_autoincrement=True),
28
Column('data', sa.PickleType(mutable=True)),
29
Column('val', sa.Unicode(30)))
32
def setup_classes(cls):
37
def setup_mappers(cls):
38
mutable_t, Foo = cls.tables.mutable_t, cls.classes.Foo
40
mapper(Foo, mutable_t)
42
def test_modified_status(self):
43
Foo = self.classes.Foo
45
f1 = Foo(data = pickleable.Bar(4,5))
51
f2 = session.query(Foo).first()
52
assert 'data' in sa.orm.attributes.instance_state(f2).unmodified
56
assert f2 in session.dirty
57
assert 'data' not in sa.orm.attributes.instance_state(f2).unmodified
59
def test_mutations_persisted(self):
60
Foo = self.classes.Foo
62
f1 = Foo(data = pickleable.Bar(4,5))
70
f2 = session.query(Foo).first()
76
f3 = session.query(Foo).first()
78
eq_(f3.data, pickleable.Bar(4, 19))
80
def test_no_unnecessary_update(self):
81
Foo = self.classes.Foo
83
f1 = Foo(data = pickleable.Bar(4,5), val = u'hi')
89
self.sql_count_(0, session.commit)
91
f1.val = u'someothervalue'
92
self.assert_sql(testing.db, session.commit, [
93
("UPDATE mutable_t SET val=:val "
94
"WHERE mutable_t.id = :mutable_t_id",
95
{'mutable_t_id': f1.id, 'val': u'someothervalue'})])
99
self.assert_sql(testing.db, session.commit, [
100
("UPDATE mutable_t SET data=:data, val=:val "
101
"WHERE mutable_t.id = :mutable_t_id",
102
{'mutable_t_id': f1.id, 'val': u'hi', 'data':f1.data})])
104
def test_mutated_state_resurrected(self):
105
Foo = self.classes.Foo
107
f1 = Foo(data = pickleable.Bar(4,5), val = u'hi')
117
assert len(session.identity_map) == 1
121
assert session.query(Foo).one().data == pickleable.Bar(4, 19)
123
def test_mutated_plus_scalar_state_change_resurrected(self):
124
"""test that a non-mutable attribute event subsequent to
125
a mutable event prevents the object from falling into
130
Foo = self.classes.Foo
132
f1 = Foo(data = pickleable.Bar(4, 5), val=u'some val')
138
f1.val=u'some new val'
140
assert sa.orm.attributes.instance_state(f1)._strong_obj is not None
145
session.query(Foo.val).all(),
149
def test_non_mutated_state_not_resurrected(self):
150
Foo = self.classes.Foo
152
f1 = Foo(data = pickleable.Bar(4,5))
159
f1 = session.query(Foo).first()
163
assert len(session.identity_map) == 0
164
f1 = session.query(Foo).first()
165
assert not attributes.instance_state(f1).modified
167
def test_scalar_no_net_change_no_update(self):
168
"""Test that a no-net-change on a scalar attribute event
169
doesn't cause an UPDATE for a mutable state.
173
Foo = self.classes.Foo
183
f1 = session.query(Foo).first()
185
self.sql_count_(0, session.commit)
187
def test_expire_attribute_set(self):
188
"""test no SELECT emitted when assigning to an expired
193
Foo = self.classes.Foo
196
f1 = Foo(data = pickleable.Bar(4, 5), val=u'some val')
201
assert 'data' not in f1.__dict__
203
f1.data = pickleable.Bar(10, 15)
204
self.sql_count_(0, go)
209
def test_expire_mutate(self):
210
"""test mutations are detected on an expired mutable
213
Foo = self.classes.Foo
216
f1 = Foo(data = pickleable.Bar(4, 5), val=u'some val')
221
assert 'data' not in f1.__dict__
224
self.sql_count_(1, go)
229
def test_deferred_attribute_set(self):
230
"""test no SELECT emitted when assigning to a deferred
235
mutable_t, Foo = self.tables.mutable_t, self.classes.Foo
237
sa.orm.clear_mappers()
238
mapper(Foo, mutable_t, properties={
239
'data':sa.orm.deferred(mutable_t.c.data)
242
f1 = Foo(data = pickleable.Bar(4, 5), val=u'some val')
249
f1 = session.query(Foo).first()
251
f1.data = pickleable.Bar(10, 15)
252
self.sql_count_(0, go)
257
def test_deferred_mutate(self):
258
"""test mutations are detected on a deferred mutable
261
mutable_t, Foo = self.tables.mutable_t, self.classes.Foo
264
sa.orm.clear_mappers()
265
mapper(Foo, mutable_t, properties={
266
'data':sa.orm.deferred(mutable_t.c.data)
269
f1 = Foo(data = pickleable.Bar(4, 5), val=u'some val')
276
f1 = session.query(Foo).first()
279
self.sql_count_(1, go)
284
self.sql_count_(1, go)
287
class PickledDictsTest(fixtures.MappedTest):
290
def define_tables(cls, metadata):
291
Table('mutable_t', metadata,
292
Column('id', Integer, primary_key=True,
293
test_needs_autoincrement=True),
295
sa.PickleType(comparator=operator.eq, mutable=True)))
298
def setup_classes(cls):
299
class Foo(cls.Basic):
303
def setup_mappers(cls):
304
mutable_t, Foo = cls.tables.mutable_t, cls.classes.Foo
306
mapper(Foo, mutable_t)
308
def test_dicts(self):
309
"""Dictionaries may not pickle the same way twice."""
311
Foo = self.classes.Foo
316
'personne': {'nom': u'Smith',
331
session = create_session(autocommit=False)
335
self.sql_count_(0, session.commit)
338
'personne': {'nom': u'Smith',
353
self.sql_count_(0, session.commit)
355
f1.data[0]['personne']['VenSoir']= False
356
self.sql_count_(1, session.commit)
358
session.expunge_all()
359
f = session.query(Foo).get(f1.id)
362
'personne': {'nom': u'Smith',
375
'SamAcc': False} } ])
added
removed
test/orm/test_legacy_mutable.py
