3
from test.lib.testing import eq_, assert_raises, assert_raises_message
6
from sqlalchemy import *
7
from sqlalchemy import sql, exc, schema, types as sqltypes
8
from sqlalchemy.dialects.mysql import base as mysql
9
from sqlalchemy.engine.url import make_url
10
from test.lib import fixtures, AssertsCompiledSQL, AssertsExecutionResults
11
from test.lib import testing, engines
12
from test.lib.engines import utf8_engine
15
class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
17
__dialect__ = mysql.dialect()
19
def test_reserved_words(self):
20
table = Table("mysql_table", MetaData(),
21
Column("col1", Integer),
22
Column("master_ssl_verify_server_cert", Integer))
23
x = select([table.c.col1, table.c.master_ssl_verify_server_cert])
25
self.assert_compile(x,
26
'''SELECT mysql_table.col1, mysql_table.`master_ssl_verify_server_cert` FROM mysql_table''')
28
def test_create_index_simple(self):
30
tbl = Table('testtbl', m, Column('data', String(255)))
31
idx = Index('test_idx1', tbl.c.data)
33
self.assert_compile(schema.CreateIndex(idx),
34
'CREATE INDEX test_idx1 ON testtbl (data)',
35
dialect=mysql.dialect())
37
def test_create_index_with_length(self):
39
tbl = Table('testtbl', m, Column('data', String(255)))
40
idx1 = Index('test_idx1', tbl.c.data, mysql_length=10)
41
idx2 = Index('test_idx2', tbl.c.data, mysql_length=5)
43
self.assert_compile(schema.CreateIndex(idx1),
44
'CREATE INDEX test_idx1 ON testtbl (data(10))',
45
dialect=mysql.dialect())
46
self.assert_compile(schema.CreateIndex(idx2),
47
'CREATE INDEX test_idx2 ON testtbl (data(5))',
48
dialect=mysql.dialect())
50
def test_create_index_with_using(self):
52
tbl = Table('testtbl', m, Column('data', String(255)))
53
idx1 = Index('test_idx1', tbl.c.data, mysql_using='btree')
54
idx2 = Index('test_idx2', tbl.c.data, mysql_using='hash')
56
self.assert_compile(schema.CreateIndex(idx1),
57
'CREATE INDEX test_idx1 ON testtbl (data) USING btree',
58
dialect=mysql.dialect())
59
self.assert_compile(schema.CreateIndex(idx2),
60
'CREATE INDEX test_idx2 ON testtbl (data) USING hash',
61
dialect=mysql.dialect())
63
def test_create_pk_plain(self):
65
tbl = Table('testtbl', m, Column('data', String(255)),
66
PrimaryKeyConstraint('data'))
68
self.assert_compile(schema.CreateTable(tbl),
69
"CREATE TABLE testtbl (data VARCHAR(255), PRIMARY KEY (data))",
70
dialect=mysql.dialect())
72
def test_create_pk_with_using(self):
74
tbl = Table('testtbl', m, Column('data', String(255)),
75
PrimaryKeyConstraint('data', mysql_using='btree'))
77
self.assert_compile(schema.CreateTable(tbl),
78
"CREATE TABLE testtbl (data VARCHAR(255), "
79
"PRIMARY KEY (data) USING btree)",
80
dialect=mysql.dialect())
82
class DialectTest(fixtures.TestBase):
85
@testing.only_on(['mysql+mysqldb', 'mysql+oursql'],
86
'requires particular SSL arguments')
87
def test_ssl_arguments(self):
88
dialect = testing.db.dialect
89
kwarg = dialect.create_connect_args(
90
make_url("mysql://scott:tiger@localhost:3306/test"
91
"?ssl_ca=/ca.pem&ssl_cert=/cert.pem&ssl_key=/key.pem")
93
# args that differ among mysqldb and oursql
94
for k in ('use_unicode', 'found_rows', 'client_flag'):
99
'passwd': 'tiger', 'db': 'test',
100
'ssl': {'ca': '/ca.pem', 'cert': '/cert.pem',
102
'host': 'localhost', 'user': 'scott',
108
class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
109
"Test MySQL column types"
111
__only_on__ = 'mysql'
112
__dialect__ = mysql.dialect()
114
@testing.uses_deprecated('Manually quoting ENUM value literals')
115
def test_basic(self):
116
meta1 = MetaData(testing.db)
118
'mysql_types', meta1,
119
Column('id', Integer, primary_key=True),
120
Column('num1', mysql.MSInteger(unsigned=True)),
121
Column('text1', mysql.MSLongText),
122
Column('text2', mysql.MSLongText()),
123
Column('num2', mysql.MSBigInteger),
124
Column('num3', mysql.MSBigInteger()),
125
Column('num4', mysql.MSDouble),
126
Column('num5', mysql.MSDouble()),
127
Column('num6', mysql.MSMediumInteger),
128
Column('enum1', mysql.ENUM("'black'", "'white'")),
129
Column('enum2', mysql.ENUM("dog", "cat")),
132
table.drop(checkfirst=True)
134
meta2 = MetaData(testing.db)
135
t2 = Table('mysql_types', meta2, autoload=True)
136
assert isinstance(t2.c.num1.type, mysql.MSInteger)
137
assert t2.c.num1.type.unsigned
138
assert isinstance(t2.c.text1.type, mysql.MSLongText)
139
assert isinstance(t2.c.text2.type, mysql.MSLongText)
140
assert isinstance(t2.c.num2.type, mysql.MSBigInteger)
141
assert isinstance(t2.c.num3.type, mysql.MSBigInteger)
142
assert isinstance(t2.c.num4.type, mysql.MSDouble)
143
assert isinstance(t2.c.num5.type, mysql.MSDouble)
144
assert isinstance(t2.c.num6.type, mysql.MSMediumInteger)
145
assert isinstance(t2.c.enum1.type, mysql.ENUM)
146
assert isinstance(t2.c.enum2.type, mysql.ENUM)
152
def test_numeric(self):
153
"Exercise type specification and options for numeric types."
156
# column type, args, kwargs, expected ddl
157
# e.g. Column(Integer(10, unsigned=True)) ==
158
# 'INTEGER(10) UNSIGNED'
159
(mysql.MSNumeric, [], {},
161
(mysql.MSNumeric, [None], {},
163
(mysql.MSNumeric, [12], {},
165
(mysql.MSNumeric, [12, 4], {'unsigned':True},
166
'NUMERIC(12, 4) UNSIGNED'),
167
(mysql.MSNumeric, [12, 4], {'zerofill':True},
168
'NUMERIC(12, 4) ZEROFILL'),
169
(mysql.MSNumeric, [12, 4], {'zerofill':True, 'unsigned':True},
170
'NUMERIC(12, 4) UNSIGNED ZEROFILL'),
172
(mysql.MSDecimal, [], {},
174
(mysql.MSDecimal, [None], {},
176
(mysql.MSDecimal, [12], {},
178
(mysql.MSDecimal, [12, None], {},
180
(mysql.MSDecimal, [12, 4], {'unsigned':True},
181
'DECIMAL(12, 4) UNSIGNED'),
182
(mysql.MSDecimal, [12, 4], {'zerofill':True},
183
'DECIMAL(12, 4) ZEROFILL'),
184
(mysql.MSDecimal, [12, 4], {'zerofill':True, 'unsigned':True},
185
'DECIMAL(12, 4) UNSIGNED ZEROFILL'),
187
(mysql.MSDouble, [None, None], {},
189
(mysql.MSDouble, [12, 4], {'unsigned':True},
190
'DOUBLE(12, 4) UNSIGNED'),
191
(mysql.MSDouble, [12, 4], {'zerofill':True},
192
'DOUBLE(12, 4) ZEROFILL'),
193
(mysql.MSDouble, [12, 4], {'zerofill':True, 'unsigned':True},
194
'DOUBLE(12, 4) UNSIGNED ZEROFILL'),
196
(mysql.MSReal, [None, None], {},
198
(mysql.MSReal, [12, 4], {'unsigned':True},
199
'REAL(12, 4) UNSIGNED'),
200
(mysql.MSReal, [12, 4], {'zerofill':True},
201
'REAL(12, 4) ZEROFILL'),
202
(mysql.MSReal, [12, 4], {'zerofill':True, 'unsigned':True},
203
'REAL(12, 4) UNSIGNED ZEROFILL'),
205
(mysql.MSFloat, [], {},
207
(mysql.MSFloat, [None], {},
209
(mysql.MSFloat, [12], {},
211
(mysql.MSFloat, [12, 4], {},
213
(mysql.MSFloat, [12, 4], {'unsigned':True},
214
'FLOAT(12, 4) UNSIGNED'),
215
(mysql.MSFloat, [12, 4], {'zerofill':True},
216
'FLOAT(12, 4) ZEROFILL'),
217
(mysql.MSFloat, [12, 4], {'zerofill':True, 'unsigned':True},
218
'FLOAT(12, 4) UNSIGNED ZEROFILL'),
220
(mysql.MSInteger, [], {},
222
(mysql.MSInteger, [4], {},
224
(mysql.MSInteger, [4], {'unsigned':True},
225
'INTEGER(4) UNSIGNED'),
226
(mysql.MSInteger, [4], {'zerofill':True},
227
'INTEGER(4) ZEROFILL'),
228
(mysql.MSInteger, [4], {'zerofill':True, 'unsigned':True},
229
'INTEGER(4) UNSIGNED ZEROFILL'),
231
(mysql.MSBigInteger, [], {},
233
(mysql.MSBigInteger, [4], {},
235
(mysql.MSBigInteger, [4], {'unsigned':True},
236
'BIGINT(4) UNSIGNED'),
237
(mysql.MSBigInteger, [4], {'zerofill':True},
238
'BIGINT(4) ZEROFILL'),
239
(mysql.MSBigInteger, [4], {'zerofill':True, 'unsigned':True},
240
'BIGINT(4) UNSIGNED ZEROFILL'),
242
(mysql.MSMediumInteger, [], {},
244
(mysql.MSMediumInteger, [4], {},
246
(mysql.MSMediumInteger, [4], {'unsigned':True},
247
'MEDIUMINT(4) UNSIGNED'),
248
(mysql.MSMediumInteger, [4], {'zerofill':True},
249
'MEDIUMINT(4) ZEROFILL'),
250
(mysql.MSMediumInteger, [4], {'zerofill':True, 'unsigned':True},
251
'MEDIUMINT(4) UNSIGNED ZEROFILL'),
253
(mysql.MSTinyInteger, [], {},
255
(mysql.MSTinyInteger, [1], {},
257
(mysql.MSTinyInteger, [1], {'unsigned':True},
258
'TINYINT(1) UNSIGNED'),
259
(mysql.MSTinyInteger, [1], {'zerofill':True},
260
'TINYINT(1) ZEROFILL'),
261
(mysql.MSTinyInteger, [1], {'zerofill':True, 'unsigned':True},
262
'TINYINT(1) UNSIGNED ZEROFILL'),
264
(mysql.MSSmallInteger, [], {},
266
(mysql.MSSmallInteger, [4], {},
268
(mysql.MSSmallInteger, [4], {'unsigned':True},
269
'SMALLINT(4) UNSIGNED'),
270
(mysql.MSSmallInteger, [4], {'zerofill':True},
271
'SMALLINT(4) ZEROFILL'),
272
(mysql.MSSmallInteger, [4], {'zerofill':True, 'unsigned':True},
273
'SMALLINT(4) UNSIGNED ZEROFILL'),
276
table_args = ['test_mysql_numeric', MetaData(testing.db)]
277
for index, spec in enumerate(columns):
278
type_, args, kw, res = spec
279
table_args.append(Column('c%s' % index, type_(*args, **kw)))
281
numeric_table = Table(*table_args)
282
gen = testing.db.dialect.ddl_compiler(testing.db.dialect, None)
284
for col in numeric_table.c:
285
index = int(col.name[1:])
286
eq_(gen.get_column_specification(col),
287
"%s %s" % (col.name, columns[index][3]))
288
self.assert_(repr(col))
291
numeric_table.create(checkfirst=True)
297
@testing.exclude('mysql', '<', (4, 1, 1), 'no charset support')
298
def test_charset(self):
299
"""Exercise CHARACTER SET and COLLATE-ish options on string types."""
302
(mysql.MSChar, [1], {},
304
(mysql.NCHAR, [1], {},
306
(mysql.MSChar, [1], {'binary':True},
308
(mysql.MSChar, [1], {'ascii':True},
310
(mysql.MSChar, [1], {'unicode':True},
312
(mysql.MSChar, [1], {'ascii':True, 'binary':True},
313
'CHAR(1) ASCII BINARY'),
314
(mysql.MSChar, [1], {'unicode':True, 'binary':True},
315
'CHAR(1) UNICODE BINARY'),
316
(mysql.MSChar, [1], {'charset':'utf8'},
317
'CHAR(1) CHARACTER SET utf8'),
318
(mysql.MSChar, [1], {'charset':'utf8', 'binary':True},
319
'CHAR(1) CHARACTER SET utf8 BINARY'),
320
(mysql.MSChar, [1], {'charset':'utf8', 'unicode':True},
321
'CHAR(1) CHARACTER SET utf8'),
322
(mysql.MSChar, [1], {'charset':'utf8', 'ascii':True},
323
'CHAR(1) CHARACTER SET utf8'),
324
(mysql.MSChar, [1], {'collation': 'utf8_bin'},
325
'CHAR(1) COLLATE utf8_bin'),
326
(mysql.MSChar, [1], {'charset': 'utf8', 'collation': 'utf8_bin'},
327
'CHAR(1) CHARACTER SET utf8 COLLATE utf8_bin'),
328
(mysql.MSChar, [1], {'charset': 'utf8', 'binary': True},
329
'CHAR(1) CHARACTER SET utf8 BINARY'),
330
(mysql.MSChar, [1], {'charset': 'utf8', 'collation': 'utf8_bin',
332
'CHAR(1) CHARACTER SET utf8 COLLATE utf8_bin'),
333
(mysql.MSChar, [1], {'national':True},
335
(mysql.MSChar, [1], {'national':True, 'charset':'utf8'},
337
(mysql.MSChar, [1], {'national':True, 'charset':'utf8',
339
'NATIONAL CHAR(1) BINARY'),
340
(mysql.MSChar, [1], {'national':True, 'binary':True,
342
'NATIONAL CHAR(1) BINARY'),
343
(mysql.MSChar, [1], {'national':True, 'collation':'utf8_bin'},
344
'NATIONAL CHAR(1) COLLATE utf8_bin'),
346
(mysql.MSString, [1], {'charset':'utf8', 'collation':'utf8_bin'},
347
'VARCHAR(1) CHARACTER SET utf8 COLLATE utf8_bin'),
348
(mysql.MSString, [1], {'national':True, 'collation':'utf8_bin'},
349
'NATIONAL VARCHAR(1) COLLATE utf8_bin'),
351
(mysql.MSTinyText, [], {'charset':'utf8', 'collation':'utf8_bin'},
352
'TINYTEXT CHARACTER SET utf8 COLLATE utf8_bin'),
354
(mysql.MSMediumText, [], {'charset':'utf8', 'binary':True},
355
'MEDIUMTEXT CHARACTER SET utf8 BINARY'),
357
(mysql.MSLongText, [], {'ascii':True},
360
(mysql.ENUM, ["foo", "bar"], {'unicode':True},
361
'''ENUM('foo','bar') UNICODE''')
364
table_args = ['test_mysql_charset', MetaData(testing.db)]
365
for index, spec in enumerate(columns):
366
type_, args, kw, res = spec
367
table_args.append(Column('c%s' % index, type_(*args, **kw)))
369
charset_table = Table(*table_args)
370
gen = testing.db.dialect.ddl_compiler(testing.db.dialect, None)
372
for col in charset_table.c:
373
index = int(col.name[1:])
374
eq_(gen.get_column_specification(col),
375
"%s %s" % (col.name, columns[index][3]))
376
self.assert_(repr(col))
379
charset_table.create(checkfirst=True)
385
@testing.exclude('mysql', '<', (5, 0, 5), 'a 5.0+ feature')
386
@testing.provide_metadata
387
def test_charset_collate_table(self):
388
t = Table('foo', self.metadata,
389
Column('id', Integer),
390
mysql_default_charset='utf8',
391
mysql_collate='utf8_unicode_ci'
394
m2 = MetaData(testing.db)
395
t2 = Table('foo', m2, autoload=True)
396
eq_(t2.kwargs['mysql_collate'], 'utf8_unicode_ci')
397
eq_(t2.kwargs['mysql_default charset'], 'utf8')
399
@testing.exclude('mysql', '<', (5, 0, 5), 'a 5.0+ feature')
400
@testing.fails_on('mysql+oursql', 'some round trips fail, oursql bug ?')
401
def test_bit_50(self):
402
"""Exercise BIT types on 5.0+ (not valid for all engine types)"""
404
meta = MetaData(testing.db)
405
bit_table = Table('mysql_bits', meta,
406
Column('b1', mysql.MSBit),
407
Column('b2', mysql.MSBit()),
408
Column('b3', mysql.MSBit(), nullable=False),
409
Column('b4', mysql.MSBit(1)),
410
Column('b5', mysql.MSBit(8)),
411
Column('b6', mysql.MSBit(32)),
412
Column('b7', mysql.MSBit(63)),
413
Column('b8', mysql.MSBit(64)))
415
eq_(colspec(bit_table.c.b1), 'b1 BIT')
416
eq_(colspec(bit_table.c.b2), 'b2 BIT')
417
eq_(colspec(bit_table.c.b3), 'b3 BIT NOT NULL')
418
eq_(colspec(bit_table.c.b4), 'b4 BIT(1)')
419
eq_(colspec(bit_table.c.b5), 'b5 BIT(8)')
420
eq_(colspec(bit_table.c.b6), 'b6 BIT(32)')
421
eq_(colspec(bit_table.c.b7), 'b7 BIT(63)')
422
eq_(colspec(bit_table.c.b8), 'b8 BIT(64)')
424
for col in bit_table.c:
425
self.assert_(repr(col))
429
meta2 = MetaData(testing.db)
430
reflected = Table('mysql_bits', meta2, autoload=True)
432
for table in bit_table, reflected:
434
def roundtrip(store, expected=None):
435
expected = expected or store
436
table.insert(store).execute()
437
row = table.select().execute().first()
439
self.assert_(list(row) == expected)
441
print "Storing %s" % store
442
print "Expected %s" % expected
443
print "Found %s" % list(row)
445
table.delete().execute().close()
448
roundtrip([None, None, 0, None, None, None, None, None])
450
roundtrip([sql.text("b'1'")] * 8, [1] * 8)
453
roundtrip([0, 0, 0, 0, i, i, i, i])
455
roundtrip([0, 0, 0, 0, 0, i, i, i])
457
roundtrip([0, 0, 0, 0, 0, 0, i, i])
459
roundtrip([0, 0, 0, 0, 0, 0, 0, i])
463
def test_boolean(self):
464
"""Test BOOL/TINYINT(1) compatibility and reflection."""
466
meta = MetaData(testing.db)
470
Column('b1', BOOLEAN),
471
Column('b2', Boolean),
472
Column('b3', mysql.MSTinyInteger(1)),
473
Column('b4', mysql.MSTinyInteger(1, unsigned=True)),
474
Column('b5', mysql.MSTinyInteger),
476
eq_(colspec(bool_table.c.b1), 'b1 BOOL')
477
eq_(colspec(bool_table.c.b2), 'b2 BOOL')
478
eq_(colspec(bool_table.c.b3), 'b3 TINYINT(1)')
479
eq_(colspec(bool_table.c.b4), 'b4 TINYINT(1) UNSIGNED')
480
eq_(colspec(bool_table.c.b5), 'b5 TINYINT')
481
for col in bool_table.c:
482
self.assert_(repr(col))
487
def roundtrip(store, expected=None):
488
expected = expected or store
489
table.insert(store).execute()
490
row = table.select().execute().first()
492
self.assert_(list(row) == expected)
493
for i, val in enumerate(expected):
494
if isinstance(val, bool):
495
self.assert_(val is row[i])
497
print 'Storing %s' % store
498
print 'Expected %s' % expected
499
print 'Found %s' % list(row)
501
table.delete().execute().close()
503
roundtrip([None, None, None, None, None])
504
roundtrip([True, True, 1, 1, 1])
505
roundtrip([False, False, 0, 0, 0])
506
roundtrip([True, True, True, True, True], [True, True, 1,
508
roundtrip([False, False, 0, 0, 0], [False, False, 0, 0, 0])
509
meta2 = MetaData(testing.db)
510
table = Table('mysql_bool', meta2, autoload=True)
511
eq_(colspec(table.c.b3), 'b3 TINYINT(1)')
512
eq_(colspec(table.c.b4), 'b4 TINYINT(1) UNSIGNED')
513
meta2 = MetaData(testing.db)
517
Column('b1', BOOLEAN),
518
Column('b2', Boolean),
519
Column('b3', BOOLEAN),
520
Column('b4', BOOLEAN),
523
eq_(colspec(table.c.b3), 'b3 BOOL')
524
eq_(colspec(table.c.b4), 'b4 BOOL')
525
roundtrip([None, None, None, None, None])
526
roundtrip([True, True, 1, 1, 1], [True, True, True, True,
528
roundtrip([False, False, 0, 0, 0], [False, False, False,
530
roundtrip([True, True, True, True, True], [True, True,
532
roundtrip([False, False, 0, 0, 0], [False, False, False,
537
@testing.exclude('mysql', '<', (4, 1, 0), '4.1+ syntax')
538
def test_timestamp(self):
539
"""Exercise funky TIMESTAMP default syntax."""
541
meta = MetaData(testing.db)
547
([mysql.MSTimeStamp],
550
DefaultClause(sql.text('CURRENT_TIMESTAMP'))],
551
"TIMESTAMP DEFAULT CURRENT_TIMESTAMP"),
553
DefaultClause(sql.text("'1999-09-09 09:09:09'"))],
554
"TIMESTAMP DEFAULT '1999-09-09 09:09:09'"),
556
DefaultClause(sql.text("'1999-09-09 09:09:09' "
557
"ON UPDATE CURRENT_TIMESTAMP"))],
558
"TIMESTAMP DEFAULT '1999-09-09 09:09:09' "
559
"ON UPDATE CURRENT_TIMESTAMP"),
561
DefaultClause(sql.text("CURRENT_TIMESTAMP "
562
"ON UPDATE CURRENT_TIMESTAMP"))],
563
"TIMESTAMP DEFAULT CURRENT_TIMESTAMP "
564
"ON UPDATE CURRENT_TIMESTAMP"),
566
for idx, (spec, expected) in enumerate(columns):
567
t = Table('mysql_ts%s' % idx, meta,
568
Column('id', Integer, primary_key=True),
570
eq_(colspec(t.c.t), "t %s" % expected)
571
self.assert_(repr(t.c.t))
573
r = Table('mysql_ts%s' % idx, MetaData(testing.db),
576
self.assert_(r.c.t is not None)
580
def test_timestamp_nullable(self):
581
meta = MetaData(testing.db)
582
ts_table = Table('mysql_timestamp', meta,
583
Column('t1', TIMESTAMP),
584
Column('t2', TIMESTAMP, nullable=False),
588
# there's a slight assumption here that this test can
589
# complete within the scope of a single second.
590
# if needed, can break out the eq_() just to check for
591
# timestamps that are within a few seconds of "now"
594
now = testing.db.execute("select now()").scalar()
596
# TIMESTAMP without NULL inserts current time when passed
597
# NULL. when not passed, generates 0000-00-00 quite
599
ts_table.insert().execute({'t1':now, 't2':None})
600
ts_table.insert().execute({'t1':None, 't2':None})
603
ts_table.select().execute().fetchall(),
604
[(now, now), (None, now)]
612
meta = MetaData(testing.db)
613
year_table = Table('mysql_year', meta,
614
Column('y1', mysql.MSYear),
615
Column('y2', mysql.MSYear),
616
Column('y3', mysql.MSYear),
617
Column('y4', mysql.MSYear(2)),
618
Column('y5', mysql.MSYear(4)))
620
for col in year_table.c:
621
self.assert_(repr(col))
624
reflected = Table('mysql_year', MetaData(testing.db),
627
for table in year_table, reflected:
628
table.insert(['1950', '50', None, 50, 1950]).execute()
629
row = table.select().execute().first()
630
eq_(list(row), [1950, 2050, None, 50, 1950])
631
table.delete().execute()
632
self.assert_(colspec(table.c.y1).startswith('y1 YEAR'))
633
eq_(colspec(table.c.y4), 'y4 YEAR(2)')
634
eq_(colspec(table.c.y5), 'y5 YEAR(4)')
640
"""Exercise the SET type."""
642
meta = MetaData(testing.db)
643
set_table = Table('mysql_set', meta, Column('s1',
644
mysql.MSSet("'dq'", "'sq'")), Column('s2',
645
mysql.MSSet("'a'")), Column('s3',
646
mysql.MSSet("'5'", "'7'", "'9'")))
647
eq_(colspec(set_table.c.s1), "s1 SET('dq','sq')")
648
eq_(colspec(set_table.c.s2), "s2 SET('a')")
649
eq_(colspec(set_table.c.s3), "s3 SET('5','7','9')")
650
for col in set_table.c:
651
self.assert_(repr(col))
654
reflected = Table('mysql_set', MetaData(testing.db),
656
for table in set_table, reflected:
658
def roundtrip(store, expected=None):
659
expected = expected or store
660
table.insert(store).execute()
661
row = table.select().execute().first()
663
self.assert_(list(row) == expected)
665
print 'Storing %s' % store
666
print 'Expected %s' % expected
667
print 'Found %s' % list(row)
669
table.delete().execute()
671
roundtrip([None, None, None], [None] * 3)
672
roundtrip(['', '', ''], [set([''])] * 3)
673
roundtrip([set(['dq']), set(['a']), set(['5'])])
674
roundtrip(['dq', 'a', '5'], [set(['dq']), set(['a']),
676
roundtrip([1, 1, 1], [set(['dq']), set(['a']), set(['5'
678
roundtrip([set(['dq', 'sq']), None, set(['9', '5', '7'
680
set_table.insert().execute({'s3': set(['5'])}, {'s3'
681
: set(['5', '7'])}, {'s3': set(['5', '7', '9'])},
682
{'s3': set(['7', '9'])})
683
rows = select([set_table.c.s3], set_table.c.s3.in_([set(['5'
684
]), set(['5', '7']), set(['7', '5'
685
])])).execute().fetchall()
686
found = set([frozenset(row[0]) for row in rows])
687
eq_(found, set([frozenset(['5']), frozenset(['5', '7'])]))
691
@testing.uses_deprecated('Manually quoting ENUM value literals')
693
"""Exercise the ENUM type."""
696
enum_table = Table('mysql_enum', MetaData(testing.db),
697
Column('e1', mysql.ENUM("'a'", "'b'")),
698
Column('e2', mysql.ENUM("'a'", "'b'"),
700
Column('e2generic', Enum("a", "b"),
702
Column('e3', mysql.ENUM("'a'", "'b'", strict=True)),
703
Column('e4', mysql.ENUM("'a'", "'b'", strict=True),
705
Column('e5', mysql.ENUM("a", "b")),
706
Column('e5generic', Enum("a", "b")),
707
Column('e6', mysql.ENUM("'a'", "b")),
710
eq_(colspec(enum_table.c.e1),
712
eq_(colspec(enum_table.c.e2),
713
"e2 ENUM('a','b') NOT NULL")
714
eq_(colspec(enum_table.c.e2generic),
715
"e2generic ENUM('a','b') NOT NULL")
716
eq_(colspec(enum_table.c.e3),
718
eq_(colspec(enum_table.c.e4),
719
"e4 ENUM('a','b') NOT NULL")
720
eq_(colspec(enum_table.c.e5),
722
eq_(colspec(enum_table.c.e5generic),
723
"e5generic ENUM('a','b')")
724
eq_(colspec(enum_table.c.e6),
725
"e6 ENUM('''a''','b')")
726
enum_table.drop(checkfirst=True)
729
assert_raises(exc.DBAPIError, enum_table.insert().execute,
730
e1=None, e2=None, e3=None, e4=None)
732
assert_raises(exc.StatementError, enum_table.insert().execute,
733
e1='c', e2='c', e2generic='c', e3='c',
734
e4='c', e5='c', e5generic='c', e6='c')
736
enum_table.insert().execute()
737
enum_table.insert().execute(e1='a', e2='a', e2generic='a', e3='a',
738
e4='a', e5='a', e5generic='a', e6="'a'")
739
enum_table.insert().execute(e1='b', e2='b', e2generic='b', e3='b',
740
e4='b', e5='b', e5generic='b', e6='b')
742
res = enum_table.select().execute().fetchall()
744
expected = [(None, 'a', 'a', None, 'a', None, None, None),
745
('a', 'a', 'a', 'a', 'a', 'a', 'a', "'a'"),
746
('b', 'b', 'b', 'b', 'b', 'b', 'b', 'b')]
748
# This is known to fail with MySQLDB 1.2.2 beta versions
749
# which return these as sets.Set(['a']), sets.Set(['b'])
750
# (even on Pythons with __builtin__.set)
751
if (testing.against('mysql+mysqldb') and
752
testing.db.dialect.dbapi.version_info < (1, 2, 2, 'beta', 3) and
753
testing.db.dialect.dbapi.version_info >= (1, 2, 2)):
754
# these mysqldb seem to always uses 'sets', even on later pythons
762
return sets.Set([value])
766
e.append(tuple([convert(c) for c in row]))
772
def test_unicode_enum(self):
773
unicode_engine = utf8_engine()
774
metadata = MetaData(unicode_engine)
775
t1 = Table('table', metadata,
776
Column('id', Integer, primary_key=True),
777
Column('value', Enum(u'réveillé', u'drôle', u'S’il')),
778
Column('value2', mysql.ENUM(u'réveillé', u'drôle', u'S’il'))
780
metadata.create_all()
782
t1.insert().execute(value=u'drôle', value2=u'drôle')
783
t1.insert().execute(value=u'réveillé', value2=u'réveillé')
784
t1.insert().execute(value=u'S’il', value2=u'S’il')
785
eq_(t1.select().order_by(t1.c.id).execute().fetchall(),
786
[(1, u'drôle', u'drôle'), (2, u'réveillé', u'réveillé'),
787
(3, u'S’il', u'S’il')]
790
# test reflection of the enum labels
792
m2 = MetaData(testing.db)
793
t2 = Table('table', m2, autoload=True)
795
# TODO: what's wrong with the last element ? is there
796
# latin-1 stuff forcing its way in ?
798
assert t2.c.value.type.enums[0:2] == \
799
(u'réveillé', u'drôle') #, u'S’il') # eh ?
800
assert t2.c.value2.type.enums[0:2] == \
801
(u'réveillé', u'drôle') #, u'S’il') # eh ?
805
def test_enum_compile(self):
806
e1 = Enum('x', 'y', 'z', name='somename')
807
t1 = Table('sometable', MetaData(), Column('somecolumn', e1))
808
self.assert_compile(schema.CreateTable(t1),
809
"CREATE TABLE sometable (somecolumn "
810
"ENUM('x','y','z'))")
811
t1 = Table('sometable', MetaData(), Column('somecolumn',
812
Enum('x', 'y', 'z', native_enum=False)))
813
self.assert_compile(schema.CreateTable(t1),
814
"CREATE TABLE sometable (somecolumn "
815
"VARCHAR(1), CHECK (somecolumn IN ('x', "
818
@testing.exclude('mysql', '<', (4,), "3.23 can't handle an ENUM of ''")
819
@testing.uses_deprecated('Manually quoting ENUM value literals')
820
def test_enum_parse(self):
821
"""More exercises for the ENUM type."""
823
# MySQL 3.23 can't handle an ENUM of ''....
825
enum_table = Table('mysql_enum', MetaData(testing.db),
826
Column('e1', mysql.ENUM("'a'")),
827
Column('e2', mysql.ENUM("''")),
828
Column('e3', mysql.ENUM('a')),
829
Column('e4', mysql.ENUM('')),
830
Column('e5', mysql.ENUM("'a'", "''")),
831
Column('e6', mysql.ENUM("''", "'a'")),
832
Column('e7', mysql.ENUM("''", "'''a'''", "'b''b'", "''''")))
834
for col in enum_table.c:
835
self.assert_(repr(col))
838
reflected = Table('mysql_enum', MetaData(testing.db),
840
for t in enum_table, reflected:
841
eq_(t.c.e1.type.enums, ("a",))
842
eq_(t.c.e2.type.enums, ("",))
843
eq_(t.c.e3.type.enums, ("a",))
844
eq_(t.c.e4.type.enums, ("",))
845
eq_(t.c.e5.type.enums, ("a", ""))
846
eq_(t.c.e6.type.enums, ("", "a"))
847
eq_(t.c.e7.type.enums, ("", "'a'", "b'b", "'"))
851
class ReflectionTest(fixtures.TestBase, AssertsExecutionResults):
853
__only_on__ = 'mysql'
855
def test_default_reflection(self):
856
"""Test reflection of column defaults."""
858
from sqlalchemy.dialects.mysql import VARCHAR
861
MetaData(testing.db),
862
Column('c1', VARCHAR(10, collation='utf8_unicode_ci'),
863
DefaultClause(''), nullable=False),
864
Column('c2', String(10), DefaultClause('0')),
865
Column('c3', String(10), DefaultClause('abc')),
866
Column('c4', TIMESTAMP, DefaultClause('2009-04-05 12:00:00'
868
Column('c5', TIMESTAMP),
869
Column('c6', TIMESTAMP,
870
DefaultClause(sql.text("CURRENT_TIMESTAMP "
871
"ON UPDATE CURRENT_TIMESTAMP"))),
875
reflected = Table('mysql_def', MetaData(testing.db),
879
assert def_table.c.c1.server_default.arg == ''
880
assert def_table.c.c2.server_default.arg == '0'
881
assert def_table.c.c3.server_default.arg == 'abc'
882
assert def_table.c.c4.server_default.arg \
883
== '2009-04-05 12:00:00'
884
assert str(reflected.c.c1.server_default.arg) == "''"
885
assert str(reflected.c.c2.server_default.arg) == "'0'"
886
assert str(reflected.c.c3.server_default.arg) == "'abc'"
887
assert str(reflected.c.c4.server_default.arg) \
888
== "'2009-04-05 12:00:00'"
889
assert reflected.c.c5.default is None
890
assert reflected.c.c5.server_default is None
891
assert reflected.c.c6.default is None
893
str(reflected.c.c6.server_default.arg).upper(),
894
"CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"
898
reflected2 = Table('mysql_def', MetaData(testing.db),
902
assert str(reflected2.c.c1.server_default.arg) == "''"
903
assert str(reflected2.c.c2.server_default.arg) == "'0'"
904
assert str(reflected2.c.c3.server_default.arg) == "'abc'"
905
assert str(reflected2.c.c4.server_default.arg) \
906
== "'2009-04-05 12:00:00'"
907
assert reflected.c.c5.default is None
908
assert reflected.c.c5.server_default is None
909
assert reflected.c.c6.default is None
911
str(reflected.c.c6.server_default.arg).upper(),
912
"CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"
915
def test_reflection_with_table_options(self):
916
comment = r"""Comment types type speedily ' " \ '' Fun!"""
918
def_table = Table('mysql_def', MetaData(testing.db),
919
Column('c1', Integer()),
920
mysql_engine='MEMORY',
921
mysql_comment=comment,
922
mysql_default_charset='utf8',
923
mysql_auto_increment='5',
924
mysql_avg_row_length='3',
925
mysql_password='secret',
926
mysql_connection='fish',
931
reflected = Table('mysql_def', MetaData(testing.db),
936
assert def_table.kwargs['mysql_engine'] == 'MEMORY'
937
assert def_table.kwargs['mysql_comment'] == comment
938
assert def_table.kwargs['mysql_default_charset'] == 'utf8'
939
assert def_table.kwargs['mysql_auto_increment'] == '5'
940
assert def_table.kwargs['mysql_avg_row_length'] == '3'
941
assert def_table.kwargs['mysql_password'] == 'secret'
942
assert def_table.kwargs['mysql_connection'] == 'fish'
944
assert reflected.kwargs['mysql_engine'] == 'MEMORY'
945
assert reflected.kwargs['mysql_comment'] == comment
946
assert reflected.kwargs['mysql_default charset'] == 'utf8'
947
assert reflected.kwargs['mysql_avg_row_length'] == '3'
948
assert reflected.kwargs['mysql_connection'] == 'fish'
950
# This field doesn't seem to be returned by mysql itself.
951
#assert reflected.kwargs['mysql_password'] == 'secret'
953
# This is explicitly ignored when reflecting schema.
954
#assert reflected.kwargs['mysql_auto_increment'] == '5'
956
def test_reflection_on_include_columns(self):
957
"""Test reflection of include_columns to be sure they respect case."""
959
case_table = Table('mysql_case', MetaData(testing.db),
960
Column('c1', String(10)),
961
Column('C2', String(10)),
962
Column('C3', String(10)))
966
reflected = Table('mysql_case', MetaData(testing.db),
967
autoload=True, include_columns=['c1', 'C2'])
968
for t in case_table, reflected:
969
assert 'c1' in t.c.keys()
970
assert 'C2' in t.c.keys()
971
reflected2 = Table('mysql_case', MetaData(testing.db),
972
autoload=True, include_columns=['c1', 'c2'])
973
assert 'c1' in reflected2.c.keys()
974
for c in ['c2', 'C2', 'C3']:
975
assert c not in reflected2.c.keys()
979
@testing.exclude('mysql', '<', (5, 0, 0), 'early types are squirrely')
980
@testing.uses_deprecated('Using String type with no length')
981
@testing.uses_deprecated('Manually quoting ENUM value literals')
982
def test_type_reflection(self):
983
# (ask_for, roundtripped_as_if_different)
984
specs = [( String(1), mysql.MSString(1), ),
985
( String(3), mysql.MSString(3), ),
986
( Text(), mysql.MSText(), ),
987
( Unicode(1), mysql.MSString(1), ),
988
( Unicode(3), mysql.MSString(3), ),
989
( UnicodeText(), mysql.MSText(), ),
990
( mysql.MSChar(1), ),
991
( mysql.MSChar(3), ),
992
( NCHAR(2), mysql.MSChar(2), ),
993
( mysql.MSNChar(2), mysql.MSChar(2), ), # N is CREATE only
994
( mysql.MSNVarChar(22), mysql.MSString(22), ),
995
( SmallInteger(), mysql.MSSmallInteger(), ),
996
( SmallInteger(), mysql.MSSmallInteger(4), ),
997
( mysql.MSSmallInteger(), ),
998
( mysql.MSSmallInteger(4), mysql.MSSmallInteger(4), ),
999
( mysql.MSMediumInteger(), mysql.MSMediumInteger(), ),
1000
( mysql.MSMediumInteger(8), mysql.MSMediumInteger(8), ),
1001
( LargeBinary(3), mysql.TINYBLOB(), ),
1002
( LargeBinary(), mysql.BLOB() ),
1003
( mysql.MSBinary(3), mysql.MSBinary(3), ),
1004
( mysql.MSVarBinary(3),),
1005
( mysql.MSTinyBlob(),),
1007
( mysql.MSBlob(1234), mysql.MSBlob()),
1008
( mysql.MSMediumBlob(),),
1009
( mysql.MSLongBlob(),),
1010
( mysql.ENUM("''","'fleem'"), ),
1013
columns = [Column('c%i' % (i + 1), t[0]) for i, t in enumerate(specs)]
1017
t_table = Table('mysql_types', m, *columns)
1022
rt = Table('mysql_types', m2, autoload=True)
1024
db.execute('CREATE OR REPLACE VIEW mysql_types_v '
1025
'AS SELECT * from mysql_types')
1026
rv = Table('mysql_types_v', m2, autoload=True)
1028
expected = [len(c) > 1 and c[1] or c[0] for c in specs]
1030
# Early 5.0 releases seem to report more "general" for columns
1031
# in a view, e.g. char -> varchar, tinyblob -> mediumblob
1033
# Not sure exactly which point version has the fix.
1034
if db.dialect.server_version_info < (5, 0, 11):
1039
for table in tables:
1040
for i, reflected in enumerate(table.c):
1041
assert isinstance(reflected.type,
1042
type(expected[i])), \
1043
'element %d: %r not instance of %r' % (i,
1044
reflected.type, type(expected[i]))
1046
db.execute('DROP VIEW mysql_types_v')
1050
def test_autoincrement(self):
1051
meta = MetaData(testing.db)
1054
Column('int_y', Integer, primary_key=True),
1055
Column('int_n', Integer, DefaultClause('0'),
1057
mysql_engine='MyISAM')
1059
Column('int_y', Integer, primary_key=True),
1060
Column('int_n', Integer, DefaultClause('0'),
1062
mysql_engine='MyISAM')
1064
Column('int_n', Integer, DefaultClause('0'),
1065
primary_key=True, autoincrement=False),
1066
Column('int_y', Integer, primary_key=True),
1067
mysql_engine='MyISAM')
1069
Column('int_n', Integer, DefaultClause('0'),
1070
primary_key=True, autoincrement=False),
1071
Column('int_n2', Integer, DefaultClause('0'),
1072
primary_key=True, autoincrement=False),
1073
mysql_engine='MyISAM')
1075
Column('int_y', Integer, primary_key=True),
1076
Column('int_n', Integer, DefaultClause('0'),
1077
primary_key=True, autoincrement=False),
1078
mysql_engine='MyISAM')
1080
Column('o1', String(1), DefaultClause('x'),
1082
Column('int_y', Integer, primary_key=True),
1083
mysql_engine='MyISAM')
1085
Column('o1', String(1), DefaultClause('x'),
1087
Column('o2', String(1), DefaultClause('x'),
1089
Column('int_y', Integer, primary_key=True),
1090
mysql_engine='MyISAM')
1092
Column('o1', String(1), DefaultClause('x'),
1094
Column('o2', String(1), DefaultClause('x'),
1096
mysql_engine='MyISAM')
1099
table_names = ['ai_1', 'ai_2', 'ai_3', 'ai_4',
1100
'ai_5', 'ai_6', 'ai_7', 'ai_8']
1101
mr = MetaData(testing.db)
1102
mr.reflect(only=table_names)
1104
for tbl in [mr.tables[name] for name in table_names]:
1106
if c.name.startswith('int_y'):
1107
assert c.autoincrement
1108
elif c.name.startswith('int_n'):
1109
assert not c.autoincrement
1110
tbl.insert().execute()
1111
if 'int_y' in tbl.c:
1112
assert select([tbl.c.int_y]).scalar() == 1
1113
assert list(tbl.select().execute().first()).count(1) == 1
1115
assert 1 not in list(tbl.select().execute().first())
1119
@testing.exclude('mysql', '<', (5, 0, 0), 'no information_schema support')
1120
def test_system_views(self):
1121
dialect = testing.db.dialect
1122
connection = testing.db.connect()
1123
view_names = dialect.get_view_names(connection, "information_schema")
1124
self.assert_('TABLES' in view_names)
1127
class SQLTest(fixtures.TestBase, AssertsCompiledSQL):
1128
"""Tests MySQL-dialect specific compilation."""
1130
__dialect__ = mysql.dialect()
1132
def test_precolumns(self):
1133
dialect = self.__dialect__
1135
def gen(distinct=None, prefixes=None):
1137
if distinct is not None:
1138
kw['distinct'] = distinct
1139
if prefixes is not None:
1140
kw['prefixes'] = prefixes
1141
return str(select(['q'], **kw).compile(dialect=dialect))
1143
eq_(gen(None), 'SELECT q')
1144
eq_(gen(True), 'SELECT DISTINCT q')
1147
exc.SADeprecationWarning,
1151
eq_(gen(prefixes=['ALL']), 'SELECT ALL q')
1152
eq_(gen(prefixes=['DISTINCTROW']),
1153
'SELECT DISTINCTROW q')
1155
# Interaction with MySQL prefix extensions
1157
gen(None, ['straight_join']),
1158
'SELECT straight_join q')
1160
gen(False, ['HIGH_PRIORITY', 'SQL_SMALL_RESULT', 'ALL']),
1161
'SELECT HIGH_PRIORITY SQL_SMALL_RESULT ALL q')
1163
gen(True, ['high_priority', sql.text('sql_cache')]),
1164
'SELECT high_priority sql_cache DISTINCT q')
1166
@testing.uses_deprecated
1167
def test_deprecated_distinct(self):
1168
dialect = self.__dialect__
1170
self.assert_compile(
1171
select(['q'], distinct='ALL'),
1175
self.assert_compile(
1176
select(['q'], distinct='distinctROW'),
1177
'SELECT DISTINCTROW q',
1180
self.assert_compile(
1181
select(['q'], distinct='ALL',
1182
prefixes=['HIGH_PRIORITY', 'SQL_SMALL_RESULT']),
1183
'SELECT HIGH_PRIORITY SQL_SMALL_RESULT ALL q'
1186
def test_backslash_escaping(self):
1187
self.assert_compile(
1188
sql.column('foo').like('bar', escape='\\'),
1189
"foo LIKE %s ESCAPE '\\\\'"
1192
dialect = mysql.dialect()
1193
dialect._backslash_escapes=False
1194
self.assert_compile(
1195
sql.column('foo').like('bar', escape='\\'),
1196
"foo LIKE %s ESCAPE '\\'",
1200
def test_limit(self):
1201
t = sql.table('t', sql.column('col1'), sql.column('col2'))
1203
self.assert_compile(
1204
select([t]).limit(10).offset(20),
1205
"SELECT t.col1, t.col2 FROM t LIMIT %s, %s",
1206
{'param_1':20, 'param_2':10}
1208
self.assert_compile(
1209
select([t]).limit(10),
1210
"SELECT t.col1, t.col2 FROM t LIMIT %s",
1213
self.assert_compile(
1214
select([t]).offset(10),
1215
"SELECT t.col1, t.col2 FROM t LIMIT %s, 18446744073709551615",
1219
def test_varchar_raise(self):
1229
type_ = sqltypes.to_instance(type_)
1230
assert_raises_message(
1232
"VARCHAR requires a length on dialect mysql",
1234
dialect=mysql.dialect())
1236
t1 = Table('sometable', MetaData(),
1237
Column('somecolumn', type_)
1239
assert_raises_message(
1241
r"\(in table 'sometable', column 'somecolumn'\)\: "
1242
r"(?:N)?VARCHAR requires a length on dialect mysql",
1243
schema.CreateTable(t1).compile,
1244
dialect=mysql.dialect()
1247
def test_update_limit(self):
1248
t = sql.table('t', sql.column('col1'), sql.column('col2'))
1250
self.assert_compile(
1251
t.update(values={'col1':123}),
1252
"UPDATE t SET col1=%s"
1254
self.assert_compile(
1255
t.update(values={'col1':123}, mysql_limit=5),
1256
"UPDATE t SET col1=%s LIMIT 5"
1258
self.assert_compile(
1259
t.update(values={'col1':123}, mysql_limit=None),
1260
"UPDATE t SET col1=%s"
1262
self.assert_compile(
1263
t.update(t.c.col2==456, values={'col1':123}, mysql_limit=1),
1264
"UPDATE t SET col1=%s WHERE t.col2 = %s LIMIT 1"
1267
def test_utc_timestamp(self):
1268
self.assert_compile(func.utc_timestamp(), "UTC_TIMESTAMP")
1270
def test_sysdate(self):
1271
self.assert_compile(func.sysdate(), "SYSDATE()")
1273
def test_cast(self):
1274
t = sql.table('t', sql.column('col'))
1278
(Integer, "CAST(t.col AS SIGNED INTEGER)"),
1279
(INT, "CAST(t.col AS SIGNED INTEGER)"),
1280
(m.MSInteger, "CAST(t.col AS SIGNED INTEGER)"),
1281
(m.MSInteger(unsigned=True), "CAST(t.col AS UNSIGNED INTEGER)"),
1282
(SmallInteger, "CAST(t.col AS SIGNED INTEGER)"),
1283
(m.MSSmallInteger, "CAST(t.col AS SIGNED INTEGER)"),
1284
(m.MSTinyInteger, "CAST(t.col AS SIGNED INTEGER)"),
1285
# 'SIGNED INTEGER' is a bigint, so this is ok.
1286
(m.MSBigInteger, "CAST(t.col AS SIGNED INTEGER)"),
1287
(m.MSBigInteger(unsigned=False), "CAST(t.col AS SIGNED INTEGER)"),
1288
(m.MSBigInteger(unsigned=True),
1289
"CAST(t.col AS UNSIGNED INTEGER)"),
1292
# this is kind of sucky. thank you default arguments!
1293
(NUMERIC, "CAST(t.col AS DECIMAL)"),
1294
(DECIMAL, "CAST(t.col AS DECIMAL)"),
1295
(Numeric, "CAST(t.col AS DECIMAL)"),
1296
(m.MSNumeric, "CAST(t.col AS DECIMAL)"),
1297
(m.MSDecimal, "CAST(t.col AS DECIMAL)"),
1301
(m.MSFloat, "t.col"),
1302
(m.MSDouble, "t.col"),
1303
(m.MSReal, "t.col"),
1305
(TIMESTAMP, "CAST(t.col AS DATETIME)"),
1306
(DATETIME, "CAST(t.col AS DATETIME)"),
1307
(DATE, "CAST(t.col AS DATE)"),
1308
(TIME, "CAST(t.col AS TIME)"),
1309
(DateTime, "CAST(t.col AS DATETIME)"),
1310
(Date, "CAST(t.col AS DATE)"),
1311
(Time, "CAST(t.col AS TIME)"),
1312
(DateTime, "CAST(t.col AS DATETIME)"),
1313
(Date, "CAST(t.col AS DATE)"),
1314
(m.MSTime, "CAST(t.col AS TIME)"),
1315
(m.MSTimeStamp, "CAST(t.col AS DATETIME)"),
1316
(m.MSYear, "t.col"),
1317
(m.MSYear(2), "t.col"),
1318
(Interval, "t.col"),
1320
(String, "CAST(t.col AS CHAR)"),
1321
(Unicode, "CAST(t.col AS CHAR)"),
1322
(UnicodeText, "CAST(t.col AS CHAR)"),
1323
(VARCHAR, "CAST(t.col AS CHAR)"),
1324
(NCHAR, "CAST(t.col AS CHAR)"),
1325
(CHAR, "CAST(t.col AS CHAR)"),
1326
(CLOB, "CAST(t.col AS CHAR)"),
1327
(TEXT, "CAST(t.col AS CHAR)"),
1328
(String(32), "CAST(t.col AS CHAR(32))"),
1329
(Unicode(32), "CAST(t.col AS CHAR(32))"),
1330
(CHAR(32), "CAST(t.col AS CHAR(32))"),
1331
(m.MSString, "CAST(t.col AS CHAR)"),
1332
(m.MSText, "CAST(t.col AS CHAR)"),
1333
(m.MSTinyText, "CAST(t.col AS CHAR)"),
1334
(m.MSMediumText, "CAST(t.col AS CHAR)"),
1335
(m.MSLongText, "CAST(t.col AS CHAR)"),
1336
(m.MSNChar, "CAST(t.col AS CHAR)"),
1337
(m.MSNVarChar, "CAST(t.col AS CHAR)"),
1339
(LargeBinary, "CAST(t.col AS BINARY)"),
1340
(BLOB, "CAST(t.col AS BINARY)"),
1341
(m.MSBlob, "CAST(t.col AS BINARY)"),
1342
(m.MSBlob(32), "CAST(t.col AS BINARY)"),
1343
(m.MSTinyBlob, "CAST(t.col AS BINARY)"),
1344
(m.MSMediumBlob, "CAST(t.col AS BINARY)"),
1345
(m.MSLongBlob, "CAST(t.col AS BINARY)"),
1346
(m.MSBinary, "CAST(t.col AS BINARY)"),
1347
(m.MSBinary(32), "CAST(t.col AS BINARY)"),
1348
(m.MSVarBinary, "CAST(t.col AS BINARY)"),
1349
(m.MSVarBinary(32), "CAST(t.col AS BINARY)"),
1351
# maybe this could be changed to something more DWIM, needs
1356
(m.MSEnum, "t.col"),
1357
(m.MSEnum("1", "2"), "t.col"),
1359
(m.MSSet("1", "2"), "t.col"),
1362
for type_, expected in specs:
1363
self.assert_compile(cast(t.c.col, type_), expected)
1365
def test_no_cast_pre_4(self):
1366
self.assert_compile(
1367
cast(Column('foo', Integer), String),
1368
"CAST(foo AS CHAR)",
1370
dialect = mysql.dialect()
1371
dialect.server_version_info = (3, 2, 3)
1372
self.assert_compile(
1373
cast(Column('foo', Integer), String),
1378
def test_cast_grouped_expression_non_castable(self):
1379
self.assert_compile(
1380
cast(sql.column('x') + sql.column('y'), Float),
1384
def test_cast_grouped_expression_pre_4(self):
1385
dialect = mysql.dialect()
1386
dialect.server_version_info = (3, 2, 3)
1387
self.assert_compile(
1388
cast(sql.column('x') + sql.column('y'), Integer),
1393
def test_extract(self):
1394
t = sql.table('t', sql.column('col1'))
1396
for field in 'year', 'month', 'day':
1397
self.assert_compile(
1398
select([extract(field, t.c.col1)]),
1399
"SELECT EXTRACT(%s FROM t.col1) AS anon_1 FROM t" % field)
1401
# millsecondS to millisecond
1402
self.assert_compile(
1403
select([extract('milliseconds', t.c.col1)]),
1404
"SELECT EXTRACT(millisecond FROM t.col1) AS anon_1 FROM t")
1406
def test_too_long_index(self):
1407
exp = 'ix_zyrenian_zyme_zyzzogeton_zyzzogeton_zyrenian_zyme_zyz_5cd2'
1408
tname = 'zyrenian_zyme_zyzzogeton_zyzzogeton'
1409
cname = 'zyrenian_zyme_zyzzogeton_zo'
1411
t1 = Table(tname, MetaData(),
1412
Column(cname, Integer, index=True),
1414
ix1 = list(t1.indexes)[0]
1416
self.assert_compile(
1417
schema.CreateIndex(ix1),
1419
"ON %s (%s)" % (exp, tname, cname),
1420
dialect=mysql.dialect()
1423
def test_innodb_autoincrement(self):
1424
t1 = Table('sometable', MetaData(), Column('assigned_id',
1425
Integer(), primary_key=True, autoincrement=False),
1426
Column('id', Integer(), primary_key=True,
1427
autoincrement=True), mysql_engine='InnoDB')
1428
self.assert_compile(schema.CreateTable(t1),
1429
'CREATE TABLE sometable (assigned_id '
1430
'INTEGER NOT NULL, id INTEGER NOT NULL '
1431
'AUTO_INCREMENT, PRIMARY KEY (assigned_id, '
1432
'id), KEY idx_autoinc_id (id))ENGINE=Inn'
1435
t1 = Table('sometable', MetaData(), Column('assigned_id',
1436
Integer(), primary_key=True, autoincrement=True),
1437
Column('id', Integer(), primary_key=True,
1438
autoincrement=False), mysql_engine='InnoDB')
1439
self.assert_compile(schema.CreateTable(t1),
1440
'CREATE TABLE sometable (assigned_id '
1441
'INTEGER NOT NULL AUTO_INCREMENT, id '
1442
'INTEGER NOT NULL, PRIMARY KEY '
1443
'(assigned_id, id))ENGINE=InnoDB')
1445
def test_innodb_autoincrement_reserved_word_column_name(self):
1447
'sometable', MetaData(),
1448
Column('id', Integer(), primary_key=True, autoincrement=False),
1449
Column('order', Integer(), primary_key=True, autoincrement=True),
1450
mysql_engine='InnoDB')
1451
self.assert_compile(
1452
schema.CreateTable(t1),
1453
'CREATE TABLE sometable ('
1454
'id INTEGER NOT NULL, '
1455
'`order` INTEGER NOT NULL AUTO_INCREMENT, '
1456
'PRIMARY KEY (id, `order`), '
1457
'KEY idx_autoinc_order (`order`)'
1461
class SQLModeDetectionTest(fixtures.TestBase):
1462
__only_on__ = 'mysql'
1464
def _options(self, modes):
1465
def connect(con, record):
1466
cursor = con.cursor()
1467
print "DOING THiS:", "set sql_mode='%s'" % (",".join(modes))
1468
cursor.execute("set sql_mode='%s'" % (",".join(modes)))
1469
e = engines.testing_engine(options={
1471
(connect, 'first_connect'),
1472
(connect, 'connect')
1477
def test_backslash_escapes(self):
1478
engine = self._options(['NO_BACKSLASH_ESCAPES'])
1479
c = engine.connect()
1480
assert not engine.dialect._backslash_escapes
1484
engine = self._options([])
1485
c = engine.connect()
1486
assert engine.dialect._backslash_escapes
1490
def test_ansi_quotes(self):
1491
engine = self._options(['ANSI_QUOTES'])
1492
c = engine.connect()
1493
assert engine.dialect._server_ansiquotes
1497
def test_combination(self):
1498
engine = self._options(['ANSI_QUOTES,NO_BACKSLASH_ESCAPES'])
1499
c = engine.connect()
1500
assert engine.dialect._server_ansiquotes
1501
assert not engine.dialect._backslash_escapes
1505
class RawReflectionTest(fixtures.TestBase):
1507
dialect = mysql.dialect()
1508
self.parser = mysql.MySQLTableDefinitionParser(dialect, dialect.identifier_preparer)
1510
def test_key_reflection(self):
1511
regex = self.parser._re_key
1513
assert regex.match(' PRIMARY KEY (`id`),')
1514
assert regex.match(' PRIMARY KEY USING BTREE (`id`),')
1515
assert regex.match(' PRIMARY KEY (`id`) USING BTREE,')
1516
assert regex.match(' PRIMARY KEY (`id`)')
1517
assert regex.match(' PRIMARY KEY USING BTREE (`id`)')
1518
assert regex.match(' PRIMARY KEY (`id`) USING BTREE')
1521
class ExecutionTest(fixtures.TestBase):
1522
"""Various MySQL execution special cases."""
1524
__only_on__ = 'mysql'
1526
def test_charset_caching(self):
1527
engine = engines.testing_engine()
1529
cx = engine.connect()
1531
charset = engine.dialect._detect_charset(cx)
1534
eq_(cx.dialect._connection_charset, charset)
1537
def test_sysdate(self):
1538
d = testing.db.scalar(func.sysdate())
1539
assert isinstance(d, datetime.datetime)
1541
class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
1542
__only_on__ = 'mysql'
1545
def setup_class(cls):
1546
global metadata, cattable, matchtable
1547
metadata = MetaData(testing.db)
1549
cattable = Table('cattable', metadata,
1550
Column('id', Integer, primary_key=True),
1551
Column('description', String(50)),
1552
mysql_engine='MyISAM'
1554
matchtable = Table('matchtable', metadata,
1555
Column('id', Integer, primary_key=True),
1556
Column('title', String(200)),
1557
Column('category_id', Integer, ForeignKey('cattable.id')),
1558
mysql_engine='MyISAM'
1560
metadata.create_all()
1562
cattable.insert().execute([
1563
{'id': 1, 'description': 'Python'},
1564
{'id': 2, 'description': 'Ruby'},
1566
matchtable.insert().execute([
1568
'title': 'Agile Web Development with Rails',
1571
'title': 'Dive Into Python',
1574
'title': "Programming Matz's Ruby",
1577
'title': 'The Definitive Guide to Django',
1580
'title': 'Python in a Nutshell',
1585
def teardown_class(cls):
1588
@testing.fails_on('mysql+mysqlconnector', 'uses pyformat')
1589
def test_expression(self):
1590
format = testing.db.dialect.paramstyle == 'format' and '%s' or '?'
1591
self.assert_compile(
1592
matchtable.c.title.match('somstr'),
1593
"MATCH (matchtable.title) AGAINST (%s IN BOOLEAN MODE)" % format)
1595
@testing.fails_on('mysql+mysqldb', 'uses format')
1596
@testing.fails_on('mysql+pymysql', 'uses format')
1597
@testing.fails_on('mysql+oursql', 'uses format')
1598
@testing.fails_on('mysql+pyodbc', 'uses format')
1599
@testing.fails_on('mysql+zxjdbc', 'uses format')
1600
def test_expression(self):
1601
format = '%(title_1)s'
1602
self.assert_compile(
1603
matchtable.c.title.match('somstr'),
1604
"MATCH (matchtable.title) AGAINST (%s IN BOOLEAN MODE)" % format)
1606
def test_simple_match(self):
1607
results = (matchtable.select().
1608
where(matchtable.c.title.match('python')).
1609
order_by(matchtable.c.id).
1612
eq_([2, 5], [r.id for r in results])
1614
def test_simple_match_with_apostrophe(self):
1615
results = (matchtable.select().
1616
where(matchtable.c.title.match("Matz's")).
1619
eq_([3], [r.id for r in results])
1621
def test_or_match(self):
1622
results1 = (matchtable.select().
1623
where(or_(matchtable.c.title.match('nutshell'),
1624
matchtable.c.title.match('ruby'))).
1625
order_by(matchtable.c.id).
1628
eq_([3, 5], [r.id for r in results1])
1629
results2 = (matchtable.select().
1630
where(matchtable.c.title.match('nutshell ruby')).
1631
order_by(matchtable.c.id).
1634
eq_([3, 5], [r.id for r in results2])
1637
def test_and_match(self):
1638
results1 = (matchtable.select().
1639
where(and_(matchtable.c.title.match('python'),
1640
matchtable.c.title.match('nutshell'))).
1643
eq_([5], [r.id for r in results1])
1644
results2 = (matchtable.select().
1645
where(matchtable.c.title.match('+python +nutshell')).
1648
eq_([5], [r.id for r in results2])
1650
def test_match_across_joins(self):
1651
results = (matchtable.select().
1652
where(and_(cattable.c.id==matchtable.c.category_id,
1653
or_(cattable.c.description.match('Ruby'),
1654
matchtable.c.title.match('nutshell')))).
1655
order_by(matchtable.c.id).
1658
eq_([1, 3, 5], [r.id for r in results])
1662
return testing.db.dialect.ddl_compiler(
1663
testing.db.dialect, None).get_column_specification(c)
added
removed
test/dialect/test_mysql.py
