summaryrefslogtreecommitdiff
path: root/test/dialect/mysql
diff options
context:
space:
mode:
Diffstat (limited to 'test/dialect/mysql')
-rw-r--r--test/dialect/mysql/test_compiler.py47
-rw-r--r--test/dialect/mysql/test_dialect.py59
-rw-r--r--test/dialect/mysql/test_reflection.py73
-rw-r--r--test/dialect/mysql/test_types.py299
4 files changed, 355 insertions, 123 deletions
diff --git a/test/dialect/mysql/test_compiler.py b/test/dialect/mysql/test_compiler.py
index a77a25cc4..45f8405c8 100644
--- a/test/dialect/mysql/test_compiler.py
+++ b/test/dialect/mysql/test_compiler.py
@@ -6,6 +6,7 @@ from sqlalchemy import sql, exc, schema, types as sqltypes
from sqlalchemy.dialects.mysql import base as mysql
from sqlalchemy.testing import fixtures, AssertsCompiledSQL
from sqlalchemy import testing
+from sqlalchemy.sql import table, column
class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
@@ -94,19 +95,57 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
"CREATE TABLE testtbl (data VARCHAR(255), "
"PRIMARY KEY (data) USING btree)")
- def test_skip_deferrable_kw(self):
+ def test_create_index_expr(self):
+ m = MetaData()
+ t1 = Table('foo', m,
+ Column('x', Integer)
+ )
+ self.assert_compile(
+ schema.CreateIndex(Index("bar", t1.c.x > 5)),
+ "CREATE INDEX bar ON foo (x > 5)"
+ )
+
+ def test_deferrable_initially_kw_not_ignored(self):
m = MetaData()
t1 = Table('t1', m, Column('id', Integer, primary_key=True))
t2 = Table('t2', m, Column('id', Integer,
- ForeignKey('t1.id', deferrable=True),
+ ForeignKey('t1.id', deferrable=True, initially="XYZ"),
primary_key=True))
self.assert_compile(
schema.CreateTable(t2),
"CREATE TABLE t2 (id INTEGER NOT NULL, "
- "PRIMARY KEY (id), FOREIGN KEY(id) REFERENCES t1 (id))"
+ "PRIMARY KEY (id), FOREIGN KEY(id) REFERENCES t1 (id) DEFERRABLE INITIALLY XYZ)"
)
+ def test_match_kw_raises(self):
+ m = MetaData()
+ t1 = Table('t1', m, Column('id', Integer, primary_key=True))
+ t2 = Table('t2', m, Column('id', Integer,
+ ForeignKey('t1.id', match="XYZ"),
+ primary_key=True))
+
+ assert_raises_message(
+ exc.CompileError,
+ "MySQL ignores the 'MATCH' keyword while at the same time causes "
+ "ON UPDATE/ON DELETE clauses to be ignored.",
+ schema.CreateTable(t2).compile, dialect=mysql.dialect()
+ )
+
+ def test_for_update(self):
+ table1 = table('mytable',
+ column('myid'), column('name'), column('description'))
+
+ self.assert_compile(
+ table1.select(table1.c.myid == 7).with_for_update(),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %s FOR UPDATE")
+
+ self.assert_compile(
+ table1.select(table1.c.myid == 7).with_for_update(read=True),
+ "SELECT mytable.myid, mytable.name, mytable.description "
+ "FROM mytable WHERE mytable.myid = %s LOCK IN SHARE MODE")
+
class SQLTest(fixtures.TestBase, AssertsCompiledSQL):
"""Tests MySQL-dialect specific compilation."""
@@ -302,8 +341,10 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL):
(VARCHAR, "CAST(t.col AS CHAR)"),
(NCHAR, "CAST(t.col AS CHAR)"),
(CHAR, "CAST(t.col AS CHAR)"),
+ (m.CHAR(charset='utf8'), "CAST(t.col AS CHAR CHARACTER SET utf8)"),
(CLOB, "CAST(t.col AS CHAR)"),
(TEXT, "CAST(t.col AS CHAR)"),
+ (m.TEXT(charset='utf8'), "CAST(t.col AS CHAR CHARACTER SET utf8)"),
(String(32), "CAST(t.col AS CHAR(32))"),
(Unicode(32), "CAST(t.col AS CHAR(32))"),
(CHAR(32), "CAST(t.col AS CHAR(32))"),
diff --git a/test/dialect/mysql/test_dialect.py b/test/dialect/mysql/test_dialect.py
index 62bdfc81b..2ff17f0f7 100644
--- a/test/dialect/mysql/test_dialect.py
+++ b/test/dialect/mysql/test_dialect.py
@@ -9,12 +9,17 @@ from sqlalchemy.testing import engines
import datetime
class DialectTest(fixtures.TestBase):
- __only_on__ = 'mysql'
+ def test_ssl_arguments_mysqldb(self):
+ from sqlalchemy.dialects.mysql import mysqldb
+ dialect = mysqldb.dialect()
+ self._test_ssl_arguments(dialect)
- @testing.only_on(['mysql+mysqldb', 'mysql+oursql'],
- 'requires particular SSL arguments')
- def test_ssl_arguments(self):
- dialect = testing.db.dialect
+ def test_ssl_arguments_oursql(self):
+ from sqlalchemy.dialects.mysql import oursql
+ dialect = oursql.dialect()
+ self._test_ssl_arguments(dialect)
+
+ def _test_ssl_arguments(self, dialect):
kwarg = dialect.create_connect_args(
make_url("mysql://scott:tiger@localhost:3306/test"
"?ssl_ca=/ca.pem&ssl_cert=/cert.pem&ssl_key=/key.pem")
@@ -33,6 +38,50 @@ class DialectTest(fixtures.TestBase):
}
)
+ def test_mysqlconnector_buffered_arg(self):
+ from sqlalchemy.dialects.mysql import mysqlconnector
+ dialect = mysqlconnector.dialect()
+ kw = dialect.create_connect_args(
+ make_url("mysql+mysqlconnector://u:p@host/db?buffered=true")
+ )[1]
+ eq_(kw['buffered'], True)
+
+ kw = dialect.create_connect_args(
+ make_url("mysql+mysqlconnector://u:p@host/db?buffered=false")
+ )[1]
+ eq_(kw['buffered'], False)
+
+ kw = dialect.create_connect_args(
+ make_url("mysql+mysqlconnector://u:p@host/db")
+ )[1]
+ eq_(kw['buffered'], True)
+
+ def test_mysqlconnector_raise_on_warnings_arg(self):
+ from sqlalchemy.dialects.mysql import mysqlconnector
+ dialect = mysqlconnector.dialect()
+ kw = dialect.create_connect_args(
+ make_url("mysql+mysqlconnector://u:p@host/db?raise_on_warnings=true")
+ )[1]
+ eq_(kw['raise_on_warnings'], True)
+
+ kw = dialect.create_connect_args(
+ make_url("mysql+mysqlconnector://u:p@host/db?raise_on_warnings=false")
+ )[1]
+ eq_(kw['raise_on_warnings'], False)
+
+ kw = dialect.create_connect_args(
+ make_url("mysql+mysqlconnector://u:p@host/db")
+ )[1]
+ eq_(kw['raise_on_warnings'], True)
+
+ @testing.only_on('mysql')
+ def test_random_arg(self):
+ dialect = testing.db.dialect
+ kw = dialect.create_connect_args(
+ make_url("mysql://u:p@host/db?foo=true")
+ )[1]
+ eq_(kw['foo'], "true")
+
class SQLModeDetectionTest(fixtures.TestBase):
__only_on__ = 'mysql'
diff --git a/test/dialect/mysql/test_reflection.py b/test/dialect/mysql/test_reflection.py
index b9e347d41..7494eaf43 100644
--- a/test/dialect/mysql/test_reflection.py
+++ b/test/dialect/mysql/test_reflection.py
@@ -140,33 +140,33 @@ class ReflectionTest(fixtures.TestBase, AssertsExecutionResults):
@testing.uses_deprecated('Manually quoting ENUM value literals')
def test_type_reflection(self):
# (ask_for, roundtripped_as_if_different)
- specs = [( String(1), mysql.MSString(1), ),
- ( String(3), mysql.MSString(3), ),
- ( Text(), mysql.MSText(), ),
- ( Unicode(1), mysql.MSString(1), ),
- ( Unicode(3), mysql.MSString(3), ),
- ( UnicodeText(), mysql.MSText(), ),
- ( mysql.MSChar(1), ),
- ( mysql.MSChar(3), ),
- ( NCHAR(2), mysql.MSChar(2), ),
- ( mysql.MSNChar(2), mysql.MSChar(2), ), # N is CREATE only
- ( mysql.MSNVarChar(22), mysql.MSString(22), ),
- ( SmallInteger(), mysql.MSSmallInteger(), ),
- ( SmallInteger(), mysql.MSSmallInteger(4), ),
- ( mysql.MSSmallInteger(), ),
- ( mysql.MSSmallInteger(4), mysql.MSSmallInteger(4), ),
- ( mysql.MSMediumInteger(), mysql.MSMediumInteger(), ),
- ( mysql.MSMediumInteger(8), mysql.MSMediumInteger(8), ),
- ( LargeBinary(3), mysql.TINYBLOB(), ),
- ( LargeBinary(), mysql.BLOB() ),
- ( mysql.MSBinary(3), mysql.MSBinary(3), ),
- ( mysql.MSVarBinary(3),),
- ( mysql.MSTinyBlob(),),
- ( mysql.MSBlob(),),
- ( mysql.MSBlob(1234), mysql.MSBlob()),
- ( mysql.MSMediumBlob(),),
- ( mysql.MSLongBlob(),),
- ( mysql.ENUM("''","'fleem'"), ),
+ specs = [(String(1), mysql.MSString(1), ),
+ (String(3), mysql.MSString(3), ),
+ (Text(), mysql.MSText(), ),
+ (Unicode(1), mysql.MSString(1), ),
+ (Unicode(3), mysql.MSString(3), ),
+ (UnicodeText(), mysql.MSText(), ),
+ (mysql.MSChar(1), ),
+ (mysql.MSChar(3), ),
+ (NCHAR(2), mysql.MSChar(2), ),
+ (mysql.MSNChar(2), mysql.MSChar(2), ), # N is CREATE only
+ (mysql.MSNVarChar(22), mysql.MSString(22), ),
+ (SmallInteger(), mysql.MSSmallInteger(), ),
+ (SmallInteger(), mysql.MSSmallInteger(4), ),
+ (mysql.MSSmallInteger(), ),
+ (mysql.MSSmallInteger(4), mysql.MSSmallInteger(4), ),
+ (mysql.MSMediumInteger(), mysql.MSMediumInteger(), ),
+ (mysql.MSMediumInteger(8), mysql.MSMediumInteger(8), ),
+ (LargeBinary(3), mysql.TINYBLOB(), ),
+ (LargeBinary(), mysql.BLOB() ),
+ (mysql.MSBinary(3), mysql.MSBinary(3), ),
+ (mysql.MSVarBinary(3),),
+ (mysql.MSTinyBlob(),),
+ (mysql.MSBlob(),),
+ (mysql.MSBlob(1234), mysql.MSBlob()),
+ (mysql.MSMediumBlob(),),
+ (mysql.MSLongBlob(),),
+ (mysql.ENUM("''","'fleem'"), ),
]
columns = [Column('c%i' % (i + 1), t[0]) for i, t in enumerate(specs)]
@@ -298,3 +298,22 @@ class RawReflectionTest(fixtures.TestBase):
assert regex.match(' PRIMARY KEY USING BTREE (`id`)')
assert regex.match(' PRIMARY KEY (`id`) USING BTREE')
+ def test_fk_reflection(self):
+ regex = self.parser._re_constraint
+
+ m = regex.match(' CONSTRAINT `addresses_user_id_fkey` '
+ 'FOREIGN KEY (`user_id`) '
+ 'REFERENCES `users` (`id`) '
+ 'ON DELETE CASCADE ON UPDATE CASCADE')
+ eq_(m.groups(), ('addresses_user_id_fkey', '`user_id`',
+ '`users`', '`id`', None, 'CASCADE', 'CASCADE'))
+
+
+ m = regex.match(' CONSTRAINT `addresses_user_id_fkey` '
+ 'FOREIGN KEY (`user_id`) '
+ 'REFERENCES `users` (`id`) '
+ 'ON DELETE CASCADE ON UPDATE SET NULL')
+ eq_(m.groups(), ('addresses_user_id_fkey', '`user_id`',
+ '`users`', '`id`', None, 'CASCADE', 'SET NULL'))
+
+
diff --git a/test/dialect/mysql/test_types.py b/test/dialect/mysql/test_types.py
index b918abe25..acf9c1e2f 100644
--- a/test/dialect/mysql/test_types.py
+++ b/test/dialect/mysql/test_types.py
@@ -4,12 +4,13 @@ from sqlalchemy.testing import eq_, assert_raises
from sqlalchemy import *
from sqlalchemy import sql, exc, schema
from sqlalchemy.util import u
+from sqlalchemy import util
from sqlalchemy.dialects.mysql import base as mysql
from sqlalchemy.testing import fixtures, AssertsCompiledSQL, AssertsExecutionResults
from sqlalchemy import testing
from sqlalchemy.testing.engines import utf8_engine
import datetime
-
+import decimal
class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
"Test MySQL column types"
@@ -141,10 +142,36 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
]
for type_, args, kw, res in columns:
+ type_inst = type_(*args, **kw)
self.assert_compile(
- type_(*args, **kw),
+ type_inst,
res
)
+ # test that repr() copies out all arguments
+ self.assert_compile(
+ eval("mysql.%r" % type_inst),
+ res
+ )
+
+ @testing.only_if('mysql')
+ @testing.provide_metadata
+ def test_precision_float_roundtrip(self):
+ t = Table('t', self.metadata,
+ Column('scale_value', mysql.DOUBLE(
+ precision=15, scale=12, asdecimal=True)),
+ Column('unscale_value', mysql.DOUBLE(
+ decimal_return_scale=12, asdecimal=True))
+ )
+ t.create(testing.db)
+ testing.db.execute(
+ t.insert(), scale_value=45.768392065789,
+ unscale_value=45.768392065789
+ )
+ result = testing.db.scalar(select([t.c.scale_value]))
+ eq_(result, decimal.Decimal("45.768392065789"))
+
+ result = testing.db.scalar(select([t.c.unscale_value]))
+ eq_(result, decimal.Decimal("45.768392065789"))
@testing.exclude('mysql', '<', (4, 1, 1), 'no charset support')
def test_charset(self):
@@ -212,14 +239,22 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
(mysql.ENUM, ["foo", "bar"], {'unicode':True},
'''ENUM('foo','bar') UNICODE'''),
- (String, [20], {"collation":"utf8"}, 'VARCHAR(20) COLLATE utf8')
+ (String, [20], {"collation": "utf8"}, 'VARCHAR(20) COLLATE utf8')
]
for type_, args, kw, res in columns:
+ type_inst = type_(*args, **kw)
self.assert_compile(
- type_(*args, **kw),
+ type_inst,
+ res
+ )
+ # test that repr() copies out all arguments
+ self.assert_compile(
+ eval("mysql.%r" % type_inst)
+ if type_ is not String
+ else eval("%r" % type_inst),
res
)
@@ -229,15 +264,23 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
def test_charset_collate_table(self):
t = Table('foo', self.metadata,
Column('id', Integer),
+ Column('data', UnicodeText),
mysql_default_charset='utf8',
- mysql_collate='utf8_unicode_ci'
+ mysql_collate='utf8_bin'
)
t.create()
m2 = MetaData(testing.db)
t2 = Table('foo', m2, autoload=True)
- eq_(t2.kwargs['mysql_collate'], 'utf8_unicode_ci')
+ eq_(t2.kwargs['mysql_collate'], 'utf8_bin')
eq_(t2.kwargs['mysql_default charset'], 'utf8')
+ # test [ticket:2906]
+ # in order to test the condition here, need to use
+ # MySQLdb 1.2.3 and also need to pass either use_unicode=1
+ # or charset=utf8 to the URL.
+ t.insert().execute(id=1, data=u('some text'))
+ assert isinstance(testing.db.scalar(select([t.c.data])), util.text_type)
+
def test_bit_50(self):
"""Exercise BIT types on 5.0+ (not valid for all engine types)"""
@@ -250,7 +293,9 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
@testing.only_if('mysql')
@testing.exclude('mysql', '<', (5, 0, 5), 'a 5.0+ feature')
- @testing.fails_on('mysql+oursql', 'some round trips fail, oursql bug ?')
+ @testing.fails_if(
+ lambda: testing.against("mysql+oursql") and util.py3k,
+ 'some round trips fail, oursql bug ?')
@testing.provide_metadata
def test_bit_50_roundtrip(self):
bit_table = Table('mysql_bits', self.metadata,
@@ -474,72 +519,24 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
self.assert_(colspec(table.c.y1).startswith('y1 YEAR'))
eq_(colspec(table.c.y5), 'y5 YEAR(4)')
- @testing.only_if('mysql')
- @testing.provide_metadata
- def test_set(self):
- """Exercise the SET type."""
- set_table = Table('mysql_set', self.metadata,
- Column('s1',
- mysql.MSSet("'dq'", "'sq'")), Column('s2',
- mysql.MSSet("'a'")), Column('s3',
- mysql.MSSet("'5'", "'7'", "'9'")))
- eq_(colspec(set_table.c.s1), "s1 SET('dq','sq')")
- eq_(colspec(set_table.c.s2), "s2 SET('a')")
- eq_(colspec(set_table.c.s3), "s3 SET('5','7','9')")
- set_table.create()
- reflected = Table('mysql_set', MetaData(testing.db),
- autoload=True)
- for table in set_table, reflected:
-
- def roundtrip(store, expected=None):
- expected = expected or store
- table.insert(store).execute()
- row = table.select().execute().first()
- self.assert_(list(row) == expected)
- table.delete().execute()
-
- roundtrip([None, None, None], [None] * 3)
- roundtrip(['', '', ''], [set([''])] * 3)
- roundtrip([set(['dq']), set(['a']), set(['5'])])
- roundtrip(['dq', 'a', '5'], [set(['dq']), set(['a']),
- set(['5'])])
- roundtrip([1, 1, 1], [set(['dq']), set(['a']), set(['5'
- ])])
- roundtrip([set(['dq', 'sq']), None, set(['9', '5', '7'
- ])])
- set_table.insert().execute({'s3': set(['5'])},
- {'s3': set(['5', '7'])}, {'s3': set(['5', '7', '9'])},
- {'s3': set(['7', '9'])})
-
- # NOTE: the string sent to MySQL here is sensitive to ordering.
- # for some reason the set ordering is always "5, 7" when we test on
- # MySQLdb but in Py3K this is not guaranteed. So basically our
- # SET type doesn't do ordering correctly (not sure how it can,
- # as we don't know how the SET was configured in the first place.)
- rows = select([set_table.c.s3],
- set_table.c.s3.in_([set(['5']), ['5', '7']])
- ).execute().fetchall()
- found = set([frozenset(row[0]) for row in rows])
- eq_(found, set([frozenset(['5']), frozenset(['5', '7'])]))
-
-class EnumTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
+class EnumSetTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
__only_on__ = 'mysql'
__dialect__ = mysql.dialect()
- @testing.uses_deprecated('Manually quoting ENUM value literals')
@testing.provide_metadata
def test_enum(self):
"""Exercise the ENUM type."""
+ with testing.expect_deprecated('Manually quoting ENUM value literals'):
+ e1, e2 = mysql.ENUM("'a'", "'b'"), mysql.ENUM("'a'", "'b'")
+
enum_table = Table('mysql_enum', self.metadata,
- Column('e1', mysql.ENUM("'a'", "'b'")),
- Column('e2', mysql.ENUM("'a'", "'b'"),
- nullable=False),
- Column('e2generic', Enum("a", "b"),
- nullable=False),
+ Column('e1', e1),
+ Column('e2', e2, nullable=False),
+ Column('e2generic', Enum("a", "b"), nullable=False),
Column('e3', mysql.ENUM("'a'", "'b'", strict=True)),
Column('e4', mysql.ENUM("'a'", "'b'", strict=True),
nullable=False),
@@ -587,6 +584,106 @@ class EnumTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
eq_(res, expected)
+ @testing.provide_metadata
+ def test_set(self):
+
+ with testing.expect_deprecated('Manually quoting SET value literals'):
+ e1, e2 = mysql.SET("'a'", "'b'"), mysql.SET("'a'", "'b'")
+
+ set_table = Table('mysql_set', self.metadata,
+ Column('e1', e1),
+ Column('e2', e2, nullable=False),
+ Column('e3', mysql.SET("a", "b")),
+ Column('e4', mysql.SET("'a'", "b")),
+ Column('e5', mysql.SET("'a'", "'b'", quoting="quoted"))
+ )
+
+ eq_(colspec(set_table.c.e1),
+ "e1 SET('a','b')")
+ eq_(colspec(set_table.c.e2),
+ "e2 SET('a','b') NOT NULL")
+ eq_(colspec(set_table.c.e3),
+ "e3 SET('a','b')")
+ eq_(colspec(set_table.c.e4),
+ "e4 SET('''a''','b')")
+ eq_(colspec(set_table.c.e5),
+ "e5 SET('a','b')")
+ set_table.create()
+
+ assert_raises(exc.DBAPIError, set_table.insert().execute,
+ e1=None, e2=None, e3=None, e4=None)
+
+ if testing.against("+oursql"):
+ assert_raises(exc.StatementError, set_table.insert().execute,
+ e1='c', e2='c', e3='c', e4='c')
+
+ set_table.insert().execute(e1='a', e2='a', e3='a', e4="'a'", e5="a,b")
+ set_table.insert().execute(e1='b', e2='b', e3='b', e4='b', e5="a,b")
+
+ res = set_table.select().execute().fetchall()
+
+ if testing.against("+oursql"):
+ expected = [
+ # 1st row with all c's, data truncated
+ (set(['']), set(['']), set(['']), set(['']), None),
+ ]
+ else:
+ expected = []
+
+ expected.extend([
+ (set(['a']), set(['a']), set(['a']), set(["'a'"]), set(['a', 'b'])),
+ (set(['b']), set(['b']), set(['b']), set(['b']), set(['a', 'b']))
+ ])
+
+ eq_(res, expected)
+
+ @testing.provide_metadata
+ def test_set_roundtrip_plus_reflection(self):
+ set_table = Table('mysql_set', self.metadata,
+ Column('s1',
+ mysql.SET("dq", "sq")),
+ Column('s2', mysql.SET("a")),
+ Column('s3', mysql.SET("5", "7", "9")))
+
+ eq_(colspec(set_table.c.s1), "s1 SET('dq','sq')")
+ eq_(colspec(set_table.c.s2), "s2 SET('a')")
+ eq_(colspec(set_table.c.s3), "s3 SET('5','7','9')")
+ set_table.create()
+ reflected = Table('mysql_set', MetaData(testing.db),
+ autoload=True)
+ for table in set_table, reflected:
+
+ def roundtrip(store, expected=None):
+ expected = expected or store
+ table.insert(store).execute()
+ row = table.select().execute().first()
+ self.assert_(list(row) == expected)
+ table.delete().execute()
+
+ roundtrip([None, None, None], [None] * 3)
+ roundtrip(['', '', ''], [set([''])] * 3)
+ roundtrip([set(['dq']), set(['a']), set(['5'])])
+ roundtrip(['dq', 'a', '5'], [set(['dq']), set(['a']),
+ set(['5'])])
+ roundtrip([1, 1, 1], [set(['dq']), set(['a']), set(['5'
+ ])])
+ roundtrip([set(['dq', 'sq']), None, set(['9', '5', '7'
+ ])])
+ set_table.insert().execute({'s3': set(['5'])},
+ {'s3': set(['5', '7'])}, {'s3': set(['5', '7', '9'])},
+ {'s3': set(['7', '9'])})
+
+ # NOTE: the string sent to MySQL here is sensitive to ordering.
+ # for some reason the set ordering is always "5, 7" when we test on
+ # MySQLdb but in Py3K this is not guaranteed. So basically our
+ # SET type doesn't do ordering correctly (not sure how it can,
+ # as we don't know how the SET was configured in the first place.)
+ rows = select([set_table.c.s3],
+ set_table.c.s3.in_([set(['5']), ['5', '7']])
+ ).execute().fetchall()
+ found = set([frozenset(row[0]) for row in rows])
+ eq_(found, set([frozenset(['5']), frozenset(['5', '7'])]))
+
def test_unicode_enum(self):
unicode_engine = utf8_engine()
metadata = MetaData(unicode_engine)
@@ -634,38 +731,64 @@ class EnumTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
"VARCHAR(1), CHECK (somecolumn IN ('x', "
"'y', 'z')))")
+ @testing.provide_metadata
@testing.exclude('mysql', '<', (4,), "3.23 can't handle an ENUM of ''")
- @testing.uses_deprecated('Manually quoting ENUM value literals')
def test_enum_parse(self):
- """More exercises for the ENUM type."""
- # MySQL 3.23 can't handle an ENUM of ''....
-
- enum_table = Table('mysql_enum', MetaData(testing.db),
- Column('e1', mysql.ENUM("'a'")),
- Column('e2', mysql.ENUM("''")),
- Column('e3', mysql.ENUM('a')),
- Column('e4', mysql.ENUM('')),
- Column('e5', mysql.ENUM("'a'", "''")),
- Column('e6', mysql.ENUM("''", "'a'")),
- Column('e7', mysql.ENUM("''", "'''a'''", "'b''b'", "''''")))
+ with testing.expect_deprecated('Manually quoting ENUM value literals'):
+ enum_table = Table('mysql_enum', self.metadata,
+ Column('e1', mysql.ENUM("'a'")),
+ Column('e2', mysql.ENUM("''")),
+ Column('e3', mysql.ENUM('a')),
+ Column('e4', mysql.ENUM('')),
+ Column('e5', mysql.ENUM("'a'", "''")),
+ Column('e6', mysql.ENUM("''", "'a'")),
+ Column('e7', mysql.ENUM("''", "'''a'''", "'b''b'", "''''")))
for col in enum_table.c:
self.assert_(repr(col))
- try:
- enum_table.create()
- reflected = Table('mysql_enum', MetaData(testing.db),
- autoload=True)
- for t in enum_table, reflected:
- eq_(t.c.e1.type.enums, ("a",))
- eq_(t.c.e2.type.enums, ("",))
- eq_(t.c.e3.type.enums, ("a",))
- eq_(t.c.e4.type.enums, ("",))
- eq_(t.c.e5.type.enums, ("a", ""))
- eq_(t.c.e6.type.enums, ("", "a"))
- eq_(t.c.e7.type.enums, ("", "'a'", "b'b", "'"))
- finally:
- enum_table.drop()
+
+ enum_table.create()
+ reflected = Table('mysql_enum', MetaData(testing.db),
+ autoload=True)
+ for t in enum_table, reflected:
+ eq_(t.c.e1.type.enums, ("a",))
+ eq_(t.c.e2.type.enums, ("",))
+ eq_(t.c.e3.type.enums, ("a",))
+ eq_(t.c.e4.type.enums, ("",))
+ eq_(t.c.e5.type.enums, ("a", ""))
+ eq_(t.c.e6.type.enums, ("", "a"))
+ eq_(t.c.e7.type.enums, ("", "'a'", "b'b", "'"))
+
+ @testing.provide_metadata
+ @testing.exclude('mysql', '<', (5,))
+ def test_set_parse(self):
+ with testing.expect_deprecated('Manually quoting SET value literals'):
+ set_table = Table('mysql_set', self.metadata,
+ Column('e1', mysql.SET("'a'")),
+ Column('e2', mysql.SET("''")),
+ Column('e3', mysql.SET('a')),
+ Column('e4', mysql.SET('')),
+ Column('e5', mysql.SET("'a'", "''")),
+ Column('e6', mysql.SET("''", "'a'")),
+ Column('e7', mysql.SET("''", "'''a'''", "'b''b'", "''''")))
+
+ for col in set_table.c:
+ self.assert_(repr(col))
+
+ set_table.create()
+
+ # don't want any warnings on reflection
+ reflected = Table('mysql_set', MetaData(testing.db),
+ autoload=True)
+ for t in set_table, reflected:
+ eq_(t.c.e1.type.values, ("a",))
+ eq_(t.c.e2.type.values, ("",))
+ eq_(t.c.e3.type.values, ("a",))
+ eq_(t.c.e4.type.values, ("",))
+ eq_(t.c.e5.type.values, ("a", ""))
+ eq_(t.c.e6.type.values, ("", "a"))
+ eq_(t.c.e7.type.values, ("", "'a'", "b'b", "'"))
def colspec(c):
return testing.db.dialect.ddl_compiler(
View Lorry Controller Status