← Back to branch summary

~ubuntu-branches/debian/jessie/sqlalchemy/jessie

« back to all changes in this revision

Viewing changes to test/orm/test_dynamic.py

  • Committer: Package Import Robot
  • Author(s): Piotr Ożarowski, Jakub Wilk, Piotr Ożarowski
  • Date: 2013-07-06 20:53:52 UTC
  • mfrom: (1.4.23) (16.1.17 experimental)
  • Revision ID: package-import@ubuntu.com-20130706205352-ryppl1eto3illd79
Tags: 0.8.2-1
[ Jakub Wilk ]
* Use canonical URIs for Vcs-* fields.

[ Piotr Ożarowski ]
* New upstream release
* Upload to unstable
* Build depend on python3-all instead of -dev, extensions are not built for
  Python 3.X 

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/orm/test_dynamic.py
1
 
from test.lib.testing import eq_, ne_
2
 
import operator
3
 
from sqlalchemy.orm import dynamic_loader, backref, configure_mappers
4
 
from test.lib import testing
5
 
from sqlalchemy import Integer, String, ForeignKey, desc, select, func, exc
6
 
from test.lib.schema import Table, Column
7
 
from sqlalchemy.orm import mapper, relationship, create_session, Query, attributes
 
1
from sqlalchemy.testing import eq_, is_
 
2
from sqlalchemy.orm import backref, configure_mappers
 
3
from sqlalchemy import testing
 
4
from sqlalchemy import desc, select, func, exc
 
5
from sqlalchemy.orm import mapper, relationship, create_session, Query, \
 
6
                    attributes, exc as orm_exc, Session
8
7
from sqlalchemy.orm.dynamic import AppenderMixin
9
 
from test.lib.testing import eq_, AssertsCompiledSQL, assert_raises_message, assert_raises
10
 
from test.lib import fixtures
 
8
from sqlalchemy.testing import AssertsCompiledSQL, \
 
9
        assert_raises_message, assert_raises
11
10
from test.orm import _fixtures
12
11
 
13
 
 
14
 
class DynamicTest(_fixtures.FixtureTest, AssertsCompiledSQL):
 
12
from sqlalchemy.testing.assertsql import CompiledSQL
 
13
 
 
14
 
 
15
 
 
16
class _DynamicFixture(object):
 
17
    def _user_address_fixture(self, addresses_args={}):
 
18
        users, Address, addresses, User = (self.tables.users,
 
19
                                self.classes.Address,
 
20
                                self.tables.addresses,
 
21
                                self.classes.User)
 
22
 
 
23
        mapper(User, users, properties={
 
24
            'addresses': relationship(Address, lazy="dynamic",
 
25
                                        **addresses_args)
 
26
        })
 
27
        mapper(Address, addresses)
 
28
        return User, Address
 
29
 
 
30
    def _order_item_fixture(self, items_args={}):
 
31
        items, Order, orders, order_items, Item = (self.tables.items,
 
32
                                self.classes.Order,
 
33
                                self.tables.orders,
 
34
                                self.tables.order_items,
 
35
                                self.classes.Item)
 
36
 
 
37
        mapper(Order, orders, properties={
 
38
            'items': relationship(Item,
 
39
                            secondary=order_items,
 
40
                            lazy="dynamic",
 
41
                            **items_args
 
42
                        )
 
43
        })
 
44
        mapper(Item, items)
 
45
        return Order, Item
 
46
 
 
47
class DynamicTest(_DynamicFixture, _fixtures.FixtureTest, AssertsCompiledSQL):
 
48
 
15
49
    def test_basic(self):
16
 
        users, Address, addresses, User = (self.tables.users,
17
 
                                self.classes.Address,
18
 
                                self.tables.addresses,
19
 
                                self.classes.User)
20
 
 
21
 
        mapper(User, users, properties={
22
 
            'addresses':dynamic_loader(mapper(Address, addresses))
23
 
        })
 
50
        User, Address = self._user_address_fixture()
24
51
        sess = create_session()
25
52
        q = sess.query(User)
26
53
 
27
 
        u = q.filter(User.id==7).first()
28
 
 
29
54
        eq_([User(id=7,
30
55
                  addresses=[Address(id=1, email_address='jack@bean.com')])],
31
 
            q.filter(User.id==7).all())
 
56
            q.filter(User.id == 7).all())
32
57
        eq_(self.static.user_address_result, q.all())
33
58
 
34
59
    def test_statement(self):
35
60
        """test that the .statement accessor returns the actual statement that
36
61
        would render, without any _clones called."""
37
62
 
38
 
        users, Address, addresses, User = (self.tables.users,
39
 
                                self.classes.Address,
40
 
                                self.tables.addresses,
41
 
                                self.classes.User)
42
 
 
43
 
 
44
 
        mapper(User, users, properties={
45
 
            'addresses':dynamic_loader(mapper(Address, addresses))
46
 
        })
 
63
        User, Address = self._user_address_fixture()
47
64
        sess = create_session()
48
65
        q = sess.query(User)
49
66
 
50
 
        u = q.filter(User.id==7).first()
 
67
        u = q.filter(User.id == 7).first()
51
68
        self.assert_compile(
52
69
            u.addresses.statement,
53
 
            "SELECT addresses.id, addresses.user_id, addresses.email_address FROM "
 
70
            "SELECT addresses.id, addresses.user_id, addresses.email_address "
 
71
            "FROM "
54
72
            "addresses WHERE :param_1 = addresses.user_id",
55
73
            use_default_dialect=True
56
74
        )
57
75
 
 
76
    def test_detached_raise(self):
 
77
        User, Address = self._user_address_fixture()
 
78
        sess = create_session()
 
79
        u = sess.query(User).get(8)
 
80
        sess.expunge(u)
 
81
        assert_raises(
 
82
            orm_exc.DetachedInstanceError,
 
83
            u.addresses.filter_by,
 
84
            email_address='e'
 
85
        )
 
86
 
58
87
    def test_no_uselist_false(self):
59
 
        users, Address, addresses, User = (self.tables.users,
60
 
                                self.classes.Address,
61
 
                                self.tables.addresses,
62
 
                                self.classes.User)
63
 
        mapper(Address, addresses)
64
 
        mapper(User, users, properties={
65
 
                "addresses": relationship(Address, lazy='dynamic', uselist=False)
66
 
            })
 
88
        User, Address = self._user_address_fixture(
 
89
                                    addresses_args={"uselist": False})
67
90
        assert_raises_message(
68
 
            exc.SAWarning,
 
91
            exc.InvalidRequestError,
69
92
            "On relationship User.addresses, 'dynamic' loaders cannot be "
70
93
            "used with many-to-one/one-to-one relationships and/or "
71
94
            "uselist=False.",
82
105
            })
83
106
        mapper(User, users)
84
107
        assert_raises_message(
85
 
            exc.SAWarning,
 
108
            exc.InvalidRequestError,
86
109
            "On relationship Address.user, 'dynamic' loaders cannot be "
87
110
            "used with many-to-one/one-to-one relationships and/or "
88
111
            "uselist=False.",
90
113
        )
91
114
 
92
115
    def test_order_by(self):
93
 
        users, Address, addresses, User = (self.tables.users,
94
 
                                self.classes.Address,
95
 
                                self.tables.addresses,
96
 
                                self.classes.User)
97
 
 
98
 
        mapper(User, users, properties={
99
 
            'addresses': dynamic_loader(mapper(Address, addresses))
100
 
        })
 
116
        User, Address = self._user_address_fixture()
101
117
        sess = create_session()
102
118
        u = sess.query(User).get(8)
103
119
        eq_(
104
120
            list(u.addresses.order_by(desc(Address.email_address))),
105
 
             [Address(email_address=u'ed@wood.com'), Address(email_address=u'ed@lala.com'),
106
 
              Address(email_address=u'ed@bettyboop.com')]
107
 
            )
 
121
             [
 
122
                Address(email_address=u'ed@wood.com'),
 
123
                Address(email_address=u'ed@lala.com'),
 
124
                Address(email_address=u'ed@bettyboop.com')
 
125
            ]
 
126
        )
108
127
 
109
128
    def test_configured_order_by(self):
110
 
        users, Address, addresses, User = (self.tables.users,
111
 
                                self.classes.Address,
112
 
                                self.tables.addresses,
113
 
                                self.classes.User)
 
129
        addresses = self.tables.addresses
 
130
        User, Address = self._user_address_fixture(
 
131
                                    addresses_args={
 
132
                                        "order_by":
 
133
                                            addresses.c.email_address.desc()})
114
134
 
115
 
        mapper(User, users, properties={
116
 
            'addresses':dynamic_loader(mapper(Address, addresses), order_by=desc(Address.email_address))
117
 
        })
118
135
        sess = create_session()
119
136
        u = sess.query(User).get(8)
120
 
        eq_(list(u.addresses), [Address(email_address=u'ed@wood.com'), Address(email_address=u'ed@lala.com'), Address(email_address=u'ed@bettyboop.com')])
 
137
        eq_(
 
138
            list(u.addresses),
 
139
            [
 
140
                Address(email_address=u'ed@wood.com'),
 
141
                Address(email_address=u'ed@lala.com'),
 
142
                Address(email_address=u'ed@bettyboop.com')
 
143
            ]
 
144
        )
121
145
 
122
146
        # test cancellation of None, replacement with something else
123
147
        eq_(
124
148
            list(u.addresses.order_by(None).order_by(Address.email_address)),
125
 
            [Address(email_address=u'ed@bettyboop.com'), Address(email_address=u'ed@lala.com'), Address(email_address=u'ed@wood.com')]
 
149
            [
 
150
                Address(email_address=u'ed@bettyboop.com'),
 
151
                Address(email_address=u'ed@lala.com'),
 
152
                Address(email_address=u'ed@wood.com')
 
153
            ]
126
154
        )
127
155
 
128
156
        # test cancellation of None, replacement with nothing
129
157
        eq_(
130
158
            set(u.addresses.order_by(None)),
131
 
            set([Address(email_address=u'ed@bettyboop.com'), Address(email_address=u'ed@lala.com'), Address(email_address=u'ed@wood.com')])
 
159
            set([
 
160
                Address(email_address=u'ed@bettyboop.com'),
 
161
                Address(email_address=u'ed@lala.com'),
 
162
                Address(email_address=u'ed@wood.com')
 
163
            ])
132
164
        )
133
165
 
134
166
    def test_count(self):
135
 
        users, Address, addresses, User = (self.tables.users,
136
 
                                self.classes.Address,
137
 
                                self.tables.addresses,
138
 
                                self.classes.User)
139
 
 
140
 
        mapper(User, users, properties={
141
 
            'addresses':dynamic_loader(mapper(Address, addresses))
142
 
        })
 
167
        User, Address = self._user_address_fixture()
143
168
        sess = create_session()
144
169
        u = sess.query(User).first()
145
170
        eq_(u.addresses.count(), 1)
146
171
 
147
 
    def test_backref(self):
 
172
    def test_dynamic_on_backref(self):
148
173
        users, Address, addresses, User = (self.tables.users,
149
174
                                self.classes.Address,
150
175
                                self.tables.addresses,
151
176
                                self.classes.User)
152
177
 
153
178
        mapper(Address, addresses, properties={
154
 
            'user':relationship(User, backref=backref('addresses', lazy='dynamic'))
 
179
            'user': relationship(User,
 
180
                                backref=backref('addresses', lazy='dynamic'))
155
181
        })
156
182
        mapper(User, users)
157
183
 
165
191
        assert ad not in u.addresses
166
192
 
167
193
    def test_no_count(self):
168
 
        users, Address, addresses, User = (self.tables.users,
169
 
                                self.classes.Address,
170
 
                                self.tables.addresses,
171
 
                                self.classes.User)
172
 
 
173
 
        mapper(User, users, properties={
174
 
            'addresses':dynamic_loader(mapper(Address, addresses))
175
 
        })
 
194
        User, Address = self._user_address_fixture()
176
195
        sess = create_session()
177
196
        q = sess.query(User)
178
197
 
180
199
        # returns a live database result), else additional count() queries are
181
200
        # issued when evaluating in a list context
182
201
        def go():
183
 
            eq_([User(id=7,
184
 
                      addresses=[Address(id=1,
185
 
                                         email_address='jack@bean.com')])],
186
 
                q.filter(User.id==7).all())
 
202
            eq_(
 
203
                q.filter(User.id == 7).all(),
 
204
                [
 
205
                    User(id=7,
 
206
                      addresses=[
 
207
                        Address(id=1, email_address='jack@bean.com')
 
208
                    ])
 
209
                ]
 
210
            )
187
211
        self.assert_sql_count(testing.db, go, 2)
188
212
 
189
213
    def test_no_populate(self):
190
 
        users, Address, addresses, User = (self.tables.users,
191
 
                                self.classes.Address,
192
 
                                self.tables.addresses,
193
 
                                self.classes.User)
194
 
 
195
 
        mapper(User, users, properties={
196
 
            'addresses':dynamic_loader(mapper(Address, addresses))
197
 
        })
 
214
        User, Address = self._user_address_fixture()
198
215
        u1 = User()
199
216
        assert_raises_message(
200
217
            NotImplementedError,
203
220
        )
204
221
 
205
222
    def test_m2m(self):
206
 
        items, Order, orders, order_items, Item = (self.tables.items,
207
 
                                self.classes.Order,
208
 
                                self.tables.orders,
209
 
                                self.tables.order_items,
210
 
                                self.classes.Item)
211
 
 
212
 
        mapper(Order, orders, properties={
213
 
            'items':relationship(Item, secondary=order_items, lazy="dynamic",
214
 
                             backref=backref('orders', lazy="dynamic"))
215
 
        })
216
 
        mapper(Item, items)
 
223
        Order, Item = self._order_item_fixture(items_args={
 
224
                "backref": backref("orders", lazy="dynamic")
 
225
            })
217
226
 
218
227
        sess = create_session()
219
228
        o1 = Order(id=15, description="order 10")
226
235
        assert i1 in o1.items.all()
227
236
 
228
237
    @testing.exclude('mysql', 'between',
229
 
            ((5, 1,49), (5, 1, 52)),
 
238
            ((5, 1, 49), (5, 1, 52)),
230
239
            'https://bugs.launchpad.net/ubuntu/+source/mysql-5.1/+bug/706988')
231
240
    def test_association_nonaliased(self):
232
241
        items, Order, orders, order_items, Item = (self.tables.items,
236
245
                                self.classes.Item)
237
246
 
238
247
        mapper(Order, orders, properties={
239
 
            'items':relationship(Item, secondary=order_items,
 
248
            'items': relationship(Item,
 
249
                                secondary=order_items,
240
250
                                lazy="dynamic",
241
251
                                order_by=order_items.c.item_id)
242
252
        })
258
268
        # filter criterion against the secondary table
259
269
        # works
260
270
        eq_(
261
 
            o.items.filter(order_items.c.item_id==2).all(),
 
271
            o.items.filter(order_items.c.item_id == 2).all(),
262
272
            [Item(id=2)]
263
273
        )
264
274
 
265
 
 
266
 
    def test_transient_detached(self):
267
 
        users, Address, addresses, User = (self.tables.users,
268
 
                                self.classes.Address,
269
 
                                self.tables.addresses,
270
 
                                self.classes.User)
271
 
 
272
 
        mapper(User, users, properties={
273
 
            'addresses':dynamic_loader(mapper(Address, addresses))
274
 
        })
275
 
        sess = create_session()
 
275
    def test_transient_count(self):
 
276
        User, Address = self._user_address_fixture()
276
277
        u1 = User()
277
278
        u1.addresses.append(Address())
278
279
        eq_(u1.addresses.count(), 1)
 
280
 
 
281
    def test_transient_access(self):
 
282
        User, Address = self._user_address_fixture()
 
283
        u1 = User()
 
284
        u1.addresses.append(Address())
279
285
        eq_(u1.addresses[0], Address())
280
286
 
281
287
    def test_custom_query(self):
282
 
        users, Address, addresses, User = (self.tables.users,
283
 
                                self.classes.Address,
284
 
                                self.tables.addresses,
285
 
                                self.classes.User)
286
 
 
287
288
        class MyQuery(Query):
288
289
            pass
 
290
        User, Address = self._user_address_fixture(
 
291
                                addresses_args={"query_class": MyQuery})
289
292
 
290
 
        mapper(User, users, properties={
291
 
            'addresses':dynamic_loader(mapper(Address, addresses),
292
 
                                       query_class=MyQuery)
293
 
        })
294
293
        sess = create_session()
295
294
        u = User()
296
295
        sess.add(u)
308
307
        eq_(type(q).__name__, 'MyQuery')
309
308
 
310
309
    def test_custom_query_with_custom_mixin(self):
311
 
        users, Address, addresses, User = (self.tables.users,
312
 
                                self.classes.Address,
313
 
                                self.tables.addresses,
314
 
                                self.classes.User)
315
 
 
316
310
        class MyAppenderMixin(AppenderMixin):
317
311
            def add(self, items):
318
312
                if isinstance(items, list):
327
321
        class MyAppenderQuery(MyAppenderMixin, MyQuery):
328
322
            query_class = MyQuery
329
323
 
330
 
        mapper(User, users, properties={
331
 
            'addresses':dynamic_loader(mapper(Address, addresses),
332
 
                                       query_class=MyAppenderQuery)
333
 
        })
 
324
        User, Address = self._user_address_fixture(
 
325
                                addresses_args={
 
326
                                    "query_class": MyAppenderQuery})
 
327
 
 
328
 
334
329
        sess = create_session()
335
330
        u = User()
336
331
        sess.add(u)
350
345
        eq_(type(q).__name__, 'MyQuery')
351
346
 
352
347
 
353
 
class SessionTest(_fixtures.FixtureTest):
 
348
class UOWTest(_DynamicFixture, _fixtures.FixtureTest,
 
349
                    testing.AssertsExecutionResults):
354
350
    run_inserts = None
355
351
 
356
 
    def test_events(self):
357
 
        users, Address, addresses, User = (self.tables.users,
358
 
                                self.classes.Address,
359
 
                                self.tables.addresses,
360
 
                                self.classes.User)
 
352
    def test_persistence(self):
 
353
        addresses = self.tables.addresses
 
354
        User, Address = self._user_address_fixture()
361
355
 
362
 
        mapper(User, users, properties={
363
 
            'addresses':dynamic_loader(mapper(Address, addresses))
364
 
        })
365
356
        sess = create_session()
366
357
        u1 = User(name='jack')
367
358
        a1 = Address(email_address='foo')
368
359
        sess.add_all([u1, a1])
369
360
        sess.flush()
370
361
 
371
 
        eq_(testing.db.scalar(select([func.count(1)]).where(addresses.c.user_id!=None)), 0)
 
362
        eq_(
 
363
            testing.db.scalar(
 
364
                select([func.count(1)]).where(addresses.c.user_id != None)
 
365
            ),
 
366
            0
 
367
        )
372
368
        u1 = sess.query(User).get(u1.id)
373
369
        u1.addresses.append(a1)
374
370
        sess.flush()
375
371
 
376
 
        eq_(testing.db.execute(select([addresses]).where(addresses.c.user_id!=None)).fetchall(),
377
 
            [(a1.id, u1.id, 'foo')])
 
372
        eq_(
 
373
            testing.db.execute(
 
374
                    select([addresses]).where(addresses.c.user_id != None)
 
375
            ).fetchall(),
 
376
            [(a1.id, u1.id, 'foo')]
 
377
        )
378
378
 
379
379
        u1.addresses.remove(a1)
380
380
        sess.flush()
381
 
        eq_(testing.db.scalar(select([func.count(1)]).where(addresses.c.user_id!=None)), 0)
 
381
        eq_(
 
382
            testing.db.scalar(
 
383
                select([func.count(1)]).where(addresses.c.user_id != None)
 
384
            ),
 
385
            0
 
386
        )
382
387
 
383
388
        u1.addresses.append(a1)
384
389
        sess.flush()
385
 
        eq_(testing.db.execute(select([addresses]).where(addresses.c.user_id!=None)).fetchall(),
386
 
            [(a1.id, u1.id, 'foo')])
 
390
        eq_(
 
391
            testing.db.execute(
 
392
                select([addresses]).where(addresses.c.user_id != None)
 
393
            ).fetchall(),
 
394
            [(a1.id, u1.id, 'foo')]
 
395
        )
387
396
 
388
397
        a2 = Address(email_address='bar')
389
398
        u1.addresses.remove(a1)
390
399
        u1.addresses.append(a2)
391
400
        sess.flush()
392
 
        eq_(testing.db.execute(select([addresses]).where(addresses.c.user_id!=None)).fetchall(),
393
 
            [(a2.id, u1.id, 'bar')])
 
401
        eq_(
 
402
            testing.db.execute(
 
403
                select([addresses]).where(addresses.c.user_id != None)
 
404
            ).fetchall(),
 
405
            [(a2.id, u1.id, 'bar')]
 
406
        )
394
407
 
395
408
 
396
409
    def test_merge(self):
397
 
        users, Address, addresses, User = (self.tables.users,
398
 
                                self.classes.Address,
399
 
                                self.tables.addresses,
400
 
                                self.classes.User)
401
 
 
402
 
        mapper(User, users, properties={
403
 
            'addresses':dynamic_loader(mapper(Address, addresses), order_by=addresses.c.email_address)
404
 
        })
 
410
        addresses = self.tables.addresses
 
411
        User, Address = self._user_address_fixture(
 
412
                                    addresses_args={
 
413
                                        "order_by": addresses.c.email_address})
405
414
        sess = create_session()
406
415
        u1 = User(name='jack')
407
416
        a1 = Address(email_address='a1')
431
440
            [a1, a3]
432
441
        )
433
442
 
434
 
    def test_flush(self):
435
 
        users, Address, addresses, User = (self.tables.users,
436
 
                                self.classes.Address,
437
 
                                self.tables.addresses,
438
 
                                self.classes.User)
439
 
 
440
 
        mapper(User, users, properties={
441
 
            'addresses':dynamic_loader(mapper(Address, addresses))
442
 
        })
443
 
        sess = create_session()
444
 
        u1 = User(name='jack')
445
 
        u2 = User(name='ed')
446
 
        u2.addresses.append(Address(email_address='foo@bar.com'))
447
 
        u1.addresses.append(Address(email_address='lala@hoho.com'))
448
 
        sess.add_all((u1, u2))
449
 
        sess.flush()
450
 
 
451
 
        from sqlalchemy.orm import attributes
452
 
        eq_(attributes.get_history(u1, 'addresses'), ([], [Address(email_address='lala@hoho.com')], []))
453
 
 
454
 
        sess.expunge_all()
455
 
 
456
 
        # test the test fixture a little bit
457
 
        ne_(User(name='jack', addresses=[Address(email_address='wrong')]),
458
 
            sess.query(User).first())
459
 
        eq_(User(name='jack', addresses=[Address(email_address='lala@hoho.com')]),
460
 
            sess.query(User).first())
461
 
 
462
 
        eq_([
463
 
            User(name='jack', addresses=[Address(email_address='lala@hoho.com')]),
464
 
            User(name='ed', addresses=[Address(email_address='foo@bar.com')])
465
 
            ],
466
 
            sess.query(User).all())
467
 
 
468
443
    def test_hasattr(self):
469
 
        users, Address, addresses, User = (self.tables.users,
470
 
                                self.classes.Address,
471
 
                                self.tables.addresses,
472
 
                                self.classes.User)
 
444
        User, Address = self._user_address_fixture()
473
445
 
474
 
        mapper(User, users, properties={
475
 
            'addresses':dynamic_loader(mapper(Address, addresses))
476
 
        })
477
446
        u1 = User(name='jack')
478
447
 
479
 
        assert 'addresses' not in u1.__dict__.keys()
 
448
        assert 'addresses' not in u1.__dict__
480
449
        u1.addresses = [Address(email_address='test')]
481
 
        assert 'addresses' in dir(u1)
 
450
        assert 'addresses' in u1.__dict__
482
451
 
483
452
    def test_collection_set(self):
484
 
        users, Address, addresses, User = (self.tables.users,
485
 
                                self.classes.Address,
486
 
                                self.tables.addresses,
487
 
                                self.classes.User)
488
 
 
489
 
        mapper(User, users, properties={
490
 
            'addresses':dynamic_loader(mapper(Address, addresses), order_by=addresses.c.email_address)
491
 
        })
 
453
        addresses = self.tables.addresses
 
454
        User, Address = self._user_address_fixture(
 
455
                                    addresses_args={
 
456
                                        "order_by": addresses.c.email_address})
492
457
        sess = create_session(autoflush=True, autocommit=False)
493
458
        u1 = User(name='jack')
494
459
        a1 = Address(email_address='a1')
506
471
        u1.addresses = []
507
472
        eq_(list(u1.addresses), [])
508
473
 
 
474
    def test_noload_append(self):
 
475
        # test that a load of User.addresses is not emitted
 
476
        # when flushing an append
 
477
        User, Address = self._user_address_fixture()
 
478
 
 
479
        sess = Session()
 
480
        u1 = User(name="jack", addresses=[Address(email_address="a1")])
 
481
        sess.add(u1)
 
482
        sess.commit()
 
483
 
 
484
        u1_id = u1.id
 
485
        sess.expire_all()
 
486
 
 
487
        u1.addresses.append(Address(email_address='a2'))
 
488
 
 
489
        self.assert_sql_execution(
 
490
            testing.db,
 
491
            sess.flush,
 
492
            CompiledSQL(
 
493
                "SELECT users.id AS users_id, users.name AS users_name "
 
494
                "FROM users WHERE users.id = :param_1",
 
495
                lambda ctx: [{"param_1": u1_id}]),
 
496
            CompiledSQL(
 
497
                "INSERT INTO addresses (user_id, email_address) "
 
498
                "VALUES (:user_id, :email_address)",
 
499
                lambda ctx: [{'email_address': 'a2', 'user_id': u1_id}]
 
500
            )
 
501
        )
 
502
 
 
503
    def test_noload_remove(self):
 
504
        # test that a load of User.addresses is not emitted
 
505
        # when flushing a remove
 
506
        User, Address = self._user_address_fixture()
 
507
 
 
508
        sess = Session()
 
509
        u1 = User(name="jack", addresses=[Address(email_address="a1")])
 
510
        a2 = Address(email_address='a2')
 
511
        u1.addresses.append(a2)
 
512
        sess.add(u1)
 
513
        sess.commit()
 
514
 
 
515
        u1_id = u1.id
 
516
        a2_id = a2.id
 
517
        sess.expire_all()
 
518
 
 
519
        u1.addresses.remove(a2)
 
520
 
 
521
        self.assert_sql_execution(
 
522
            testing.db,
 
523
            sess.flush,
 
524
            CompiledSQL(
 
525
                "SELECT users.id AS users_id, users.name AS users_name "
 
526
                "FROM users WHERE users.id = :param_1",
 
527
                lambda ctx: [{"param_1": u1_id}]),
 
528
            CompiledSQL(
 
529
                "SELECT addresses.id AS addresses_id, addresses.email_address "
 
530
                "AS addresses_email_address FROM addresses "
 
531
                "WHERE addresses.id = :param_1",
 
532
                lambda ctx: [{u'param_1': a2_id}]
 
533
            ),
 
534
            CompiledSQL(
 
535
                "UPDATE addresses SET user_id=:user_id WHERE addresses.id = "
 
536
                ":addresses_id",
 
537
                lambda ctx: [{u'addresses_id': a2_id, 'user_id': None}]
 
538
            )
 
539
        )
 
540
 
509
541
    def test_rollback(self):
510
 
        users, Address, addresses, User = (self.tables.users,
511
 
                                self.classes.Address,
512
 
                                self.tables.addresses,
513
 
                                self.classes.User)
514
 
 
515
 
        mapper(User, users, properties={
516
 
            'addresses':dynamic_loader(mapper(Address, addresses))
517
 
        })
518
 
        sess = create_session(expire_on_commit=False, autocommit=False, autoflush=True)
 
542
        User, Address = self._user_address_fixture()
 
543
        sess = create_session(
 
544
                    expire_on_commit=False, autocommit=False, autoflush=True)
519
545
        u1 = User(name='jack')
520
546
        u1.addresses.append(Address(email_address='lala@hoho.com'))
521
547
        sess.add(u1)
522
548
        sess.flush()
523
549
        sess.commit()
524
550
        u1.addresses.append(Address(email_address='foo@bar.com'))
525
 
        eq_(u1.addresses.order_by(Address.id).all(),
526
 
                 [Address(email_address='lala@hoho.com'), Address(email_address='foo@bar.com')])
 
551
        eq_(
 
552
            u1.addresses.order_by(Address.id).all(),
 
553
            [
 
554
                Address(email_address='lala@hoho.com'),
 
555
                Address(email_address='foo@bar.com')
 
556
            ]
 
557
        )
527
558
        sess.rollback()
528
 
        eq_(u1.addresses.all(), [Address(email_address='lala@hoho.com')])
529
 
 
530
 
    @testing.fails_on('maxdb', 'FIXME: unknown')
 
559
        eq_(
 
560
            u1.addresses.all(),
 
561
            [Address(email_address='lala@hoho.com')]
 
562
        )
 
563
 
 
564
    def _test_delete_cascade(self, expected):
 
565
        addresses = self.tables.addresses
 
566
        User, Address = self._user_address_fixture(addresses_args={
 
567
                            "order_by": addresses.c.id,
 
568
                            "backref": "user",
 
569
                            "cascade": "save-update" if expected \
 
570
                                            else "all, delete"
 
571
            })
 
572
 
 
573
        sess = create_session(autoflush=True, autocommit=False)
 
574
        u = User(name='ed')
 
575
        u.addresses.extend(
 
576
            [Address(email_address=letter) for letter in 'abcdef']
 
577
        )
 
578
        sess.add(u)
 
579
        sess.commit()
 
580
        eq_(testing.db.scalar(
 
581
                addresses.count(addresses.c.user_id == None)), 0)
 
582
        eq_(testing.db.scalar(
 
583
                addresses.count(addresses.c.user_id != None)), 6)
 
584
 
 
585
        sess.delete(u)
 
586
 
 
587
        sess.commit()
 
588
 
 
589
        if expected:
 
590
            eq_(testing.db.scalar(
 
591
                    addresses.count(addresses.c.user_id == None)), 6)
 
592
            eq_(testing.db.scalar(
 
593
                    addresses.count(addresses.c.user_id != None)), 0)
 
594
        else:
 
595
            eq_(testing.db.scalar(addresses.count()), 0)
 
596
 
531
597
    def test_delete_nocascade(self):
532
 
        users, Address, addresses, User = (self.tables.users,
533
 
                                self.classes.Address,
534
 
                                self.tables.addresses,
535
 
                                self.classes.User)
536
 
 
537
 
        mapper(User, users, properties={
538
 
            'addresses':dynamic_loader(mapper(Address, addresses), order_by=Address.id,
539
 
                                       backref='user')
540
 
        })
541
 
        sess = create_session(autoflush=True, autocommit=False)
542
 
        u = User(name='ed')
543
 
        u.addresses.append(Address(email_address='a'))
544
 
        u.addresses.append(Address(email_address='b'))
545
 
        u.addresses.append(Address(email_address='c'))
546
 
        u.addresses.append(Address(email_address='d'))
547
 
        u.addresses.append(Address(email_address='e'))
548
 
        u.addresses.append(Address(email_address='f'))
549
 
        sess.add(u)
550
 
 
551
 
        eq_(Address(email_address='c'), u.addresses[2])
552
 
        sess.delete(u.addresses[2])
553
 
        sess.delete(u.addresses[4])
554
 
        sess.delete(u.addresses[3])
555
 
        eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='d')],
556
 
            list(u.addresses))
557
 
 
558
 
        sess.expunge_all()
559
 
        u = sess.query(User).get(u.id)
560
 
 
561
 
        sess.delete(u)
562
 
 
563
 
        # u.addresses relationship will have to force the load
564
 
        # of all addresses so that they can be updated
565
 
        sess.flush()
566
 
        sess.close()
567
 
 
568
 
        eq_(testing.db.scalar(addresses.count(addresses.c.user_id != None)), 0)
569
 
 
570
 
    @testing.fails_on('maxdb', 'FIXME: unknown')
 
598
        self._test_delete_cascade(True)
 
599
 
571
600
    def test_delete_cascade(self):
572
 
        users, Address, addresses, User = (self.tables.users,
573
 
                                self.classes.Address,
574
 
                                self.tables.addresses,
575
 
                                self.classes.User)
576
 
 
577
 
        mapper(User, users, properties={
578
 
            'addresses':dynamic_loader(mapper(Address, addresses), order_by=Address.id,
579
 
                                       backref='user', cascade="all, delete-orphan")
580
 
        })
581
 
        sess = create_session(autoflush=True, autocommit=False)
582
 
        u = User(name='ed')
583
 
        u.addresses.append(Address(email_address='a'))
584
 
        u.addresses.append(Address(email_address='b'))
585
 
        u.addresses.append(Address(email_address='c'))
586
 
        u.addresses.append(Address(email_address='d'))
587
 
        u.addresses.append(Address(email_address='e'))
588
 
        u.addresses.append(Address(email_address='f'))
589
 
        sess.add(u)
590
 
 
591
 
        eq_(Address(email_address='c'), u.addresses[2])
592
 
        sess.delete(u.addresses[2])
593
 
        sess.delete(u.addresses[4])
594
 
        sess.delete(u.addresses[3])
595
 
        eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='d')],
596
 
            list(u.addresses))
597
 
 
598
 
        sess.expunge_all()
599
 
        u = sess.query(User).get(u.id)
600
 
 
601
 
        sess.delete(u)
602
 
 
603
 
        # u.addresses relationship will have to force the load
604
 
        # of all addresses so that they can be updated
605
 
        sess.flush()
606
 
        sess.close()
607
 
 
608
 
        eq_(testing.db.scalar(addresses.count()), 0)
609
 
 
610
 
    @testing.fails_on('maxdb', 'FIXME: unknown')
 
601
        self._test_delete_cascade(False)
 
602
 
611
603
    def test_remove_orphans(self):
612
 
        users, Address, addresses, User = (self.tables.users,
613
 
                                self.classes.Address,
614
 
                                self.tables.addresses,
615
 
                                self.classes.User)
 
604
        addresses = self.tables.addresses
 
605
        User, Address = self._user_address_fixture(addresses_args={
 
606
                            "order_by": addresses.c.id,
 
607
                            "backref": "user",
 
608
                            "cascade": "all, delete-orphan"
 
609
            })
616
610
 
617
 
        mapper(User, users, properties={
618
 
            'addresses':dynamic_loader(mapper(Address, addresses), order_by=Address.id,
619
 
                                       cascade="all, delete-orphan", backref='user')
620
 
        })
621
611
        sess = create_session(autoflush=True, autocommit=False)
622
612
        u = User(name='ed')
623
 
        u.addresses.append(Address(email_address='a'))
624
 
        u.addresses.append(Address(email_address='b'))
625
 
        u.addresses.append(Address(email_address='c'))
626
 
        u.addresses.append(Address(email_address='d'))
627
 
        u.addresses.append(Address(email_address='e'))
628
 
        u.addresses.append(Address(email_address='f'))
 
613
        u.addresses.extend(
 
614
            [Address(email_address=letter) for letter in 'abcdef']
 
615
        )
629
616
        sess.add(u)
630
617
 
631
 
        eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='c'),
632
 
             Address(email_address='d'), Address(email_address='e'), Address(email_address='f')],
633
 
            sess.query(Address).all())
634
 
 
635
 
        eq_(Address(email_address='c'), u.addresses[2])
636
 
 
637
 
        def go():
638
 
            del u.addresses[3]
639
 
        assert_raises(
640
 
            TypeError,
641
 
            go
642
 
        )
643
 
 
644
 
        for a in u.addresses.filter(Address.email_address.in_(['c', 'e', 'f'])):
 
618
        for a in u.addresses.filter(
 
619
                Address.email_address.in_(['c', 'e', 'f'])):
645
620
            u.addresses.remove(a)
646
621
 
647
 
        eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='d')],
648
 
            list(u.addresses))
649
 
 
650
 
        eq_([Address(email_address='a'), Address(email_address='b'), Address(email_address='d')],
651
 
            sess.query(Address).all())
652
 
 
653
 
        sess.delete(u)
654
 
        sess.close()
 
622
        eq_(
 
623
            set(ad for ad, in sess.query(Address.email_address)),
 
624
            set(['a', 'b', 'd'])
 
625
        )
655
626
 
656
627
    def _backref_test(self, autoflush, saveuser):
657
 
        users, Address, addresses, User = (self.tables.users,
658
 
                                self.classes.Address,
659
 
                                self.tables.addresses,
660
 
                                self.classes.User)
661
 
 
662
 
        mapper(User, users, properties={
663
 
            'addresses':dynamic_loader(mapper(Address, addresses), backref='user')
664
 
        })
 
628
        User, Address = self._user_address_fixture(addresses_args={
 
629
                            "backref": "user",
 
630
            })
665
631
        sess = create_session(autoflush=autoflush, autocommit=False)
666
632
 
667
633
        u = User(name='buffy')
702
668
    def test_backref_savead(self):
703
669
        self._backref_test(False, False)
704
670
 
705
 
class DontDereferenceTest(fixtures.MappedTest):
706
 
    @classmethod
707
 
    def define_tables(cls, metadata):
708
 
        Table('users', metadata,
709
 
              Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
710
 
              Column('name', String(40)),
711
 
              Column('fullname', String(100)),
712
 
              Column('password', String(15)))
713
 
 
714
 
        Table('addresses', metadata,
715
 
              Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
716
 
              Column('email_address', String(100), nullable=False),
717
 
              Column('user_id', Integer, ForeignKey('users.id')))
718
 
 
719
 
    @classmethod
720
 
    def setup_mappers(cls):
721
 
        users, addresses = cls.tables.users, cls.tables.addresses
722
 
 
723
 
        class User(cls.Comparable):
724
 
            pass
725
 
 
726
 
        class Address(cls.Comparable):
727
 
            pass
728
 
 
729
 
        mapper(User, users, properties={
730
 
            'addresses': relationship(Address, backref='user', lazy='dynamic')
 
671
    def test_backref_events(self):
 
672
        User, Address = self._user_address_fixture(addresses_args={
 
673
                            "backref": "user",
731
674
            })
732
 
        mapper(Address, addresses)
 
675
 
 
676
        u1 = User()
 
677
        a1 = Address()
 
678
        u1.addresses.append(a1)
 
679
        is_(a1.user, u1)
733
680
 
734
681
    def test_no_deref(self):
735
 
        User, Address = self.classes.User, self.classes.Address
 
682
        User, Address = self._user_address_fixture(addresses_args={
 
683
                            "backref": "user",
 
684
            })
736
685
 
737
686
        session = create_session()
738
687
        user = User()
764
713
        eq_(query2(), [Address(email_address='joe@joesdomain.example')])
765
714
        eq_(query3(), [Address(email_address='joe@joesdomain.example')])
766
715
 
767
 
 
 
716
class HistoryTest(_DynamicFixture, _fixtures.FixtureTest):
 
717
    run_inserts = None
 
718
 
 
719
    def _transient_fixture(self, addresses_args={}):
 
720
        User, Address = self._user_address_fixture(
 
721
                                    addresses_args=addresses_args)
 
722
 
 
723
        u1 = User()
 
724
        a1 = Address()
 
725
        return u1, a1
 
726
 
 
727
    def _persistent_fixture(self, autoflush=True, addresses_args={}):
 
728
        User, Address = self._user_address_fixture(
 
729
                                    addresses_args=addresses_args)
 
730
 
 
731
        u1 = User(name='u1')
 
732
        a1 = Address(email_address='a1')
 
733
        s = Session(autoflush=autoflush)
 
734
        s.add(u1)
 
735
        s.flush()
 
736
        return u1, a1, s
 
737
 
 
738
    def _persistent_m2m_fixture(self, autoflush=True, items_args={}):
 
739
        Order, Item = self._order_item_fixture(items_args=items_args)
 
740
 
 
741
        o1 = Order()
 
742
        i1 = Item(description="i1")
 
743
        s = Session(autoflush=autoflush)
 
744
        s.add(o1)
 
745
        s.flush()
 
746
        return o1, i1, s
 
747
 
 
748
    def _assert_history(self, obj, compare, compare_passive=None):
 
749
        if isinstance(obj, self.classes.User):
 
750
            attrname = "addresses"
 
751
        elif isinstance(obj, self.classes.Order):
 
752
            attrname = "items"
 
753
 
 
754
        eq_(
 
755
            attributes.get_history(obj, attrname),
 
756
            compare
 
757
        )
 
758
 
 
759
        if compare_passive is None:
 
760
            compare_passive = compare
 
761
 
 
762
        eq_(
 
763
            attributes.get_history(obj, attrname,
 
764
                        attributes.LOAD_AGAINST_COMMITTED),
 
765
            compare_passive
 
766
        )
 
767
 
 
768
    def test_append_transient(self):
 
769
        u1, a1 = self._transient_fixture()
 
770
        u1.addresses.append(a1)
 
771
 
 
772
        self._assert_history(u1,
 
773
            ([a1], [], [])
 
774
        )
 
775
 
 
776
    def test_append_persistent(self):
 
777
        u1, a1, s = self._persistent_fixture()
 
778
        u1.addresses.append(a1)
 
779
 
 
780
        self._assert_history(u1,
 
781
            ([a1], [], [])
 
782
        )
 
783
 
 
784
    def test_remove_transient(self):
 
785
        u1, a1 = self._transient_fixture()
 
786
        u1.addresses.append(a1)
 
787
        u1.addresses.remove(a1)
 
788
 
 
789
        self._assert_history(u1,
 
790
            ([], [], [])
 
791
        )
 
792
 
 
793
    def test_backref_pop_transient(self):
 
794
        u1, a1 = self._transient_fixture(addresses_args={"backref": "user"})
 
795
        u1.addresses.append(a1)
 
796
 
 
797
        self._assert_history(u1,
 
798
            ([a1], [], []),
 
799
        )
 
800
 
 
801
        a1.user = None
 
802
 
 
803
        # removed from added
 
804
        self._assert_history(u1,
 
805
            ([], [], []),
 
806
        )
 
807
 
 
808
    def test_remove_persistent(self):
 
809
        u1, a1, s = self._persistent_fixture()
 
810
        u1.addresses.append(a1)
 
811
        s.flush()
 
812
        s.expire_all()
 
813
 
 
814
        u1.addresses.remove(a1)
 
815
 
 
816
        self._assert_history(u1,
 
817
            ([], [], [a1])
 
818
        )
 
819
 
 
820
    def test_backref_pop_persistent_autoflush_o2m_active_hist(self):
 
821
        u1, a1, s = self._persistent_fixture(
 
822
                    addresses_args={"backref":
 
823
                        backref("user", active_history=True)})
 
824
        u1.addresses.append(a1)
 
825
        s.flush()
 
826
        s.expire_all()
 
827
 
 
828
        a1.user = None
 
829
 
 
830
        self._assert_history(u1,
 
831
            ([], [], [a1]),
 
832
        )
 
833
 
 
834
    def test_backref_pop_persistent_autoflush_m2m(self):
 
835
        o1, i1, s = self._persistent_m2m_fixture(
 
836
                            items_args={"backref": "orders"})
 
837
        o1.items.append(i1)
 
838
        s.flush()
 
839
        s.expire_all()
 
840
 
 
841
        i1.orders.remove(o1)
 
842
 
 
843
        self._assert_history(o1,
 
844
            ([], [], [i1]),
 
845
        )
 
846
 
 
847
    def test_backref_pop_persistent_noflush_m2m(self):
 
848
        o1, i1, s = self._persistent_m2m_fixture(
 
849
                            items_args={"backref": "orders"}, autoflush=False)
 
850
        o1.items.append(i1)
 
851
        s.flush()
 
852
        s.expire_all()
 
853
 
 
854
        i1.orders.remove(o1)
 
855
 
 
856
        self._assert_history(o1,
 
857
            ([], [], [i1]),
 
858
        )
 
859
 
 
860
    def test_unchanged_persistent(self):
 
861
        Address = self.classes.Address
 
862
 
 
863
        u1, a1, s = self._persistent_fixture()
 
864
        a2, a3 = Address(email_address='a2'), Address(email_address='a3')
 
865
 
 
866
        u1.addresses.append(a1)
 
867
        u1.addresses.append(a2)
 
868
        s.flush()
 
869
 
 
870
        u1.addresses.append(a3)
 
871
        u1.addresses.remove(a2)
 
872
 
 
873
        self._assert_history(u1,
 
874
            ([a3], [a1], [a2]),
 
875
            compare_passive=([a3], [], [a2])
 
876
        )
 
877
 
 
878
    def test_replace_transient(self):
 
879
        Address = self.classes.Address
 
880
 
 
881
        u1, a1 = self._transient_fixture()
 
882
        a2, a3, a4, a5 = Address(email_address='a2'), \
 
883
                            Address(email_address='a3'), \
 
884
                            Address(email_address='a4'), \
 
885
                            Address(email_address='a5')
 
886
 
 
887
        u1.addresses = [a1, a2]
 
888
        u1.addresses = [a2, a3, a4, a5]
 
889
 
 
890
        self._assert_history(u1,
 
891
            ([a2, a3, a4, a5], [], [])
 
892
        )
 
893
 
 
894
    def test_replace_persistent_noflush(self):
 
895
        Address = self.classes.Address
 
896
 
 
897
        u1, a1, s = self._persistent_fixture(autoflush=False)
 
898
        a2, a3, a4, a5 = Address(email_address='a2'), \
 
899
                            Address(email_address='a3'), \
 
900
                            Address(email_address='a4'), \
 
901
                            Address(email_address='a5')
 
902
 
 
903
        u1.addresses = [a1, a2]
 
904
        u1.addresses = [a2, a3, a4, a5]
 
905
 
 
906
        self._assert_history(u1,
 
907
            ([a2, a3, a4, a5], [], [])
 
908
        )
 
909
 
 
910
    def test_replace_persistent_autoflush(self):
 
911
        Address = self.classes.Address
 
912
 
 
913
        u1, a1, s = self._persistent_fixture(autoflush=True)
 
914
        a2, a3, a4, a5 = Address(email_address='a2'), \
 
915
                            Address(email_address='a3'), \
 
916
                            Address(email_address='a4'), \
 
917
                            Address(email_address='a5')
 
918
 
 
919
        u1.addresses = [a1, a2]
 
920
        u1.addresses = [a2, a3, a4, a5]
 
921
 
 
922
        self._assert_history(u1,
 
923
            ([a3, a4, a5], [a2], [a1]),
 
924
            compare_passive=([a3, a4, a5], [], [a1])
 
925
        )
 
926
 
 
927
 
 
928
    def test_persistent_but_readded_noflush(self):
 
929
        u1, a1, s = self._persistent_fixture(autoflush=False)
 
930
        u1.addresses.append(a1)
 
931
        s.flush()
 
932
 
 
933
        u1.addresses.append(a1)
 
934
 
 
935
        self._assert_history(u1,
 
936
            ([], [a1], []),
 
937
            compare_passive=([a1], [], [])
 
938
        )
 
939
 
 
940
    def test_persistent_but_readded_autoflush(self):
 
941
        u1, a1, s = self._persistent_fixture(autoflush=True)
 
942
        u1.addresses.append(a1)
 
943
        s.flush()
 
944
 
 
945
        u1.addresses.append(a1)
 
946
 
 
947
        self._assert_history(u1,
 
948
            ([], [a1], []),
 
949
            compare_passive=([a1], [], [])
 
950
        )
 
951
 
 
952
    def test_missing_but_removed_noflush(self):
 
953
        u1, a1, s = self._persistent_fixture(autoflush=False)
 
954
 
 
955
        u1.addresses.remove(a1)
 
956
 
 
957
        self._assert_history(u1,
 
958
                    ([], [], []),
 
959
                    compare_passive=([], [], [a1])
 
960
                )