← Back to branch summary

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

« back to all changes in this revision

Viewing changes to test/ext/test_sqlsoup.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/test_sqlsoup.py
1
 
from sqlalchemy.ext import sqlsoup
2
 
from test.lib.testing import eq_, assert_raises, \
3
 
    assert_raises_message
4
 
from sqlalchemy import create_engine, or_, desc, select, func, exc, \
5
 
    Table, util, Column, Integer
6
 
from sqlalchemy.orm import scoped_session, sessionmaker
7
 
import datetime
8
 
from test.lib import fixtures
9
 
 
10
 
class SQLSoupTest(fixtures.TestBase):
11
 
 
12
 
    __requires__ = 'sqlite',
13
 
 
14
 
    @classmethod
15
 
    def setup_class(cls):
16
 
        global engine
17
 
        engine = create_engine('sqlite://')
18
 
        for sql in _ddl:
19
 
            engine.execute(sql)
20
 
 
21
 
    @classmethod
22
 
    def teardown_class(cls):
23
 
        engine.dispose()
24
 
 
25
 
    def setup(self):
26
 
        for sql in _data:
27
 
            engine.execute(sql)
28
 
 
29
 
    def teardown(self):
30
 
        sqlsoup.Session.remove()
31
 
        for sql in _teardown:
32
 
            engine.execute(sql)
33
 
 
34
 
    def test_map_to_attr_present(self):
35
 
        db = sqlsoup.SqlSoup(engine)
36
 
 
37
 
        users = db.users
38
 
        assert_raises_message(
39
 
            exc.InvalidRequestError,
40
 
            "Attribute 'users' is already mapped",
41
 
            db.map_to, 'users', tablename='users'
42
 
        )
43
 
 
44
 
    def test_map_to_table_not_string(self):
45
 
        db = sqlsoup.SqlSoup(engine)
46
 
 
47
 
        table = Table('users', db._metadata, Column('id', Integer, primary_key=True))
48
 
        assert_raises_message(
49
 
            exc.ArgumentError,
50
 
            "'tablename' argument must be a string.",
51
 
            db.map_to, 'users', tablename=table
52
 
        )
53
 
 
54
 
    def test_map_to_table_or_selectable(self):
55
 
        db = sqlsoup.SqlSoup(engine)
56
 
 
57
 
        table = Table('users', db._metadata, Column('id', Integer, primary_key=True))
58
 
        assert_raises_message(
59
 
            exc.ArgumentError,
60
 
            "'tablename' and 'selectable' arguments are mutually exclusive",
61
 
            db.map_to, 'users', tablename='users', selectable=table
62
 
        )
63
 
 
64
 
    def test_map_to_no_pk_selectable(self):
65
 
        db = sqlsoup.SqlSoup(engine)
66
 
 
67
 
        table = Table('users', db._metadata, Column('id', Integer))
68
 
        assert_raises_message(
69
 
            sqlsoup.PKNotFoundError,
70
 
            "table 'users' does not have a primary ",
71
 
            db.map_to, 'users', selectable=table
72
 
        )
73
 
    def test_map_to_invalid_schema(self):
74
 
        db = sqlsoup.SqlSoup(engine)
75
 
 
76
 
        table = Table('users', db._metadata, Column('id', Integer))
77
 
        assert_raises_message(
78
 
            exc.ArgumentError,
79
 
            "'tablename' argument is required when "
80
 
                                "using 'schema'.",
81
 
            db.map_to, 'users', selectable=table, schema='hoho'
82
 
        )
83
 
    def test_map_to_nothing(self):
84
 
        db = sqlsoup.SqlSoup(engine)
85
 
 
86
 
        assert_raises_message(
87
 
            exc.ArgumentError,
88
 
            "'tablename' or 'selectable' argument is "
89
 
                                    "required.",
90
 
            db.map_to, 'users',
91
 
        )
92
 
 
93
 
    def test_map_to_string_not_selectable(self):
94
 
        db = sqlsoup.SqlSoup(engine)
95
 
 
96
 
        assert_raises_message(
97
 
            exc.ArgumentError,
98
 
            "'selectable' argument must be a "
99
 
                                    "table, select, join, or other "
100
 
                                    "selectable construct.",
101
 
            db.map_to, 'users', selectable='users'
102
 
        )
103
 
 
104
 
    def test_bad_names(self):
105
 
        db = sqlsoup.SqlSoup(engine)
106
 
 
107
 
        # print db.bad_names.c.id
108
 
 
109
 
        print db.bad_names.c.query
110
 
 
111
 
    def test_load(self):
112
 
        db = sqlsoup.SqlSoup(engine)
113
 
        MappedUsers = db.users
114
 
        users = db.users.all()
115
 
        users.sort()
116
 
        eq_(users, [MappedUsers(name=u'Joe Student',
117
 
            email=u'student@example.edu', password=u'student',
118
 
            classname=None, admin=0),
119
 
            MappedUsers(name=u'Bhargan Basepair',
120
 
            email=u'basepair@example.edu', password=u'basepair',
121
 
            classname=None, admin=1)])
122
 
 
123
 
    def test_order_by(self):
124
 
        db = sqlsoup.SqlSoup(engine)
125
 
        MappedUsers = db.users
126
 
        users = db.users.order_by(db.users.name).all()
127
 
        eq_(users, [MappedUsers(name=u'Bhargan Basepair',
128
 
            email=u'basepair@example.edu', password=u'basepair',
129
 
            classname=None, admin=1), MappedUsers(name=u'Joe Student',
130
 
            email=u'student@example.edu', password=u'student',
131
 
            classname=None, admin=0)])
132
 
 
133
 
    def test_whereclause(self):
134
 
        db = sqlsoup.SqlSoup(engine)
135
 
        MappedUsers = db.users
136
 
        where = or_(db.users.name == 'Bhargan Basepair', db.users.email
137
 
                    == 'student@example.edu')
138
 
        users = \
139
 
            db.users.filter(where).order_by(desc(db.users.name)).all()
140
 
        eq_(users, [MappedUsers(name=u'Joe Student',
141
 
            email=u'student@example.edu', password=u'student',
142
 
            classname=None, admin=0),
143
 
            MappedUsers(name=u'Bhargan Basepair',
144
 
            email=u'basepair@example.edu', password=u'basepair',
145
 
            classname=None, admin=1)])
146
 
 
147
 
    def test_first(self):
148
 
        db = sqlsoup.SqlSoup(engine)
149
 
        MappedUsers = db.users
150
 
        user = db.users.filter(db.users.name == 'Bhargan Basepair'
151
 
                               ).one()
152
 
        eq_(user, MappedUsers(name=u'Bhargan Basepair',
153
 
            email=u'basepair@example.edu', password=u'basepair',
154
 
            classname=None, admin=1))
155
 
        db.rollback()
156
 
        user = db.users.get('Bhargan Basepair')
157
 
        eq_(user, MappedUsers(name=u'Bhargan Basepair',
158
 
            email=u'basepair@example.edu', password=u'basepair',
159
 
            classname=None, admin=1))
160
 
        db.rollback()
161
 
        user = db.users.filter_by(name='Bhargan Basepair').one()
162
 
        eq_(user, MappedUsers(name=u'Bhargan Basepair',
163
 
            email=u'basepair@example.edu', password=u'basepair',
164
 
            classname=None, admin=1))
165
 
        db.rollback()
166
 
 
167
 
    def test_crud(self):
168
 
 
169
 
        # note we're testing autoflush here too...
170
 
 
171
 
        db = sqlsoup.SqlSoup(engine)
172
 
        MappedLoans = db.loans
173
 
        user = db.users.filter_by(name='Bhargan Basepair').one()
174
 
        book_id = db.books.filter_by(title='Regional Variation in Moss'
175
 
                ).first().id
176
 
        loan_insert = db.loans.insert(book_id=book_id,
177
 
                user_name=user.name)
178
 
        loan = db.loans.filter_by(book_id=2,
179
 
                                  user_name='Bhargan Basepair').one()
180
 
        eq_(loan, loan_insert)
181
 
        l2 = MappedLoans(book_id=2, user_name=u'Bhargan Basepair',
182
 
                         loan_date=loan.loan_date)
183
 
        eq_(loan, l2)
184
 
        db.expunge(l2)
185
 
        db.delete(loan)
186
 
        loan = db.loans.filter_by(book_id=2,
187
 
                                  user_name='Bhargan Basepair').first()
188
 
        assert loan is None
189
 
 
190
 
    def test_cls_crud(self):
191
 
        db = sqlsoup.SqlSoup(engine)
192
 
        MappedUsers = db.users
193
 
        db.users.filter_by(name='Bhargan Basepair'
194
 
                           ).update(dict(name='Some New Name'))
195
 
        u1 = db.users.filter_by(name='Some New Name').one()
196
 
        eq_(u1, MappedUsers(name=u'Some New Name',
197
 
            email=u'basepair@example.edu', password=u'basepair',
198
 
            classname=None, admin=1))
199
 
 
200
 
    def test_map_table(self):
201
 
        db = sqlsoup.SqlSoup(engine)
202
 
        users = Table('users', db._metadata, autoload=True)
203
 
        MappedUsers = db.map(users)
204
 
        users = MappedUsers.order_by(db.users.name).all()
205
 
        eq_(users, [MappedUsers(name=u'Bhargan Basepair',
206
 
            email=u'basepair@example.edu', password=u'basepair',
207
 
            classname=None, admin=1), MappedUsers(name=u'Joe Student',
208
 
            email=u'student@example.edu', password=u'student',
209
 
            classname=None, admin=0)])
210
 
 
211
 
    def test_mapped_join(self):
212
 
        db = sqlsoup.SqlSoup(engine)
213
 
        join1 = MappedJoin = db.join(db.users, db.loans, isouter=True)
214
 
        mj = join1.filter_by(name='Joe Student').all()
215
 
        eq_(mj, [MappedJoin(
216
 
            name=u'Joe Student',
217
 
            email=u'student@example.edu',
218
 
            password=u'student',
219
 
            classname=None,
220
 
            admin=0,
221
 
            book_id=1,
222
 
            user_name=u'Joe Student',
223
 
            loan_date=datetime.datetime(2006, 7, 12, 0, 0),
224
 
            )])
225
 
        db.rollback()
226
 
        join2 = MappedJoin = db.join(join1, db.books)
227
 
        mj = join2.all()
228
 
        eq_(mj, [MappedJoin(
229
 
            name=u'Joe Student',
230
 
            email=u'student@example.edu',
231
 
            password=u'student',
232
 
            classname=None,
233
 
            admin=0,
234
 
            book_id=1,
235
 
            user_name=u'Joe Student',
236
 
            loan_date=datetime.datetime(2006, 7, 12, 0, 0),
237
 
            id=1,
238
 
            title=u'Mustards I Have Known',
239
 
            published_year=u'1989',
240
 
            authors=u'Jones',
241
 
            )])
242
 
        eq_(db.with_labels(join1).c.keys(), [
243
 
            u'users_name',
244
 
            u'users_email',
245
 
            u'users_password',
246
 
            u'users_classname',
247
 
            u'users_admin',
248
 
            u'loans_book_id',
249
 
            u'loans_user_name',
250
 
            u'loans_loan_date',
251
 
            ])
252
 
        labeled_loans = db.with_labels(db.loans)
253
 
        eq_(db.join(db.users, labeled_loans, isouter=True).c.keys(), [
254
 
            u'name',
255
 
            u'email',
256
 
            u'password',
257
 
            u'classname',
258
 
            u'admin',
259
 
            u'loans_book_id',
260
 
            u'loans_user_name',
261
 
            u'loans_loan_date',
262
 
            ])
263
 
 
264
 
    def test_relations(self):
265
 
        db = sqlsoup.SqlSoup(engine)
266
 
        db.users.relate('loans', db.loans)
267
 
        MappedLoans = db.loans
268
 
        MappedUsers = db.users
269
 
        eq_(db.users.get('Joe Student').loans, [MappedLoans(book_id=1,
270
 
            user_name=u'Joe Student', loan_date=datetime.datetime(2006,
271
 
            7, 12, 0, 0))])
272
 
        db.rollback()
273
 
        eq_(db.users.filter(~db.users.loans.any()).all(),
274
 
            [MappedUsers(name=u'Bhargan Basepair',
275
 
            email='basepair@example.edu', password=u'basepair',
276
 
            classname=None, admin=1)])
277
 
        db.rollback()
278
 
        del db._cache['users']
279
 
        db.users.relate('loans', db.loans, order_by=db.loans.loan_date,
280
 
                        cascade='all, delete-orphan')
281
 
 
282
 
    def test_relate_m2o(self):
283
 
        db = sqlsoup.SqlSoup(engine)
284
 
        db.loans.relate('user', db.users)
285
 
        u1 = db.users.filter(db.users.c.name=='Joe Student').one()
286
 
        eq_(db.loans.first().user, u1)
287
 
 
288
 
    def test_explicit_session(self):
289
 
        Session = scoped_session(sessionmaker())
290
 
        db = sqlsoup.SqlSoup(engine, session=Session)
291
 
        try:
292
 
            MappedUsers = db.users
293
 
            sess = Session()
294
 
            assert db.users._query.session is db.users.session is sess
295
 
            row = db.users.insert(name='new name', email='new email')
296
 
            assert row in sess
297
 
        finally:
298
 
            sess.rollback()
299
 
            sess.close()
300
 
 
301
 
    def test_selectable(self):
302
 
        db = sqlsoup.SqlSoup(engine)
303
 
        MappedBooks = db.books
304
 
        b = db.books._table
305
 
        s = select([b.c.published_year, func.count('*').label('n')],
306
 
                   from_obj=[b], group_by=[b.c.published_year])
307
 
        s = s.alias('years_with_count')
308
 
        years_with_count = db.map(s, primary_key=[s.c.published_year])
309
 
        eq_(years_with_count.filter_by(published_year='1989').all(),
310
 
            [MappedBooks(published_year=u'1989', n=1)])
311
 
 
312
 
    def test_raw_sql(self):
313
 
        db = sqlsoup.SqlSoup(engine)
314
 
        rp = db.execute('select name, email from users order by name')
315
 
        eq_(rp.fetchall(), [('Bhargan Basepair', 'basepair@example.edu'
316
 
            ), ('Joe Student', 'student@example.edu')])
317
 
 
318
 
        # test that execute() shares the same transactional context as
319
 
        # the session
320
 
 
321
 
        db.execute("update users set email='foo bar'")
322
 
        eq_(db.execute('select distinct email from users').fetchall(),
323
 
            [('foo bar', )])
324
 
        db.rollback()
325
 
        eq_(db.execute('select distinct email from users').fetchall(),
326
 
            [(u'basepair@example.edu', ), (u'student@example.edu', )])
327
 
 
328
 
    def test_connection(self):
329
 
        db = sqlsoup.SqlSoup(engine)
330
 
        conn = db.connection()
331
 
        rp = conn.execute('select name, email from users order by name')
332
 
        eq_(rp.fetchall(), [('Bhargan Basepair', 'basepair@example.edu'
333
 
            ), ('Joe Student', 'student@example.edu')])
334
 
 
335
 
    def test_entity(self):
336
 
        db = sqlsoup.SqlSoup(engine)
337
 
        tablename = 'loans'
338
 
        eq_(db.entity(tablename), db.loans)
339
 
 
340
 
    def test_entity_with_different_base(self):
341
 
        class subclass(object):
342
 
            pass
343
 
 
344
 
        db = sqlsoup.SqlSoup(engine, base=subclass)
345
 
        assert issubclass(db.entity('loans'), subclass)
346
 
 
347
 
    def test_filter_by_order_by(self):
348
 
        db = sqlsoup.SqlSoup(engine)
349
 
        MappedUsers = db.users
350
 
        users = \
351
 
            db.users.filter_by(classname=None).order_by(db.users.name).all()
352
 
        eq_(users, [MappedUsers(name=u'Bhargan Basepair',
353
 
            email=u'basepair@example.edu', password=u'basepair',
354
 
            classname=None, admin=1), MappedUsers(name=u'Joe Student',
355
 
            email=u'student@example.edu', password=u'student',
356
 
            classname=None, admin=0)])
357
 
 
358
 
    def test_no_pk_reflected(self):
359
 
        db = sqlsoup.SqlSoup(engine)
360
 
        assert_raises(sqlsoup.PKNotFoundError, getattr, db, 'nopk')
361
 
 
362
 
    def test_nosuchtable(self):
363
 
        db = sqlsoup.SqlSoup(engine)
364
 
        assert_raises(exc.NoSuchTableError, getattr, db, 'nosuchtable')
365
 
 
366
 
    def test_dont_persist_alias(self):
367
 
        db = sqlsoup.SqlSoup(engine)
368
 
        MappedBooks = db.books
369
 
        b = db.books._table
370
 
        s = select([b.c.published_year, func.count('*').label('n')],
371
 
                   from_obj=[b], group_by=[b.c.published_year])
372
 
        s = s.alias('years_with_count')
373
 
        years_with_count = db.map(s, primary_key=[s.c.published_year])
374
 
        assert_raises(exc.InvalidRequestError, years_with_count.insert,
375
 
                      published_year='2007', n=1)
376
 
 
377
 
    def test_clear(self):
378
 
        db = sqlsoup.SqlSoup(engine)
379
 
        eq_(db.loans.count(), 1)
380
 
        _ = db.loans.insert(book_id=1, user_name='Bhargan Basepair')
381
 
        db.expunge_all()
382
 
        db.flush()
383
 
        eq_(db.loans.count(), 1)
384
 
 
385
 
 
386
 
_ddl = \
387
 
    u"""
388
 
CREATE TABLE books (
389
 
    id                   integer PRIMARY KEY, -- auto-increments in sqlite
390
 
    title                text NOT NULL,
391
 
    published_year       char(4) NOT NULL,
392
 
    authors              text NOT NULL
393
 
);
394
 
 
395
 
CREATE TABLE users (
396
 
    name                 varchar(32) PRIMARY KEY,
397
 
    email                varchar(128) NOT NULL,
398
 
    password             varchar(128) NOT NULL,
399
 
    classname            text,
400
 
    admin                int NOT NULL -- 0 = false
401
 
);
402
 
 
403
 
CREATE TABLE loans (
404
 
    book_id              int PRIMARY KEY REFERENCES books(id),
405
 
    user_name            varchar(32) references users(name)
406
 
        ON DELETE SET NULL ON UPDATE CASCADE,
407
 
    loan_date            datetime DEFAULT current_timestamp
408
 
);
409
 
 
410
 
CREATE TABLE nopk (
411
 
    i                    int
412
 
);
413
 
 
414
 
CREATE TABLE bad_names (
415
 
   id int primary key,
416
 
   query  varchar(100)
417
 
)
418
 
""".split(';'
419
 
        )
420
 
_data = \
421
 
    """
422
 
insert into users(name, email, password, admin)
423
 
values('Bhargan Basepair', 'basepair@example.edu', 'basepair', 1);
424
 
insert into users(name, email, password, admin)
425
 
values('Joe Student', 'student@example.edu', 'student', 0);
426
 
 
427
 
insert into books(title, published_year, authors)
428
 
values('Mustards I Have Known', '1989', 'Jones');
429
 
insert into books(title, published_year, authors)
430
 
values('Regional Variation in Moss', '1971', 'Flim and Flam');
431
 
 
432
 
insert into loans(book_id, user_name, loan_date)
433
 
values (
434
 
    (select min(id) from books),
435
 
    (select name from users where name like 'Joe%'),
436
 
    '2006-07-12 0:0:0')
437
 
;
438
 
""".split(';'
439
 
        )
440
 
_teardown = \
441
 
    """
442
 
delete from loans;
443
 
delete from books;
444
 
delete from users;
445
 
delete from nopk;
446
 
""".split(';'
447
 
        )