← Back to branch summary

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

« back to all changes in this revision

Viewing changes to test/ext/declarative/test_reflection.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/ext/declarative/test_reflection.py
 
1
from sqlalchemy.testing import eq_, assert_raises
 
2
from sqlalchemy.ext import declarative as decl
 
3
from sqlalchemy import testing
 
4
from sqlalchemy import MetaData, Integer, String, ForeignKey
 
5
from sqlalchemy.testing.schema import Table, Column
 
6
from sqlalchemy.orm import relationship, create_session, \
 
7
    clear_mappers, \
 
8
    Session
 
9
from sqlalchemy.testing import fixtures
 
10
 
 
11
class DeclarativeReflectionBase(fixtures.TablesTest):
 
12
    __requires__ = 'reflectable_autoincrement',
 
13
 
 
14
    def setup(self):
 
15
        global Base
 
16
        Base = decl.declarative_base(testing.db)
 
17
 
 
18
    def teardown(self):
 
19
        super(DeclarativeReflectionBase, self).teardown()
 
20
        clear_mappers()
 
21
 
 
22
class DeclarativeReflectionTest(DeclarativeReflectionBase):
 
23
 
 
24
    @classmethod
 
25
    def define_tables(cls, metadata):
 
26
        Table('users', metadata,
 
27
            Column('id', Integer,
 
28
                primary_key=True, test_needs_autoincrement=True),
 
29
              Column('name', String(50)), test_needs_fk=True)
 
30
        Table(
 
31
            'addresses',
 
32
            metadata,
 
33
            Column('id', Integer, primary_key=True,
 
34
                   test_needs_autoincrement=True),
 
35
            Column('email', String(50)),
 
36
            Column('user_id', Integer, ForeignKey('users.id')),
 
37
            test_needs_fk=True,
 
38
            )
 
39
        Table(
 
40
            'imhandles',
 
41
            metadata,
 
42
            Column('id', Integer, primary_key=True,
 
43
                   test_needs_autoincrement=True),
 
44
            Column('user_id', Integer),
 
45
            Column('network', String(50)),
 
46
            Column('handle', String(50)),
 
47
            test_needs_fk=True,
 
48
            )
 
49
 
 
50
    def test_basic(self):
 
51
        meta = MetaData(testing.db)
 
52
 
 
53
        class User(Base, fixtures.ComparableEntity):
 
54
 
 
55
            __tablename__ = 'users'
 
56
            __autoload__ = True
 
57
            if testing.against('oracle', 'firebird'):
 
58
                id = Column('id', Integer, primary_key=True,
 
59
                            test_needs_autoincrement=True)
 
60
            addresses = relationship('Address', backref='user')
 
61
 
 
62
        class Address(Base, fixtures.ComparableEntity):
 
63
 
 
64
            __tablename__ = 'addresses'
 
65
            __autoload__ = True
 
66
            if testing.against('oracle', 'firebird'):
 
67
                id = Column('id', Integer, primary_key=True,
 
68
                            test_needs_autoincrement=True)
 
69
 
 
70
        u1 = User(name='u1', addresses=[Address(email='one'),
 
71
                  Address(email='two')])
 
72
        sess = create_session()
 
73
        sess.add(u1)
 
74
        sess.flush()
 
75
        sess.expunge_all()
 
76
        eq_(sess.query(User).all(), [User(name='u1',
 
77
            addresses=[Address(email='one'), Address(email='two')])])
 
78
        a1 = sess.query(Address).filter(Address.email == 'two').one()
 
79
        eq_(a1, Address(email='two'))
 
80
        eq_(a1.user, User(name='u1'))
 
81
 
 
82
    def test_rekey(self):
 
83
        meta = MetaData(testing.db)
 
84
 
 
85
        class User(Base, fixtures.ComparableEntity):
 
86
 
 
87
            __tablename__ = 'users'
 
88
            __autoload__ = True
 
89
            if testing.against('oracle', 'firebird'):
 
90
                id = Column('id', Integer, primary_key=True,
 
91
                            test_needs_autoincrement=True)
 
92
            nom = Column('name', String(50), key='nom')
 
93
            addresses = relationship('Address', backref='user')
 
94
 
 
95
        class Address(Base, fixtures.ComparableEntity):
 
96
 
 
97
            __tablename__ = 'addresses'
 
98
            __autoload__ = True
 
99
            if testing.against('oracle', 'firebird'):
 
100
                id = Column('id', Integer, primary_key=True,
 
101
                            test_needs_autoincrement=True)
 
102
 
 
103
        u1 = User(nom='u1', addresses=[Address(email='one'),
 
104
                  Address(email='two')])
 
105
        sess = create_session()
 
106
        sess.add(u1)
 
107
        sess.flush()
 
108
        sess.expunge_all()
 
109
        eq_(sess.query(User).all(), [User(nom='u1',
 
110
            addresses=[Address(email='one'), Address(email='two')])])
 
111
        a1 = sess.query(Address).filter(Address.email == 'two').one()
 
112
        eq_(a1, Address(email='two'))
 
113
        eq_(a1.user, User(nom='u1'))
 
114
        assert_raises(TypeError, User, name='u3')
 
115
 
 
116
    def test_supplied_fk(self):
 
117
        meta = MetaData(testing.db)
 
118
 
 
119
        class IMHandle(Base, fixtures.ComparableEntity):
 
120
 
 
121
            __tablename__ = 'imhandles'
 
122
            __autoload__ = True
 
123
            if testing.against('oracle', 'firebird'):
 
124
                id = Column('id', Integer, primary_key=True,
 
125
                            test_needs_autoincrement=True)
 
126
            user_id = Column('user_id', Integer, ForeignKey('users.id'))
 
127
 
 
128
        class User(Base, fixtures.ComparableEntity):
 
129
 
 
130
            __tablename__ = 'users'
 
131
            __autoload__ = True
 
132
            if testing.against('oracle', 'firebird'):
 
133
                id = Column('id', Integer, primary_key=True,
 
134
                            test_needs_autoincrement=True)
 
135
            handles = relationship('IMHandle', backref='user')
 
136
 
 
137
        u1 = User(name='u1', handles=[IMHandle(network='blabber',
 
138
                  handle='foo'), IMHandle(network='lol', handle='zomg'
 
139
                  )])
 
140
        sess = create_session()
 
141
        sess.add(u1)
 
142
        sess.flush()
 
143
        sess.expunge_all()
 
144
        eq_(sess.query(User).all(), [User(name='u1',
 
145
            handles=[IMHandle(network='blabber', handle='foo'),
 
146
            IMHandle(network='lol', handle='zomg')])])
 
147
        a1 = sess.query(IMHandle).filter(IMHandle.handle == 'zomg'
 
148
                ).one()
 
149
        eq_(a1, IMHandle(network='lol', handle='zomg'))
 
150
        eq_(a1.user, User(name='u1'))
 
151
 
 
152
class DeferredReflectBase(DeclarativeReflectionBase):
 
153
    def teardown(self):
 
154
        super(DeferredReflectBase,self).teardown()
 
155
        from sqlalchemy.ext.declarative.base import _MapperConfig
 
156
        _MapperConfig.configs.clear()
 
157
 
 
158
Base = None
 
159
 
 
160
class DeferredReflectPKFKTest(DeferredReflectBase):
 
161
    @classmethod
 
162
    def define_tables(cls, metadata):
 
163
        Table("a", metadata,
 
164
            Column('id', Integer,
 
165
                primary_key=True, test_needs_autoincrement=True),
 
166
        )
 
167
        Table("b", metadata,
 
168
            Column('id', Integer,
 
169
                ForeignKey('a.id'),
 
170
                primary_key=True),
 
171
            Column('x', Integer, primary_key=True)
 
172
        )
 
173
 
 
174
    def test_pk_fk(self):
 
175
        class B(decl.DeferredReflection, fixtures.ComparableEntity,
 
176
                            Base):
 
177
            __tablename__ = 'b'
 
178
            a = relationship("A")
 
179
 
 
180
        class A(decl.DeferredReflection, fixtures.ComparableEntity,
 
181
                            Base):
 
182
            __tablename__ = 'a'
 
183
 
 
184
        decl.DeferredReflection.prepare(testing.db)
 
185
 
 
186
class DeferredReflectionTest(DeferredReflectBase):
 
187
 
 
188
    @classmethod
 
189
    def define_tables(cls, metadata):
 
190
        Table('users', metadata,
 
191
            Column('id', Integer,
 
192
                primary_key=True, test_needs_autoincrement=True),
 
193
              Column('name', String(50)), test_needs_fk=True)
 
194
        Table(
 
195
            'addresses',
 
196
            metadata,
 
197
            Column('id', Integer, primary_key=True,
 
198
                   test_needs_autoincrement=True),
 
199
            Column('email', String(50)),
 
200
            Column('user_id', Integer, ForeignKey('users.id')),
 
201
            test_needs_fk=True,
 
202
            )
 
203
 
 
204
    def _roundtrip(self):
 
205
 
 
206
        User = Base._decl_class_registry['User']
 
207
        Address = Base._decl_class_registry['Address']
 
208
 
 
209
        u1 = User(name='u1', addresses=[Address(email='one'),
 
210
                  Address(email='two')])
 
211
        sess = create_session()
 
212
        sess.add(u1)
 
213
        sess.flush()
 
214
        sess.expunge_all()
 
215
        eq_(sess.query(User).all(), [User(name='u1',
 
216
            addresses=[Address(email='one'), Address(email='two')])])
 
217
        a1 = sess.query(Address).filter(Address.email == 'two').one()
 
218
        eq_(a1, Address(email='two'))
 
219
        eq_(a1.user, User(name='u1'))
 
220
 
 
221
    def test_basic_deferred(self):
 
222
        class User(decl.DeferredReflection, fixtures.ComparableEntity,
 
223
                            Base):
 
224
            __tablename__ = 'users'
 
225
            addresses = relationship("Address", backref="user")
 
226
 
 
227
        class Address(decl.DeferredReflection, fixtures.ComparableEntity,
 
228
                            Base):
 
229
            __tablename__ = 'addresses'
 
230
 
 
231
        decl.DeferredReflection.prepare(testing.db)
 
232
        self._roundtrip()
 
233
 
 
234
    def test_abstract_base(self):
 
235
        class DefBase(decl.DeferredReflection, Base):
 
236
            __abstract__ = True
 
237
 
 
238
        class OtherDefBase(decl.DeferredReflection, Base):
 
239
            __abstract__ = True
 
240
 
 
241
        class User(fixtures.ComparableEntity, DefBase):
 
242
            __tablename__ = 'users'
 
243
            addresses = relationship("Address", backref="user")
 
244
 
 
245
        class Address(fixtures.ComparableEntity, DefBase):
 
246
            __tablename__ = 'addresses'
 
247
 
 
248
        class Fake(OtherDefBase):
 
249
            __tablename__ = 'nonexistent'
 
250
 
 
251
        DefBase.prepare(testing.db)
 
252
        self._roundtrip()
 
253
 
 
254
    def test_redefine_fk_double(self):
 
255
        class User(decl.DeferredReflection, fixtures.ComparableEntity,
 
256
                            Base):
 
257
            __tablename__ = 'users'
 
258
            addresses = relationship("Address", backref="user")
 
259
 
 
260
        class Address(decl.DeferredReflection, fixtures.ComparableEntity,
 
261
                            Base):
 
262
            __tablename__ = 'addresses'
 
263
            user_id = Column(Integer, ForeignKey('users.id'))
 
264
 
 
265
        decl.DeferredReflection.prepare(testing.db)
 
266
        self._roundtrip()
 
267
 
 
268
    def test_mapper_args_deferred(self):
 
269
        """test that __mapper_args__ is not called until *after* table reflection"""
 
270
 
 
271
        class User(decl.DeferredReflection, fixtures.ComparableEntity,
 
272
                            Base):
 
273
            __tablename__ = 'users'
 
274
 
 
275
            @decl.declared_attr
 
276
            def __mapper_args__(cls):
 
277
                return {
 
278
                    "order_by":cls.__table__.c.name
 
279
                }
 
280
 
 
281
        decl.DeferredReflection.prepare(testing.db)
 
282
        sess = Session()
 
283
        sess.add_all([
 
284
            User(name='G'),
 
285
            User(name='Q'),
 
286
            User(name='A'),
 
287
            User(name='C'),
 
288
        ])
 
289
        sess.commit()
 
290
        eq_(
 
291
            sess.query(User).all(),
 
292
            [
 
293
                User(name='A'),
 
294
                User(name='C'),
 
295
                User(name='G'),
 
296
                User(name='Q'),
 
297
            ]
 
298
        )
 
299
 
 
300
class DeferredInhReflectBase(DeferredReflectBase):
 
301
    def _roundtrip(self):
 
302
        Foo = Base._decl_class_registry['Foo']
 
303
        Bar = Base._decl_class_registry['Bar']
 
304
 
 
305
        s = Session(testing.db)
 
306
 
 
307
        s.add_all([
 
308
            Bar(data='d1', bar_data='b1'),
 
309
            Bar(data='d2', bar_data='b2'),
 
310
            Bar(data='d3', bar_data='b3'),
 
311
            Foo(data='d4')
 
312
        ])
 
313
        s.commit()
 
314
 
 
315
        eq_(
 
316
            s.query(Foo).order_by(Foo.id).all(),
 
317
            [
 
318
                Bar(data='d1', bar_data='b1'),
 
319
                Bar(data='d2', bar_data='b2'),
 
320
                Bar(data='d3', bar_data='b3'),
 
321
                Foo(data='d4')
 
322
            ]
 
323
        )
 
324
 
 
325
class DeferredSingleInhReflectionTest(DeferredInhReflectBase):
 
326
 
 
327
    @classmethod
 
328
    def define_tables(cls, metadata):
 
329
        Table("foo", metadata,
 
330
            Column('id', Integer, primary_key=True,
 
331
                        test_needs_autoincrement=True),
 
332
            Column('type', String(32)),
 
333
            Column('data', String(30)),
 
334
            Column('bar_data', String(30))
 
335
        )
 
336
 
 
337
    def test_basic(self):
 
338
        class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
 
339
                    Base):
 
340
            __tablename__ = 'foo'
 
341
            __mapper_args__ = {"polymorphic_on":"type",
 
342
                        "polymorphic_identity":"foo"}
 
343
 
 
344
        class Bar(Foo):
 
345
            __mapper_args__ = {"polymorphic_identity":"bar"}
 
346
 
 
347
        decl.DeferredReflection.prepare(testing.db)
 
348
        self._roundtrip()
 
349
 
 
350
    def test_add_subclass_column(self):
 
351
        class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
 
352
                    Base):
 
353
            __tablename__ = 'foo'
 
354
            __mapper_args__ = {"polymorphic_on":"type",
 
355
                        "polymorphic_identity":"foo"}
 
356
 
 
357
        class Bar(Foo):
 
358
            __mapper_args__ = {"polymorphic_identity":"bar"}
 
359
            bar_data = Column(String(30))
 
360
 
 
361
        decl.DeferredReflection.prepare(testing.db)
 
362
        self._roundtrip()
 
363
 
 
364
    def test_add_pk_column(self):
 
365
        class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
 
366
                    Base):
 
367
            __tablename__ = 'foo'
 
368
            __mapper_args__ = {"polymorphic_on":"type",
 
369
                        "polymorphic_identity":"foo"}
 
370
            id = Column(Integer, primary_key=True)
 
371
 
 
372
        class Bar(Foo):
 
373
            __mapper_args__ = {"polymorphic_identity":"bar"}
 
374
 
 
375
        decl.DeferredReflection.prepare(testing.db)
 
376
        self._roundtrip()
 
377
 
 
378
class DeferredJoinedInhReflectionTest(DeferredInhReflectBase):
 
379
    @classmethod
 
380
    def define_tables(cls, metadata):
 
381
        Table("foo", metadata,
 
382
            Column('id', Integer, primary_key=True,
 
383
                        test_needs_autoincrement=True),
 
384
            Column('type', String(32)),
 
385
            Column('data', String(30)),
 
386
            test_needs_fk=True,
 
387
        )
 
388
        Table('bar', metadata,
 
389
            Column('id', Integer, ForeignKey('foo.id'), primary_key=True),
 
390
            Column('bar_data', String(30)),
 
391
            test_needs_fk=True,
 
392
        )
 
393
 
 
394
    def test_basic(self):
 
395
        class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
 
396
                    Base):
 
397
            __tablename__ = 'foo'
 
398
            __mapper_args__ = {"polymorphic_on":"type",
 
399
                        "polymorphic_identity":"foo"}
 
400
 
 
401
        class Bar(Foo):
 
402
            __tablename__ = 'bar'
 
403
            __mapper_args__ = {"polymorphic_identity":"bar"}
 
404
 
 
405
        decl.DeferredReflection.prepare(testing.db)
 
406
        self._roundtrip()
 
407
 
 
408
    def test_add_subclass_column(self):
 
409
        class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
 
410
                    Base):
 
411
            __tablename__ = 'foo'
 
412
            __mapper_args__ = {"polymorphic_on":"type",
 
413
                        "polymorphic_identity":"foo"}
 
414
 
 
415
        class Bar(Foo):
 
416
            __tablename__ = 'bar'
 
417
            __mapper_args__ = {"polymorphic_identity":"bar"}
 
418
            bar_data = Column(String(30))
 
419
 
 
420
        decl.DeferredReflection.prepare(testing.db)
 
421
        self._roundtrip()
 
422
 
 
423
    def test_add_pk_column(self):
 
424
        class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
 
425
                    Base):
 
426
            __tablename__ = 'foo'
 
427
            __mapper_args__ = {"polymorphic_on":"type",
 
428
                        "polymorphic_identity":"foo"}
 
429
            id = Column(Integer, primary_key=True)
 
430
 
 
431
        class Bar(Foo):
 
432
            __tablename__ = 'bar'
 
433
            __mapper_args__ = {"polymorphic_identity":"bar"}
 
434
 
 
435
        decl.DeferredReflection.prepare(testing.db)
 
436
        self._roundtrip()
 
437
 
 
438
    def test_add_fk_pk_column(self):
 
439
        class Foo(decl.DeferredReflection, fixtures.ComparableEntity,
 
440
                    Base):
 
441
            __tablename__ = 'foo'
 
442
            __mapper_args__ = {"polymorphic_on":"type",
 
443
                        "polymorphic_identity":"foo"}
 
444
 
 
445
        class Bar(Foo):
 
446
            __tablename__ = 'bar'
 
447
            __mapper_args__ = {"polymorphic_identity":"bar"}
 
448
            id = Column(Integer, ForeignKey('foo.id'), primary_key=True)
 
449
 
 
450
        decl.DeferredReflection.prepare(testing.db)
 
451
        self._roundtrip()