← Back to branch summary

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

« back to all changes in this revision

Viewing changes to test/dialect/mssql/test_query.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/dialect/mssql/test_query.py
 
1
# -*- encoding: utf-8
 
2
from sqlalchemy.testing import eq_, engines
 
3
from sqlalchemy import *
 
4
from sqlalchemy.sql import table, column
 
5
from sqlalchemy.databases import mssql
 
6
from sqlalchemy.testing import fixtures, AssertsCompiledSQL
 
7
from sqlalchemy import testing
 
8
from sqlalchemy.util import ue
 
9
from sqlalchemy import util
 
10
 
 
11
 
 
12
 
 
13
class SchemaAliasingTest(fixtures.TestBase, AssertsCompiledSQL):
 
14
    """SQL server cannot reference schema-qualified tables in a SELECT statement, they
 
15
    must be aliased.
 
16
    """
 
17
    __dialect__ = mssql.dialect()
 
18
 
 
19
    def setup(self):
 
20
        metadata = MetaData()
 
21
        self.t1 = table('t1',
 
22
            column('a', Integer),
 
23
            column('b', String),
 
24
            column('c', String),
 
25
        )
 
26
        self.t2 = Table(
 
27
            't2', metadata,
 
28
            Column("a", Integer),
 
29
            Column("b", Integer),
 
30
            Column("c", Integer),
 
31
            schema = 'schema'
 
32
        )
 
33
 
 
34
    def test_result_map(self):
 
35
        s = self.t2.select()
 
36
        c = s.compile(dialect=self.__dialect__)
 
37
        assert self.t2.c.a in set(c.result_map['a'][1])
 
38
 
 
39
    def test_result_map_use_labels(self):
 
40
        s = self.t2.select(use_labels=True)
 
41
        c = s.compile(dialect=self.__dialect__)
 
42
        assert self.t2.c.a in set(c.result_map['schema_t2_a'][1])
 
43
 
 
44
    def test_straight_select(self):
 
45
        self.assert_compile(self.t2.select(),
 
46
            "SELECT t2_1.a, t2_1.b, t2_1.c FROM [schema].t2 AS t2_1"
 
47
        )
 
48
 
 
49
    def test_straight_select_use_labels(self):
 
50
        self.assert_compile(
 
51
            self.t2.select(use_labels=True),
 
52
            "SELECT t2_1.a AS schema_t2_a, t2_1.b AS schema_t2_b, "
 
53
            "t2_1.c AS schema_t2_c FROM [schema].t2 AS t2_1"
 
54
        )
 
55
 
 
56
    def test_join_to_schema(self):
 
57
        t1, t2 = self.t1, self.t2
 
58
        self.assert_compile(
 
59
            t1.join(t2, t1.c.a==t2.c.a).select(),
 
60
            "SELECT t1.a, t1.b, t1.c, t2_1.a, t2_1.b, t2_1.c FROM t1 "
 
61
            "JOIN [schema].t2 AS t2_1 ON t2_1.a = t1.a"
 
62
        )
 
63
 
 
64
    def test_union_schema_to_non(self):
 
65
        t1, t2 = self.t1, self.t2
 
66
        s = select([t2.c.a, t2.c.b]).apply_labels().\
 
67
                union(
 
68
                    select([t1.c.a, t1.c.b]).apply_labels()
 
69
                ).alias().select()
 
70
        self.assert_compile(
 
71
            s,
 
72
            "SELECT anon_1.schema_t2_a, anon_1.schema_t2_b FROM "
 
73
            "(SELECT t2_1.a AS schema_t2_a, t2_1.b AS schema_t2_b "
 
74
            "FROM [schema].t2 AS t2_1 UNION SELECT t1.a AS t1_a, "
 
75
            "t1.b AS t1_b FROM t1) AS anon_1"
 
76
        )
 
77
 
 
78
    def test_column_subquery_to_alias(self):
 
79
        a1 = self.t2.alias('a1')
 
80
        s = select([self.t2, select([a1.c.a]).as_scalar()])
 
81
        self.assert_compile(
 
82
            s,
 
83
            "SELECT t2_1.a, t2_1.b, t2_1.c, "
 
84
            "(SELECT a1.a FROM [schema].t2 AS a1) "
 
85
            "AS anon_1 FROM [schema].t2 AS t2_1"
 
86
 
 
87
        )
 
88
 
 
89
class IdentityInsertTest(fixtures.TestBase, AssertsCompiledSQL):
 
90
    __only_on__ = 'mssql'
 
91
    __dialect__ = mssql.MSDialect()
 
92
 
 
93
    @classmethod
 
94
    def setup_class(cls):
 
95
        global metadata, cattable
 
96
        metadata = MetaData(testing.db)
 
97
 
 
98
        cattable = Table('cattable', metadata,
 
99
            Column('id', Integer),
 
100
            Column('description', String(50)),
 
101
            PrimaryKeyConstraint('id', name='PK_cattable'),
 
102
        )
 
103
 
 
104
    def setup(self):
 
105
        metadata.create_all()
 
106
 
 
107
    def teardown(self):
 
108
        metadata.drop_all()
 
109
 
 
110
    def test_compiled(self):
 
111
        self.assert_compile(cattable.insert().values(id=9,
 
112
                            description='Python'),
 
113
                            'INSERT INTO cattable (id, description) '
 
114
                            'VALUES (:id, :description)')
 
115
 
 
116
    def test_execute(self):
 
117
        cattable.insert().values(id=9, description='Python').execute()
 
118
 
 
119
        cats = cattable.select().order_by(cattable.c.id).execute()
 
120
        eq_([(9, 'Python')], list(cats))
 
121
 
 
122
        result = cattable.insert().values(description='PHP').execute()
 
123
        eq_([10], result.inserted_primary_key)
 
124
        lastcat = cattable.select().order_by(desc(cattable.c.id)).execute()
 
125
        eq_((10, 'PHP'), lastcat.first())
 
126
 
 
127
    def test_executemany(self):
 
128
        cattable.insert().execute([{'id': 89, 'description': 'Python'},
 
129
                                  {'id': 8, 'description': 'Ruby'},
 
130
                                  {'id': 3, 'description': 'Perl'},
 
131
                                  {'id': 1, 'description': 'Java'}])
 
132
        cats = cattable.select().order_by(cattable.c.id).execute()
 
133
        eq_([(1, 'Java'), (3, 'Perl'), (8, 'Ruby'), (89, 'Python')],
 
134
            list(cats))
 
135
        cattable.insert().execute([{'description': 'PHP'},
 
136
                                  {'description': 'Smalltalk'}])
 
137
        lastcats = \
 
138
            cattable.select().order_by(desc(cattable.c.id)).limit(2).execute()
 
139
        eq_([(91, 'Smalltalk'), (90, 'PHP')], list(lastcats))
 
140
 
 
141
class QueryUnicodeTest(fixtures.TestBase):
 
142
 
 
143
    __only_on__ = 'mssql'
 
144
 
 
145
    def test_convert_unicode(self):
 
146
        meta = MetaData(testing.db)
 
147
        t1 = Table('unitest_table', meta, Column('id', Integer,
 
148
                   primary_key=True), Column('descr',
 
149
                   mssql.MSText(convert_unicode=True)))
 
150
        meta.create_all()
 
151
        con = testing.db.connect()
 
152
 
 
153
        # encode in UTF-8 (sting object) because this is the default
 
154
        # dialect encoding
 
155
 
 
156
        con.execute(ue("insert into unitest_table values ('bien u\
 
157
                    umang\xc3\xa9')").encode('UTF-8'))
 
158
        try:
 
159
            r = t1.select().execute().first()
 
160
            assert isinstance(r[1], util.text_type), \
 
161
                '%s is %s instead of unicode, working on %s' % (r[1],
 
162
                    type(r[1]), meta.bind)
 
163
        finally:
 
164
            meta.drop_all()
 
165
 
 
166
from sqlalchemy.testing.assertsql import ExactSQL
 
167
class QueryTest(testing.AssertsExecutionResults, fixtures.TestBase):
 
168
    __only_on__ = 'mssql'
 
169
 
 
170
    def test_fetchid_trigger(self):
 
171
        """
 
172
        Verify identity return value on inserting to a trigger table.
 
173
 
 
174
        MSSQL's OUTPUT INSERTED clause does not work for the
 
175
        case of a table having an identity (autoincrement)
 
176
        primary key column, and which also has a trigger configured
 
177
        to fire upon each insert and subsequently perform an
 
178
        insert into a different table.
 
179
 
 
180
        SQLALchemy's MSSQL dialect by default will attempt to
 
181
        use an OUTPUT_INSERTED clause, which in this case will
 
182
        raise the following error:
 
183
 
 
184
        ProgrammingError: (ProgrammingError) ('42000', 334,
 
185
        "[Microsoft][SQL Server Native Client 10.0][SQL Server]The
 
186
        target table 't1' of the DML statement cannot have any enabled
 
187
        triggers if the statement contains an OUTPUT clause without
 
188
        INTO clause.", 7748) 'INSERT INTO t1 (descr) OUTPUT inserted.id
 
189
        VALUES (?)' ('hello',)
 
190
 
 
191
        This test verifies a workaround, which is to rely on the
 
192
        older SCOPE_IDENTITY() call, which still works for this scenario.
 
193
        To enable the workaround, the Table must be instantiated
 
194
        with the init parameter 'implicit_returning = False'.
 
195
        """
 
196
 
 
197
        #todo: this same test needs to be tried in a multithreaded context
 
198
        #      with multiple threads inserting to the same table.
 
199
        #todo: check whether this error also occurs with clients other
 
200
        #      than the SQL Server Native Client. Maybe an assert_raises
 
201
        #      test should be written.
 
202
        meta = MetaData(testing.db)
 
203
        t1 = Table('t1', meta,
 
204
                Column('id', Integer, Sequence('fred', 100, 1),
 
205
                                primary_key=True),
 
206
                Column('descr', String(200)),
 
207
                # the following flag will prevent the
 
208
                # MSSQLCompiler.returning_clause from getting called,
 
209
                # though the ExecutionContext will still have a
 
210
                # _select_lastrowid, so the SELECT SCOPE_IDENTITY() will
 
211
                # hopefully be called instead.
 
212
                implicit_returning = False
 
213
                )
 
214
        t2 = Table('t2', meta,
 
215
                Column('id', Integer, Sequence('fred', 200, 1),
 
216
                                primary_key=True),
 
217
                Column('descr', String(200)))
 
218
        meta.create_all()
 
219
        con = testing.db.connect()
 
220
        con.execute("""create trigger paj on t1 for insert as
 
221
            insert into t2 (descr) select descr from inserted""")
 
222
 
 
223
        try:
 
224
            tr = con.begin()
 
225
            r = con.execute(t2.insert(), descr='hello')
 
226
            self.assert_(r.inserted_primary_key == [200])
 
227
            r = con.execute(t1.insert(), descr='hello')
 
228
            self.assert_(r.inserted_primary_key == [100])
 
229
 
 
230
        finally:
 
231
            tr.commit()
 
232
            con.execute("""drop trigger paj""")
 
233
            meta.drop_all()
 
234
 
 
235
    @testing.provide_metadata
 
236
    def test_disable_scope_identity(self):
 
237
        engine = engines.testing_engine(options={"use_scope_identity":False})
 
238
        metadata = self.metadata
 
239
        metadata.bind = engine
 
240
        t1 = Table('t1', metadata,
 
241
                Column('id', Integer, primary_key=True),
 
242
                implicit_returning=False
 
243
        )
 
244
        metadata.create_all()
 
245
 
 
246
        self.assert_sql_execution(
 
247
                testing.db,
 
248
                lambda: engine.execute(t1.insert()),
 
249
                ExactSQL("INSERT INTO t1 DEFAULT VALUES"),
 
250
                # we dont have an event for
 
251
                # "SELECT @@IDENTITY" part here.
 
252
                # this will be in 0.8 with #2459
 
253
        )
 
254
        assert not engine.dialect.use_scope_identity
 
255
 
 
256
    def test_insertid_schema(self):
 
257
        meta = MetaData(testing.db)
 
258
        con = testing.db.connect()
 
259
        con.execute('create schema paj')
 
260
        tbl = Table('test', meta,
 
261
                    Column('id', Integer, primary_key=True), schema='paj')
 
262
        tbl.create()
 
263
        try:
 
264
            tbl.insert().execute({'id':1})
 
265
        finally:
 
266
            tbl.drop()
 
267
            con.execute('drop schema paj')
 
268
 
 
269
    def test_returning_no_autoinc(self):
 
270
        meta = MetaData(testing.db)
 
271
        table = Table('t1', meta, Column('id', Integer,
 
272
                      primary_key=True), Column('data', String(50)))
 
273
        table.create()
 
274
        try:
 
275
            result = table.insert().values(id=1,
 
276
                    data=func.lower('SomeString'
 
277
                    )).returning(table.c.id, table.c.data).execute()
 
278
            eq_(result.fetchall(), [(1, 'somestring')])
 
279
        finally:
 
280
 
 
281
            # this will hang if the "SET IDENTITY_INSERT t1 OFF" occurs
 
282
            # before the result is fetched
 
283
 
 
284
            table.drop()
 
285
 
 
286
    def test_delete_schema(self):
 
287
        meta = MetaData(testing.db)
 
288
        con = testing.db.connect()
 
289
        con.execute('create schema paj')
 
290
        tbl = Table('test', meta, Column('id', Integer,
 
291
                    primary_key=True), schema='paj')
 
292
        tbl.create()
 
293
        try:
 
294
            tbl.insert().execute({'id': 1})
 
295
            tbl.delete(tbl.c.id == 1).execute()
 
296
        finally:
 
297
            tbl.drop()
 
298
            con.execute('drop schema paj')
 
299
 
 
300
    def test_insertid_reserved(self):
 
301
        meta = MetaData(testing.db)
 
302
        table = Table(
 
303
            'select', meta,
 
304
            Column('col', Integer, primary_key=True)
 
305
        )
 
306
        table.create()
 
307
 
 
308
        meta2 = MetaData(testing.db)
 
309
        try:
 
310
            table.insert().execute(col=7)
 
311
        finally:
 
312
            table.drop()
 
313
 
 
314
 
 
315
class Foo(object):
 
316
    def __init__(self, **kw):
 
317
        for k in kw:
 
318
            setattr(self, k, kw[k])
 
319
 
 
320
 
 
321
def full_text_search_missing():
 
322
    """Test if full text search is not implemented and return False if
 
323
    it is and True otherwise."""
 
324
 
 
325
    try:
 
326
        connection = testing.db.connect()
 
327
        try:
 
328
            connection.execute('CREATE FULLTEXT CATALOG Catalog AS '
 
329
                               'DEFAULT')
 
330
            return False
 
331
        except:
 
332
            return True
 
333
    finally:
 
334
        connection.close()
 
335
 
 
336
class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
 
337
 
 
338
    __only_on__ = 'mssql'
 
339
    __skip_if__ = full_text_search_missing,
 
340
 
 
341
    @classmethod
 
342
    def setup_class(cls):
 
343
        global metadata, cattable, matchtable
 
344
        metadata = MetaData(testing.db)
 
345
        cattable = Table('cattable', metadata, Column('id', Integer),
 
346
                         Column('description', String(50)),
 
347
                         PrimaryKeyConstraint('id', name='PK_cattable'))
 
348
        matchtable = Table(
 
349
            'matchtable',
 
350
            metadata,
 
351
            Column('id', Integer),
 
352
            Column('title', String(200)),
 
353
            Column('category_id', Integer, ForeignKey('cattable.id')),
 
354
            PrimaryKeyConstraint('id', name='PK_matchtable'),
 
355
            )
 
356
        DDL("""CREATE FULLTEXT INDEX
 
357
                       ON cattable (description)
 
358
                       KEY INDEX PK_cattable""").execute_at('after-create'
 
359
                , matchtable)
 
360
        DDL("""CREATE FULLTEXT INDEX
 
361
                       ON matchtable (title)
 
362
                       KEY INDEX PK_matchtable""").execute_at('after-create'
 
363
                , matchtable)
 
364
        metadata.create_all()
 
365
        cattable.insert().execute([{'id': 1, 'description': 'Python'},
 
366
                                  {'id': 2, 'description': 'Ruby'}])
 
367
        matchtable.insert().execute([{'id': 1, 'title'
 
368
                                    : 'Agile Web Development with Rails'
 
369
                                    , 'category_id': 2}, {'id': 2,
 
370
                                    'title': 'Dive Into Python',
 
371
                                    'category_id': 1}, {'id': 3, 'title'
 
372
                                    : "Programming Matz's Ruby",
 
373
                                    'category_id': 2}, {'id': 4, 'title'
 
374
                                    : 'The Definitive Guide to Django',
 
375
                                    'category_id': 1}, {'id': 5, 'title'
 
376
                                    : 'Python in a Nutshell',
 
377
                                    'category_id': 1}])
 
378
        DDL("WAITFOR DELAY '00:00:05'"
 
379
            ).execute(bind=engines.testing_engine())
 
380
 
 
381
    @classmethod
 
382
    def teardown_class(cls):
 
383
        metadata.drop_all()
 
384
        connection = testing.db.connect()
 
385
        connection.execute("DROP FULLTEXT CATALOG Catalog")
 
386
        connection.close()
 
387
 
 
388
    def test_expression(self):
 
389
        self.assert_compile(matchtable.c.title.match('somstr'),
 
390
                            'CONTAINS (matchtable.title, ?)')
 
391
 
 
392
    def test_simple_match(self):
 
393
        results = \
 
394
            matchtable.select().where(matchtable.c.title.match('python'
 
395
                )).order_by(matchtable.c.id).execute().fetchall()
 
396
        eq_([2, 5], [r.id for r in results])
 
397
 
 
398
    def test_simple_match_with_apostrophe(self):
 
399
        results = \
 
400
            matchtable.select().where(matchtable.c.title.match("Matz's"
 
401
                )).execute().fetchall()
 
402
        eq_([3], [r.id for r in results])
 
403
 
 
404
    def test_simple_prefix_match(self):
 
405
        results = \
 
406
            matchtable.select().where(matchtable.c.title.match('"nut*"'
 
407
                )).execute().fetchall()
 
408
        eq_([5], [r.id for r in results])
 
409
 
 
410
    def test_simple_inflectional_match(self):
 
411
        results = \
 
412
            matchtable.select().where(
 
413
                matchtable.c.title.match('FORMSOF(INFLECTIONAL, "dives")'
 
414
                )).execute().fetchall()
 
415
        eq_([2], [r.id for r in results])
 
416
 
 
417
    def test_or_match(self):
 
418
        results1 = \
 
419
            matchtable.select().where(or_(matchtable.c.title.match('nutshell'
 
420
                ), matchtable.c.title.match('ruby'
 
421
                ))).order_by(matchtable.c.id).execute().fetchall()
 
422
        eq_([3, 5], [r.id for r in results1])
 
423
        results2 = \
 
424
            matchtable.select().where(
 
425
                matchtable.c.title.match('nutshell OR ruby'
 
426
                )).order_by(matchtable.c.id).execute().fetchall()
 
427
        eq_([3, 5], [r.id for r in results2])
 
428
 
 
429
    def test_and_match(self):
 
430
        results1 = \
 
431
            matchtable.select().where(and_(matchtable.c.title.match('python'
 
432
                ), matchtable.c.title.match('nutshell'
 
433
                ))).execute().fetchall()
 
434
        eq_([5], [r.id for r in results1])
 
435
        results2 = \
 
436
            matchtable.select().where(
 
437
                matchtable.c.title.match('python AND nutshell'
 
438
                )).execute().fetchall()
 
439
        eq_([5], [r.id for r in results2])
 
440
 
 
441
    def test_match_across_joins(self):
 
442
        results = matchtable.select().where(and_(cattable.c.id
 
443
                == matchtable.c.category_id,
 
444
                or_(cattable.c.description.match('Ruby'),
 
445
                matchtable.c.title.match('nutshell'
 
446
                )))).order_by(matchtable.c.id).execute().fetchall()
 
447
        eq_([1, 3, 5], [r.id for r in results])
 
448
 
 
449