← Back to branch summary

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

« back to all changes in this revision

Viewing changes to test/dialect/mysql/test_compiler.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/mysql/test_compiler.py
 
1
# coding: utf-8
 
2
 
 
3
from sqlalchemy.testing import eq_, assert_raises_message
 
4
from sqlalchemy import *
 
5
from sqlalchemy import sql, exc, schema, types as sqltypes
 
6
from sqlalchemy.dialects.mysql import base as mysql
 
7
from sqlalchemy.testing import fixtures, AssertsCompiledSQL
 
8
from sqlalchemy import testing
 
9
 
 
10
class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
 
11
 
 
12
    __dialect__ = mysql.dialect()
 
13
 
 
14
    def test_reserved_words(self):
 
15
        table = Table("mysql_table", MetaData(),
 
16
            Column("col1", Integer),
 
17
            Column("master_ssl_verify_server_cert", Integer))
 
18
        x = select([table.c.col1, table.c.master_ssl_verify_server_cert])
 
19
 
 
20
        self.assert_compile(x,
 
21
            "SELECT mysql_table.col1, "
 
22
            "mysql_table.`master_ssl_verify_server_cert` FROM mysql_table")
 
23
 
 
24
    def test_create_index_simple(self):
 
25
        m = MetaData()
 
26
        tbl = Table('testtbl', m, Column('data', String(255)))
 
27
        idx = Index('test_idx1', tbl.c.data)
 
28
 
 
29
        self.assert_compile(schema.CreateIndex(idx),
 
30
            'CREATE INDEX test_idx1 ON testtbl (data)')
 
31
 
 
32
    def test_create_index_with_length(self):
 
33
        m = MetaData()
 
34
        tbl = Table('testtbl', m, Column('data', String(255)))
 
35
        idx1 = Index('test_idx1', tbl.c.data, mysql_length=10)
 
36
        idx2 = Index('test_idx2', tbl.c.data, mysql_length=5)
 
37
 
 
38
        self.assert_compile(schema.CreateIndex(idx1),
 
39
            'CREATE INDEX test_idx1 ON testtbl (data(10))')
 
40
        self.assert_compile(schema.CreateIndex(idx2),
 
41
            'CREATE INDEX test_idx2 ON testtbl (data(5))')
 
42
 
 
43
    def test_create_composite_index_with_length(self):
 
44
        m = MetaData()
 
45
        tbl = Table('testtbl', m,
 
46
                    Column('a', String(255)),
 
47
                    Column('b', String(255)))
 
48
 
 
49
        idx1 = Index('test_idx1', tbl.c.a, tbl.c.b,
 
50
                     mysql_length={'a': 10, 'b': 20})
 
51
        idx2 = Index('test_idx2', tbl.c.a, tbl.c.b,
 
52
                     mysql_length={'a': 15})
 
53
        idx3 = Index('test_idx3', tbl.c.a, tbl.c.b,
 
54
                     mysql_length=30)
 
55
 
 
56
        self.assert_compile(
 
57
            schema.CreateIndex(idx1),
 
58
            'CREATE INDEX test_idx1 ON testtbl (a(10), b(20))'
 
59
        )
 
60
        self.assert_compile(
 
61
            schema.CreateIndex(idx2),
 
62
            'CREATE INDEX test_idx2 ON testtbl (a(15), b)'
 
63
        )
 
64
        self.assert_compile(
 
65
            schema.CreateIndex(idx3),
 
66
            'CREATE INDEX test_idx3 ON testtbl (a(30), b(30))'
 
67
        )
 
68
 
 
69
    def test_create_index_with_using(self):
 
70
        m = MetaData()
 
71
        tbl = Table('testtbl', m, Column('data', String(255)))
 
72
        idx1 = Index('test_idx1', tbl.c.data, mysql_using='btree')
 
73
        idx2 = Index('test_idx2', tbl.c.data, mysql_using='hash')
 
74
 
 
75
        self.assert_compile(schema.CreateIndex(idx1),
 
76
            'CREATE INDEX test_idx1 ON testtbl (data) USING btree')
 
77
        self.assert_compile(schema.CreateIndex(idx2),
 
78
            'CREATE INDEX test_idx2 ON testtbl (data) USING hash')
 
79
 
 
80
    def test_create_pk_plain(self):
 
81
        m = MetaData()
 
82
        tbl = Table('testtbl', m, Column('data', String(255)),
 
83
            PrimaryKeyConstraint('data'))
 
84
 
 
85
        self.assert_compile(schema.CreateTable(tbl),
 
86
            "CREATE TABLE testtbl (data VARCHAR(255), PRIMARY KEY (data))")
 
87
 
 
88
    def test_create_pk_with_using(self):
 
89
        m = MetaData()
 
90
        tbl = Table('testtbl', m, Column('data', String(255)),
 
91
            PrimaryKeyConstraint('data', mysql_using='btree'))
 
92
 
 
93
        self.assert_compile(schema.CreateTable(tbl),
 
94
            "CREATE TABLE testtbl (data VARCHAR(255), "
 
95
            "PRIMARY KEY (data) USING btree)")
 
96
 
 
97
    def test_skip_deferrable_kw(self):
 
98
        m = MetaData()
 
99
        t1 = Table('t1', m, Column('id', Integer, primary_key=True))
 
100
        t2 = Table('t2', m, Column('id', Integer,
 
101
                        ForeignKey('t1.id', deferrable=True),
 
102
                            primary_key=True))
 
103
 
 
104
        self.assert_compile(
 
105
            schema.CreateTable(t2),
 
106
            "CREATE TABLE t2 (id INTEGER NOT NULL, "
 
107
            "PRIMARY KEY (id), FOREIGN KEY(id) REFERENCES t1 (id))"
 
108
        )
 
109
 
 
110
class SQLTest(fixtures.TestBase, AssertsCompiledSQL):
 
111
    """Tests MySQL-dialect specific compilation."""
 
112
 
 
113
    __dialect__ = mysql.dialect()
 
114
 
 
115
    def test_precolumns(self):
 
116
        dialect = self.__dialect__
 
117
 
 
118
        def gen(distinct=None, prefixes=None):
 
119
            kw = {}
 
120
            if distinct is not None:
 
121
                kw['distinct'] = distinct
 
122
            if prefixes is not None:
 
123
                kw['prefixes'] = prefixes
 
124
            return str(select(['q'], **kw).compile(dialect=dialect))
 
125
 
 
126
        eq_(gen(None), 'SELECT q')
 
127
        eq_(gen(True), 'SELECT DISTINCT q')
 
128
 
 
129
        eq_(gen(prefixes=['ALL']), 'SELECT ALL q')
 
130
        eq_(gen(prefixes=['DISTINCTROW']),
 
131
                'SELECT DISTINCTROW q')
 
132
 
 
133
        # Interaction with MySQL prefix extensions
 
134
        eq_(
 
135
            gen(None, ['straight_join']),
 
136
            'SELECT straight_join q')
 
137
        eq_(
 
138
            gen(False, ['HIGH_PRIORITY', 'SQL_SMALL_RESULT', 'ALL']),
 
139
            'SELECT HIGH_PRIORITY SQL_SMALL_RESULT ALL q')
 
140
        eq_(
 
141
            gen(True, ['high_priority', sql.text('sql_cache')]),
 
142
            'SELECT high_priority sql_cache DISTINCT q')
 
143
 
 
144
    @testing.uses_deprecated
 
145
    def test_deprecated_distinct(self):
 
146
        dialect = self.__dialect__
 
147
 
 
148
        self.assert_compile(
 
149
            select(['q'], distinct='ALL'),
 
150
            'SELECT ALL q',
 
151
        )
 
152
 
 
153
        self.assert_compile(
 
154
            select(['q'], distinct='distinctROW'),
 
155
            'SELECT DISTINCTROW q',
 
156
        )
 
157
 
 
158
        self.assert_compile(
 
159
            select(['q'], distinct='ALL',
 
160
                    prefixes=['HIGH_PRIORITY', 'SQL_SMALL_RESULT']),
 
161
            'SELECT HIGH_PRIORITY SQL_SMALL_RESULT ALL q'
 
162
        )
 
163
 
 
164
    def test_backslash_escaping(self):
 
165
        self.assert_compile(
 
166
            sql.column('foo').like('bar', escape='\\'),
 
167
            "foo LIKE %s ESCAPE '\\\\'"
 
168
        )
 
169
 
 
170
        dialect = mysql.dialect()
 
171
        dialect._backslash_escapes=False
 
172
        self.assert_compile(
 
173
            sql.column('foo').like('bar', escape='\\'),
 
174
            "foo LIKE %s ESCAPE '\\'",
 
175
            dialect=dialect
 
176
        )
 
177
 
 
178
    def test_limit(self):
 
179
        t = sql.table('t', sql.column('col1'), sql.column('col2'))
 
180
 
 
181
        self.assert_compile(
 
182
            select([t]).limit(10).offset(20),
 
183
            "SELECT t.col1, t.col2 FROM t  LIMIT %s, %s",
 
184
            {'param_1':20, 'param_2':10}
 
185
            )
 
186
        self.assert_compile(
 
187
            select([t]).limit(10),
 
188
            "SELECT t.col1, t.col2 FROM t  LIMIT %s",
 
189
            {'param_1':10})
 
190
 
 
191
        self.assert_compile(
 
192
            select([t]).offset(10),
 
193
            "SELECT t.col1, t.col2 FROM t  LIMIT %s, 18446744073709551615",
 
194
            {'param_1':10}
 
195
            )
 
196
 
 
197
    def test_varchar_raise(self):
 
198
        for type_ in (
 
199
            String,
 
200
            VARCHAR,
 
201
            String(),
 
202
            VARCHAR(),
 
203
            NVARCHAR(),
 
204
            Unicode,
 
205
            Unicode(),
 
206
        ):
 
207
            type_ = sqltypes.to_instance(type_)
 
208
            assert_raises_message(
 
209
                exc.CompileError,
 
210
                "VARCHAR requires a length on dialect mysql",
 
211
                type_.compile,
 
212
                dialect=mysql.dialect()
 
213
            )
 
214
 
 
215
            t1 = Table('sometable', MetaData(),
 
216
                Column('somecolumn', type_)
 
217
            )
 
218
            assert_raises_message(
 
219
                exc.CompileError,
 
220
                r"\(in table 'sometable', column 'somecolumn'\)\: "
 
221
                r"(?:N)?VARCHAR requires a length on dialect mysql",
 
222
                schema.CreateTable(t1).compile,
 
223
                dialect=mysql.dialect()
 
224
            )
 
225
 
 
226
    def test_update_limit(self):
 
227
        t = sql.table('t', sql.column('col1'), sql.column('col2'))
 
228
 
 
229
        self.assert_compile(
 
230
            t.update(values={'col1':123}),
 
231
            "UPDATE t SET col1=%s"
 
232
            )
 
233
        self.assert_compile(
 
234
            t.update(values={'col1':123}, mysql_limit=5),
 
235
            "UPDATE t SET col1=%s LIMIT 5"
 
236
            )
 
237
        self.assert_compile(
 
238
            t.update(values={'col1':123}, mysql_limit=None),
 
239
            "UPDATE t SET col1=%s"
 
240
            )
 
241
        self.assert_compile(
 
242
            t.update(t.c.col2==456, values={'col1':123}, mysql_limit=1),
 
243
            "UPDATE t SET col1=%s WHERE t.col2 = %s LIMIT 1"
 
244
            )
 
245
 
 
246
    def test_utc_timestamp(self):
 
247
        self.assert_compile(func.utc_timestamp(), "UTC_TIMESTAMP")
 
248
 
 
249
    def test_sysdate(self):
 
250
        self.assert_compile(func.sysdate(), "SYSDATE()")
 
251
 
 
252
    def test_cast(self):
 
253
        t = sql.table('t', sql.column('col'))
 
254
        m = mysql
 
255
 
 
256
        specs = [
 
257
            (Integer, "CAST(t.col AS SIGNED INTEGER)"),
 
258
            (INT, "CAST(t.col AS SIGNED INTEGER)"),
 
259
            (m.MSInteger, "CAST(t.col AS SIGNED INTEGER)"),
 
260
            (m.MSInteger(unsigned=True), "CAST(t.col AS UNSIGNED INTEGER)"),
 
261
            (SmallInteger, "CAST(t.col AS SIGNED INTEGER)"),
 
262
            (m.MSSmallInteger, "CAST(t.col AS SIGNED INTEGER)"),
 
263
            (m.MSTinyInteger, "CAST(t.col AS SIGNED INTEGER)"),
 
264
            # 'SIGNED INTEGER' is a bigint, so this is ok.
 
265
            (m.MSBigInteger, "CAST(t.col AS SIGNED INTEGER)"),
 
266
            (m.MSBigInteger(unsigned=False), "CAST(t.col AS SIGNED INTEGER)"),
 
267
            (m.MSBigInteger(unsigned=True),
 
268
                            "CAST(t.col AS UNSIGNED INTEGER)"),
 
269
            (m.MSBit, "t.col"),
 
270
 
 
271
            # this is kind of sucky.  thank you default arguments!
 
272
            (NUMERIC, "CAST(t.col AS DECIMAL)"),
 
273
            (DECIMAL, "CAST(t.col AS DECIMAL)"),
 
274
            (Numeric, "CAST(t.col AS DECIMAL)"),
 
275
            (m.MSNumeric, "CAST(t.col AS DECIMAL)"),
 
276
            (m.MSDecimal, "CAST(t.col AS DECIMAL)"),
 
277
 
 
278
            (FLOAT, "t.col"),
 
279
            (Float, "t.col"),
 
280
            (m.MSFloat, "t.col"),
 
281
            (m.MSDouble, "t.col"),
 
282
            (m.MSReal, "t.col"),
 
283
 
 
284
            (TIMESTAMP, "CAST(t.col AS DATETIME)"),
 
285
            (DATETIME, "CAST(t.col AS DATETIME)"),
 
286
            (DATE, "CAST(t.col AS DATE)"),
 
287
            (TIME, "CAST(t.col AS TIME)"),
 
288
            (DateTime, "CAST(t.col AS DATETIME)"),
 
289
            (Date, "CAST(t.col AS DATE)"),
 
290
            (Time, "CAST(t.col AS TIME)"),
 
291
            (DateTime, "CAST(t.col AS DATETIME)"),
 
292
            (Date, "CAST(t.col AS DATE)"),
 
293
            (m.MSTime, "CAST(t.col AS TIME)"),
 
294
            (m.MSTimeStamp, "CAST(t.col AS DATETIME)"),
 
295
            (m.MSYear, "t.col"),
 
296
            (m.MSYear(2), "t.col"),
 
297
            (Interval, "t.col"),
 
298
 
 
299
            (String, "CAST(t.col AS CHAR)"),
 
300
            (Unicode, "CAST(t.col AS CHAR)"),
 
301
            (UnicodeText, "CAST(t.col AS CHAR)"),
 
302
            (VARCHAR, "CAST(t.col AS CHAR)"),
 
303
            (NCHAR, "CAST(t.col AS CHAR)"),
 
304
            (CHAR, "CAST(t.col AS CHAR)"),
 
305
            (CLOB, "CAST(t.col AS CHAR)"),
 
306
            (TEXT, "CAST(t.col AS CHAR)"),
 
307
            (String(32), "CAST(t.col AS CHAR(32))"),
 
308
            (Unicode(32), "CAST(t.col AS CHAR(32))"),
 
309
            (CHAR(32), "CAST(t.col AS CHAR(32))"),
 
310
            (m.MSString, "CAST(t.col AS CHAR)"),
 
311
            (m.MSText, "CAST(t.col AS CHAR)"),
 
312
            (m.MSTinyText, "CAST(t.col AS CHAR)"),
 
313
            (m.MSMediumText, "CAST(t.col AS CHAR)"),
 
314
            (m.MSLongText, "CAST(t.col AS CHAR)"),
 
315
            (m.MSNChar, "CAST(t.col AS CHAR)"),
 
316
            (m.MSNVarChar, "CAST(t.col AS CHAR)"),
 
317
 
 
318
            (LargeBinary, "CAST(t.col AS BINARY)"),
 
319
            (BLOB, "CAST(t.col AS BINARY)"),
 
320
            (m.MSBlob, "CAST(t.col AS BINARY)"),
 
321
            (m.MSBlob(32), "CAST(t.col AS BINARY)"),
 
322
            (m.MSTinyBlob, "CAST(t.col AS BINARY)"),
 
323
            (m.MSMediumBlob, "CAST(t.col AS BINARY)"),
 
324
            (m.MSLongBlob, "CAST(t.col AS BINARY)"),
 
325
            (m.MSBinary, "CAST(t.col AS BINARY)"),
 
326
            (m.MSBinary(32), "CAST(t.col AS BINARY)"),
 
327
            (m.MSVarBinary, "CAST(t.col AS BINARY)"),
 
328
            (m.MSVarBinary(32), "CAST(t.col AS BINARY)"),
 
329
 
 
330
            # maybe this could be changed to something more DWIM, needs
 
331
            # testing
 
332
            (Boolean, "t.col"),
 
333
            (BOOLEAN, "t.col"),
 
334
 
 
335
            (m.MSEnum, "t.col"),
 
336
            (m.MSEnum("1", "2"), "t.col"),
 
337
            (m.MSSet, "t.col"),
 
338
            (m.MSSet("1", "2"), "t.col"),
 
339
            ]
 
340
 
 
341
        for type_, expected in specs:
 
342
            self.assert_compile(cast(t.c.col, type_), expected)
 
343
 
 
344
    def test_no_cast_pre_4(self):
 
345
        self.assert_compile(
 
346
                    cast(Column('foo', Integer), String),
 
347
                    "CAST(foo AS CHAR)",
 
348
            )
 
349
        dialect = mysql.dialect()
 
350
        dialect.server_version_info = (3, 2, 3)
 
351
        self.assert_compile(
 
352
                    cast(Column('foo', Integer), String),
 
353
                    "foo",
 
354
                    dialect=dialect
 
355
            )
 
356
 
 
357
    def test_cast_grouped_expression_non_castable(self):
 
358
        self.assert_compile(
 
359
            cast(sql.column('x') + sql.column('y'), Float),
 
360
            "(x + y)"
 
361
        )
 
362
 
 
363
    def test_cast_grouped_expression_pre_4(self):
 
364
        dialect = mysql.dialect()
 
365
        dialect.server_version_info = (3, 2, 3)
 
366
        self.assert_compile(
 
367
            cast(sql.column('x') + sql.column('y'), Integer),
 
368
            "(x + y)",
 
369
            dialect=dialect
 
370
        )
 
371
 
 
372
    def test_extract(self):
 
373
        t = sql.table('t', sql.column('col1'))
 
374
 
 
375
        for field in 'year', 'month', 'day':
 
376
            self.assert_compile(
 
377
                select([extract(field, t.c.col1)]),
 
378
                "SELECT EXTRACT(%s FROM t.col1) AS anon_1 FROM t" % field)
 
379
 
 
380
        # millsecondS to millisecond
 
381
        self.assert_compile(
 
382
            select([extract('milliseconds', t.c.col1)]),
 
383
            "SELECT EXTRACT(millisecond FROM t.col1) AS anon_1 FROM t")
 
384
 
 
385
    def test_too_long_index(self):
 
386
        exp = 'ix_zyrenian_zyme_zyzzogeton_zyzzogeton_zyrenian_zyme_zyz_5cd2'
 
387
        tname = 'zyrenian_zyme_zyzzogeton_zyzzogeton'
 
388
        cname = 'zyrenian_zyme_zyzzogeton_zo'
 
389
 
 
390
        t1 = Table(tname, MetaData(),
 
391
                    Column(cname, Integer, index=True),
 
392
                )
 
393
        ix1 = list(t1.indexes)[0]
 
394
 
 
395
        self.assert_compile(
 
396
            schema.CreateIndex(ix1),
 
397
            "CREATE INDEX %s "
 
398
            "ON %s (%s)" % (exp, tname, cname)
 
399
        )
 
400
 
 
401
    def test_innodb_autoincrement(self):
 
402
        t1 = Table('sometable', MetaData(), Column('assigned_id',
 
403
                   Integer(), primary_key=True, autoincrement=False),
 
404
                   Column('id', Integer(), primary_key=True,
 
405
                   autoincrement=True), mysql_engine='InnoDB')
 
406
        self.assert_compile(schema.CreateTable(t1),
 
407
                            'CREATE TABLE sometable (assigned_id '
 
408
                            'INTEGER NOT NULL, id INTEGER NOT NULL '
 
409
                            'AUTO_INCREMENT, PRIMARY KEY (assigned_id, '
 
410
                            'id), KEY idx_autoinc_id (id))ENGINE=Inn'
 
411
                            'oDB')
 
412
 
 
413
        t1 = Table('sometable', MetaData(), Column('assigned_id',
 
414
                   Integer(), primary_key=True, autoincrement=True),
 
415
                   Column('id', Integer(), primary_key=True,
 
416
                   autoincrement=False), mysql_engine='InnoDB')
 
417
        self.assert_compile(schema.CreateTable(t1),
 
418
                            'CREATE TABLE sometable (assigned_id '
 
419
                            'INTEGER NOT NULL AUTO_INCREMENT, id '
 
420
                            'INTEGER NOT NULL, PRIMARY KEY '
 
421
                            '(assigned_id, id))ENGINE=InnoDB')
 
422
 
 
423
    def test_innodb_autoincrement_reserved_word_column_name(self):
 
424
        t1 = Table(
 
425
            'sometable', MetaData(),
 
426
            Column('id', Integer(), primary_key=True, autoincrement=False),
 
427
            Column('order', Integer(), primary_key=True, autoincrement=True),
 
428
            mysql_engine='InnoDB')
 
429
        self.assert_compile(
 
430
            schema.CreateTable(t1),
 
431
            'CREATE TABLE sometable ('
 
432
            'id INTEGER NOT NULL, '
 
433
            '`order` INTEGER NOT NULL AUTO_INCREMENT, '
 
434
            'PRIMARY KEY (id, `order`), '
 
435
            'KEY idx_autoinc_order (`order`)'
 
436
            ')ENGINE=InnoDB')
 
437