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