← Back to branch summary

~ubuntu-branches/debian/sid/sqlalchemy/sid

« back to all changes in this revision

Viewing changes to test/orm/test_expire.py

  • Committer: Bazaar Package Importer
  • Author(s): Piotr Ożarowski
  • Date: 2011-03-27 14:22:50 UTC
  • mfrom: (16.1.9 experimental)
  • Revision ID: james.westby@ubuntu.com-20110327142250-aip46dv6a3r2jwvs
Tags: 0.6.6-2
Upload to unstable

Show diffs side-by-side

added added

removed removed

Lines of Context:
test/orm/test_expire.py
9
9
from sqlalchemy.test.schema import Column
10
10
from sqlalchemy.orm import mapper, relationship, create_session, \
11
11
                        attributes, deferred, exc as orm_exc, defer, undefer,\
12
 
                        strategies, state, lazyload, backref
 
12
                        strategies, state, lazyload, backref, Session
13
13
from test.orm import _base, _fixtures
14
14
 
15
15
 
89
89
        assert s.query(User).get(10) is None
90
90
        assert u not in s # and expunges
91
91
 
 
92
        # trick the "deleted" flag so we can re-add for the sake
 
93
        # of this test
 
94
        del attributes.instance_state(u).deleted
 
95
 
92
96
        # add it back
93
97
        s.add(u)
94
98
        # nope, raises ObjectDeletedError
97
101
        # do a get()/remove u from session again
98
102
        assert s.query(User).get(10) is None
99
103
        assert u not in s
100
 
        
 
104
 
101
105
        s.rollback()
102
106
 
103
107
        assert u in s
108
112
    @testing.resolve_artifact_names
109
113
    def test_deferred(self):
110
114
        """test that unloaded, deferred attributes aren't included in the expiry list."""
111
 
        
 
115
 
112
116
        mapper(Order, orders, properties={'description':deferred(orders.c.description)})
113
 
        
 
117
 
114
118
        s = create_session()
115
119
        o1 = s.query(Order).first()
116
120
        assert 'description' not in o1.__dict__
118
122
        assert o1.isopen is not None
119
123
        assert 'description' not in o1.__dict__
120
124
        assert o1.description
121
 
        
 
125
 
122
126
    @testing.resolve_artifact_names
123
127
    def test_lazyload_autoflushes(self):
124
128
        mapper(User, users, properties={
144
148
 
145
149
    @testing.resolve_artifact_names
146
150
    def test_refresh_collection_exception(self):
147
 
        """test graceful failure for currently unsupported immediate refresh of a collection"""
148
 
        
 
151
        """test graceful failure for currently unsupported 
 
152
        immediate refresh of a collection"""
 
153
 
149
154
        mapper(User, users, properties={
150
155
            'addresses':relationship(Address, order_by=addresses.c.email_address)
151
156
        })
152
157
        mapper(Address, addresses)
153
158
        s = create_session(autoflush=True, autocommit=False)
154
159
        u = s.query(User).get(8)
155
 
        assert_raises_message(sa_exc.InvalidRequestError, "properties specified for refresh", s.refresh, u, ['addresses'])
156
 
        
 
160
        assert_raises_message(sa_exc.InvalidRequestError, 
 
161
                        "properties specified for refresh", 
 
162
                        s.refresh, u, ['addresses'])
 
163
 
157
164
        # in contrast to a regular query with no columns
158
 
        assert_raises_message(sa_exc.InvalidRequestError, "no columns with which to SELECT", s.query().all)
159
 
        
 
165
        assert_raises_message(sa_exc.InvalidRequestError, 
 
166
                        "no columns with which to SELECT", s.query().all)
 
167
 
160
168
    @testing.resolve_artifact_names
161
169
    def test_refresh_cancels_expire(self):
162
170
        mapper(User, users)
221
229
        assert 'name' not in u.__dict__
222
230
        sess.add(u)
223
231
        assert u.name == 'jack'
224
 
    
 
232
 
225
233
    @testing.resolve_artifact_names
226
234
    def test_no_instance_key_no_pk(self):
227
235
        # same as test_no_instance_key, but the PK columns
236
244
        assert 'name' not in u.__dict__
237
245
        sess.add(u)
238
246
        assert_raises(sa_exc.InvalidRequestError, getattr, u, 'name')
239
 
        
240
 
        
 
247
 
 
248
 
241
249
    @testing.resolve_artifact_names
242
250
    def test_expire_preserves_changes(self):
243
251
        """test that the expire load operation doesn't revert post-expire changes"""
342
350
    def test_refresh_cascade_pending(self):
343
351
        cascade = 'save-update, refresh-expire'
344
352
        self._test_cascade_to_pending(cascade, False)
345
 
        
 
353
 
346
354
    @testing.resolve_artifact_names
347
355
    def _test_cascade_to_pending(self, cascade, expire_or_refresh):
348
356
        mapper(User, users, properties={
353
361
 
354
362
        u = s.query(User).get(8)
355
363
        a = Address(id=12, email_address='foobar')
356
 
        
 
364
 
357
365
        u.addresses.append(a)
358
366
        if expire_or_refresh:
359
367
            s.expire(u)
363
371
            assert a not in s
364
372
        else:
365
373
            assert a in s
366
 
        
 
374
 
367
375
        assert a not in u.addresses
368
376
        s.flush()
369
377
 
728
736
    @testing.resolve_artifact_names
729
737
    def test_expire_all(self):
730
738
        mapper(User, users, properties={
731
 
            'addresses':relationship(Address, backref='user', lazy='joined'),
 
739
            'addresses':relationship(Address, backref='user', lazy='joined', 
 
740
                                    order_by=addresses.c.id),
732
741
            })
733
742
        mapper(Address, addresses)
734
743
 
750
759
        """Behavioral test to verify the current activity of loader callables."""
751
760
 
752
761
        mapper(User, users)
753
 
        
 
762
 
754
763
        sess = create_session()
755
 
        
 
764
 
756
765
        # deferred attribute option, gets the LoadDeferredColumns
757
766
        # callable
758
767
        u1 = sess.query(User).options(defer(User.name)).first()
760
769
                    attributes.instance_state(u1).callables['name'],
761
770
                    strategies.LoadDeferredColumns
762
771
                )
763
 
                
 
772
 
764
773
        # expire the attr, it gets the InstanceState callable
765
774
        sess.expire(u1, ['name'])
766
775
        assert isinstance(
767
776
                    attributes.instance_state(u1).callables['name'],
768
777
                    state.InstanceState
769
778
                )
770
 
                
 
779
 
771
780
        # load it, callable is gone
772
781
        u1.name
773
782
        assert 'name' not in attributes.instance_state(u1).callables
780
789
                    attributes.instance_state(u1).callables['name'],
781
790
                    state.InstanceState
782
791
                )
783
 
        
 
792
 
784
793
        # load over it.  everything normal.
785
794
        sess.query(User).first()
786
795
        assert 'name' not in attributes.instance_state(u1).callables
787
 
        
 
796
 
788
797
        sess.expunge_all()
789
798
        u1 = sess.query(User).first()
790
799
        # for non present, still expires the same way
791
800
        del u1.name
792
801
        sess.expire(u1)
793
802
        assert 'name' in attributes.instance_state(u1).callables
794
 
        
 
803
 
795
804
    @testing.resolve_artifact_names
796
805
    def test_state_deferred_to_col(self):
797
806
        """Behavioral test to verify the current activity of loader callables."""
798
 
        
 
807
 
799
808
        mapper(User, users, properties={'name':deferred(users.c.name)})
800
809
 
801
810
        sess = create_session()
802
811
        u1 = sess.query(User).options(undefer(User.name)).first()
803
812
        assert 'name' not in attributes.instance_state(u1).callables
804
 
        
 
813
 
805
814
        # mass expire, the attribute was loaded, 
806
815
        # the attribute gets the callable
807
816
        sess.expire(u1)
813
822
        # load it, callable is gone
814
823
        u1.name
815
824
        assert 'name' not in attributes.instance_state(u1).callables
816
 
        
 
825
 
817
826
        # mass expire, attribute was loaded but then deleted,
818
827
        # the callable goes away - the state wants to flip 
819
828
        # it back to its "deferred" loader.
838
847
 
839
848
        mapper(User, users, properties={'addresses':relationship(Address, lazy='noload')})
840
849
        mapper(Address, addresses)
841
 
        
 
850
 
842
851
        sess = create_session()
843
852
        u1 = sess.query(User).options(lazyload(User.addresses)).first()
844
853
        assert isinstance(
851
860
                    attributes.instance_state(u1).callables['addresses'],
852
861
                    strategies.LoadLazyAttribute
853
862
                )
854
 
        
 
863
 
855
864
        # load over it.  callable goes away.
856
865
        sess.query(User).first()
857
866
        assert 'addresses' not in attributes.instance_state(u1).callables
858
 
        
 
867
 
859
868
        sess.expunge_all()
860
869
        u1 = sess.query(User).options(lazyload(User.addresses)).first()
861
870
        sess.expire(u1, ['addresses'])
863
872
                    attributes.instance_state(u1).callables['addresses'],
864
873
                    strategies.LoadLazyAttribute
865
874
                )
866
 
        
 
875
 
867
876
        # load the attr, goes away
868
877
        u1.addresses
869
878
        assert 'addresses' not in attributes.instance_state(u1).callables
870
 
        
871
 
        
872
 
        
 
879
 
 
880
 
 
881
 
873
882
class PolymorphicExpireTest(_base.MappedTest):
874
883
    run_inserts = 'once'
875
884
    run_deletes = None
882
891
           Column('person_id', Integer, primary_key=True,
883
892
                  test_needs_autoincrement=True),
884
893
           Column('name', String(50)),
885
 
           Column('type', String(30)))
 
894
           Column('type', String(30)),
 
895
           )
886
896
 
887
897
        engineers = Table('engineers', metadata,
888
898
           Column('person_id', Integer, ForeignKey('people.person_id'),
910
920
            {'person_id':3, 'status':'old engineer'},
911
921
        )
912
922
 
 
923
    @classmethod
913
924
    @testing.resolve_artifact_names
914
 
    def test_poly_deferred(self):
 
925
    def setup_mappers(cls):
915
926
        mapper(Person, people, polymorphic_on=people.c.type, polymorphic_identity='person')
916
927
        mapper(Engineer, engineers, inherits=Person, polymorphic_identity='engineer')
917
928
 
 
929
    @testing.resolve_artifact_names
 
930
    def test_poly_deferred(self):
 
931
 
918
932
        sess = create_session()
919
933
        [p1, e1, e2] = sess.query(Person).order_by(people.c.person_id).all()
920
934
 
949
963
        self.assert_sql_count(testing.db, go, 2)
950
964
        eq_(Engineer.name.get_history(e1), (['new engineer name'],(), ['engineer1']))
951
965
 
 
966
    @testing.resolve_artifact_names
 
967
    def test_no_instance_key(self):
 
968
 
 
969
        sess = create_session()
 
970
        e1 = sess.query(Engineer).get(2)
 
971
 
 
972
        sess.expire(e1, attribute_names=['name'])
 
973
        sess.expunge(e1)
 
974
        attributes.instance_state(e1).key = None
 
975
        assert 'name' not in e1.__dict__
 
976
        sess.add(e1)
 
977
        assert e1.name == 'engineer1'
 
978
 
 
979
    @testing.resolve_artifact_names
 
980
    def test_no_instance_key(self):
 
981
        # same as test_no_instance_key, but the PK columns
 
982
        # are absent.  ensure an error is raised.
 
983
        sess = create_session()
 
984
        e1 = sess.query(Engineer).get(2)
 
985
 
 
986
        sess.expire(e1, attribute_names=['name', 'person_id'])
 
987
        sess.expunge(e1)
 
988
        attributes.instance_state(e1).key = None
 
989
        assert 'name' not in e1.__dict__
 
990
        sess.add(e1)
 
991
        assert_raises(sa_exc.InvalidRequestError, getattr, e1, 'name')
 
992
 
 
993
 
952
994
class ExpiredPendingTest(_fixtures.FixtureTest):
953
995
    run_define_tables = 'once'
954
996
    run_setup_classes = 'once'
955
997
    run_setup_mappers = None
956
998
    run_inserts = None
957
 
    
 
999
 
958
1000
    @testing.resolve_artifact_names
959
1001
    def test_expired_pending(self):
960
1002
        mapper(User, users, properties={
966
1008
        a1 = Address(email_address='a1')
967
1009
        sess.add(a1)
968
1010
        sess.flush()
969
 
        
 
1011
 
970
1012
        u1 = User(name='u1')
971
1013
        a1.user = u1
972
1014
        sess.flush()
983
1025
        # expire u1.addresses again.  this expires
984
1026
        # "pending" as well.
985
1027
        sess.expire(u1, ['addresses'])
986
 
        
 
1028
 
987
1029
        # insert a new row
988
1030
        sess.execute(addresses.insert(), dict(email_address='a3', user_id=u1.id))
989
 
        
 
1031
 
990
1032
        # only two addresses pulled from the DB, no "pending"
991
1033
        assert len(u1.addresses) == 2
992
 
        
 
1034
 
993
1035
        sess.flush()
994
1036
        sess.expire_all()
995
1037
        assert len(u1.addresses) == 3
996
 
    
 
1038
 
997
1039
 
998
1040
class RefreshTest(_fixtures.FixtureTest):
999
1041