root / elixir / trunk / tests / test_options.py @ 336

Revision 336, 8.9 kB (checked in by ged, 6 years ago)

update Elixir to work with SA0.5

Line 
1"""
2test options
3"""
4
5from sqlalchemy import UniqueConstraint, create_engine, Column
6from sqlalchemy.orm import create_session
7from sqlalchemy.exceptions import SQLError, ConcurrentModificationError 
8from elixir import *
9
10class TestOptions(object):
11    def setup(self):
12        metadata.bind = 'sqlite:///'
13
14    def teardown(self):
15        cleanup_all()
16
17    # this test is a rip-off SQLAlchemy's activemapper's update test
18    def test_version_id_col(self):
19        class Person(Entity):
20            name = Field(String(30))
21 
22            using_options(version_id_col=True)
23 
24        setup_all()
25        Person.table.create()
26 
27        p1 = Person(name='Daniel')
28        session.flush()
29        session.clear()
30       
31        person = Person.query.first()
32        person.name = 'Gaetan'
33        session.flush()
34        session.clear()
35        assert person.row_version == 2
36 
37        person = Person.query.first()
38        person.name = 'Jonathan'
39        session.flush()
40        session.clear()
41        assert person.row_version == 3
42 
43        # check that a concurrent modification raises exception
44        p1 = Person.query.first()
45        s2 = create_session()
46        p2 = s2.query(Person).first()
47        p1.name = "Daniel"
48        p2.name = "Gaetan"
49        s2.flush()
50        try:
51            session.flush()
52            assert False
53        except ConcurrentModificationError:
54            pass
55        s2.close()
56
57    def test_allowcoloverride_false(self):
58        class MyEntity(Entity):
59            name = Field(String(30))
60
61        setup_all(True)
62
63        raised = False
64        try:
65            MyEntity._descriptor.add_column(Column('name', String(30)))
66        except Exception:
67            raised = True
68
69        assert raised
70
71    def test_allowcoloverride_true(self):
72        class MyEntity(Entity):
73            name = Field(String(30))
74            using_options(allowcoloverride=True)
75
76        setup_all()
77
78        # Note that this test is bogus as you cannot just change a column this
79        # way since the mapper is already constructed at this point and will
80        # use the old column!!! This test is only meant as a way to check no
81        # exception is raised.
82        #TODO: provide a proper test (using autoloaded tables)
83        MyEntity._descriptor.add_column(Column('name', String(30),
84                                               default='test'))
85
86    def test_tablename_func(self):
87        import re
88
89        def camel_to_underscore(entity):
90            return re.sub(r'(.+?)([A-Z])+?', r'\1_\2', entity.__name__).lower()
91
92        options_defaults['tablename'] = camel_to_underscore
93
94        class MyEntity(Entity):
95            name = Field(String(30))
96
97        class MySuperTestEntity(Entity):
98            name = Field(String(30))
99
100        setup_all(True)
101
102        assert MyEntity.table.name == 'my_entity'
103        assert MySuperTestEntity.table.name == 'my_super_test_entity'
104
105        options_defaults['tablename'] = None
106
107
108class TestSessionOptions(object):
109    def setup(self):
110        metadata.bind = None
111
112    def teardown(self):
113        cleanup_all()
114
115    def test_session_context(self):
116        try:
117            from sqlalchemy.ext.sessioncontext import SessionContext
118        except ImportError:
119            # we are probably on SQLAlchemy 0.5, no need to test this.
120            return
121
122        engine = create_engine('sqlite:///')
123       
124        ctx = SessionContext(lambda: create_session(bind=engine))
125       
126        class Person(Entity):
127            using_options(session=ctx)
128            firstname = Field(String(30))
129            surname = Field(String(30))
130
131        setup_all()
132        create_all(engine)
133
134        homer = Person(firstname="Homer", surname='Simpson')
135        bart = Person(firstname="Bart", surname='Simpson')
136        ctx.current.flush()
137       
138        assert Person.query.session is ctx.current
139        assert Person.query.filter_by(firstname='Homer').one() is homer
140
141    def test_manual_session(self):
142        engine = create_engine('sqlite:///')
143       
144        class Person(Entity):
145            using_options(session=None)
146            firstname = Field(String(30))
147            surname = Field(String(30))
148
149        setup_all()
150        create_all(engine)
151
152        session = create_session(bind=engine)
153
154        homer = Person(firstname="Homer", surname='Simpson')
155        bart = Person(firstname="Bart", surname='Simpson')
156
157        session.save(homer)
158        session.save(bart)
159        session.flush()
160       
161        bart.delete()
162        session.flush()
163
164        assert session.query(Person).filter_by(firstname='Homer').one() is homer
165        assert session.query(Person).count() == 1
166
167    def test_activemapper_session(self):
168        try:
169            from sqlalchemy.orm import scoped_session, sessionmaker
170            #TODO: this test, as-is has no sense on SA 0.4 since activemapper
171            # session uses scoped_session, but we need to provide a new
172            # test for that.
173            return
174        except ImportError:
175            pass
176
177        try:
178            from sqlalchemy.ext import activemapper
179        except ImportError:
180            return
181           
182        engine = create_engine('sqlite:///')
183
184        store = activemapper.Objectstore(lambda: create_session(bind=engine))
185
186        class Person(Entity):
187            using_options(session=store)
188            firstname = Field(String(30))
189            surname = Field(String(30))
190
191        setup_all()
192        create_all(engine)
193
194        homer = Person(firstname="Homer", surname='Simpson')
195        bart = Person(firstname="Bart", surname='Simpson')
196
197        store.flush()
198
199        assert Person.query.session is store.context.current
200        assert Person.query.filter_by(firstname='Homer').one() is homer
201
202    def test_scoped_session(self):
203        try:
204            from sqlalchemy.orm import scoped_session, sessionmaker
205        except ImportError:
206            print "Not on version 0.4 of sqlalchemy"
207            return
208           
209        engine = create_engine('sqlite:///')
210
211        Session = scoped_session(sessionmaker(bind=engine))
212
213        class Person(Entity):
214            using_options(session=Session)
215            firstname = Field(String(30))
216            surname = Field(String(30))
217
218        setup_all()
219        create_all(engine)
220
221        homer = Person(firstname="Homer", surname='Simpson')
222        bart = Person(firstname="Bart", surname='Simpson')
223        Session.flush()
224
225        assert Person.query.session is Session()
226        assert Person.query.filter_by(firstname='Homer').one() is homer
227       
228    def test_global_scoped_session(self):
229        try:
230            from sqlalchemy.orm import scoped_session, sessionmaker
231        except ImportError:
232            print "Not on version 0.4 of sqlalchemy"
233            return
234
235        global __session__
236       
237        engine = create_engine('sqlite:///')
238       
239        session = scoped_session(sessionmaker(bind=engine))
240        __session__ = session
241       
242        class Person(Entity):
243            firstname = Field(String(30))
244            surname = Field(String(30))
245
246        setup_all()
247        create_all(engine)
248
249        homer = Person(firstname="Homer", surname='Simpson')
250        bart = Person(firstname="Bart", surname='Simpson')
251        session.flush()
252       
253        assert Person.query.session is session()
254        assert Person.query.filter_by(firstname='Homer').one() is homer
255
256        del __session__
257       
258class TestTableOptions(object):
259    def setup(self):
260        metadata.bind = 'sqlite:///'
261
262    def teardown(self):
263        cleanup_all()
264
265    def test_unique_constraint(self):
266       
267        class Person(Entity):
268            firstname = Field(String(30))
269            surname = Field(String(30))
270
271            using_table_options(UniqueConstraint('firstname', 'surname'))
272
273        setup_all(True)
274
275        homer = Person(firstname="Homer", surname='Simpson')
276        bart = Person(firstname="Bart", surname='Simpson')
277
278        session.flush()
279
280        homer2 = Person(firstname="Homer", surname='Simpson')
281
282        raised = False
283        try:
284            session.flush()
285        except SQLError:
286            raised = True
287
288        assert raised
289
290    def test_unique_constraint_belongs_to(self):
291        class Author(Entity):
292            name = Field(String(50))
293
294        class Book(Entity):
295            title = Field(String(200), required=True)
296            author = ManyToOne("Author")
297
298            using_table_options(UniqueConstraint("title", "author_id"))
299
300        setup_all(True)
301
302        tolkien = Author(name="J. R. R. Tolkien")
303        lotr = Book(title="The Lord of the Rings", author=tolkien)
304        hobbit = Book(title="The Hobbit", author=tolkien)
305
306        session.flush()
307
308        tolkien2 = Author(name="Tolkien")
309        hobbit2 = Book(title="The Hobbit", author=tolkien2)
310
311        session.flush()
312
313        hobbit3 = Book(title="The Hobbit", author=tolkien)
314
315        raised = False
316        try:
317            session.flush()
318        except SQLError:
319            raised = True
320
321        assert raised
Note: See TracBrowser for help on using the browser.