Merge branch 'master' into pr157

5 files changed, 393 insertions, 138 deletions

diff --git a/test/dialect/mysql/test_compiler.py b/test/dialect/mysql/test_compiler.py
index 304c31012..0571ce526 100644
--- a/test/dialect/mysql/test_compiler.py
+++ b/test/dialect/mysql/test_compiler.py
@@ -184,6 +184,12 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
             schema.CreateTable(t2).compile, dialect=mysql.dialect()
         )
 
+    def test_match(self):
+        matchtable = table('matchtable', column('title', String))
+        self.assert_compile(
+            matchtable.c.title.match('somstr'),
+            "MATCH (matchtable.title) AGAINST (%s IN BOOLEAN MODE)")
+
     def test_for_update(self):
         table1 = table('mytable',
                        column('myid'), column('name'), column('description'))
@@ -511,9 +517,8 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL):
         self.assert_compile(schema.CreateTable(t1),
                             'CREATE TABLE sometable (assigned_id '
                             'INTEGER NOT NULL, id INTEGER NOT NULL '
-                            'AUTO_INCREMENT, PRIMARY KEY (assigned_id, '
-                            'id), KEY idx_autoinc_id (id))ENGINE=Inn'
-                            'oDB')
+                            'AUTO_INCREMENT, PRIMARY KEY (id, assigned_id)'
+                            ')ENGINE=InnoDB')
 
         t1 = Table('sometable', MetaData(),
                    Column('assigned_id', Integer(), primary_key=True,
@@ -537,8 +542,7 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL):
             'CREATE TABLE sometable ('
             'id INTEGER NOT NULL, '
             '`order` INTEGER NOT NULL AUTO_INCREMENT, '
-            'PRIMARY KEY (id, `order`), '
-            'KEY idx_autoinc_order (`order`)'
+            'PRIMARY KEY (`order`, id)'
             ')ENGINE=InnoDB')
 
     def test_create_table_with_partition(self):
diff --git a/test/dialect/mysql/test_dialect.py b/test/dialect/mysql/test_dialect.py
index 03f4c494a..1288b50d7 100644
--- a/test/dialect/mysql/test_dialect.py
+++ b/test/dialect/mysql/test_dialect.py
@@ -8,7 +8,10 @@ from sqlalchemy import testing
 from sqlalchemy.testing import engines
 import datetime
 
+
 class DialectTest(fixtures.TestBase):
+    __backend__ = True
+
     def test_ssl_arguments_mysqldb(self):
         from sqlalchemy.dialects.mysql import mysqldb
         dialect = mysqldb.dialect()
@@ -83,6 +86,17 @@ class DialectTest(fixtures.TestBase):
             )[1]
         eq_(kw['foo'], "true")
 
+    @testing.only_on('mysql')
+    @testing.skip_if('mysql+mysqlconnector', "totally broken for the moment")
+    @testing.fails_on('mysql+oursql', "unsupported")
+    def test_special_encodings(self):
+
+        for enc in ['utf8mb4', 'utf8']:
+            eng = engines.testing_engine(
+                options={"connect_args": {'charset': enc, 'use_unicode': 0}})
+            conn = eng.connect()
+            eq_(conn.dialect._connection_charset, enc)
+
 class SQLModeDetectionTest(fixtures.TestBase):
     __only_on__ = 'mysql'
     __backend__ = True
diff --git a/test/dialect/mysql/test_query.py b/test/dialect/mysql/test_query.py
index f19177c2a..c6b7a1036 100644
--- a/test/dialect/mysql/test_query.py
+++ b/test/dialect/mysql/test_query.py
@@ -2,11 +2,11 @@
 
 from sqlalchemy.testing import eq_, is_
 from sqlalchemy import *
-from sqlalchemy.testing import fixtures, AssertsCompiledSQL
+from sqlalchemy.testing import fixtures
 from sqlalchemy import testing
 
 
-class IdiosyncrasyTest(fixtures.TestBase, AssertsCompiledSQL):
+class IdiosyncrasyTest(fixtures.TestBase):
     __only_on__ = 'mysql'
     __backend__ = True
 
@@ -28,7 +28,7 @@ class IdiosyncrasyTest(fixtures.TestBase, AssertsCompiledSQL):
         )
 
 
-class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
+class MatchTest(fixtures.TestBase):
     __only_on__ = 'mysql'
     __backend__ = True
 
@@ -76,25 +76,6 @@ class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
     def teardown_class(cls):
         metadata.drop_all()
 
-    @testing.fails_on('mysql+mysqlconnector', 'uses pyformat')
-    def test_expression_format(self):
-        format = testing.db.dialect.paramstyle == 'format' and '%s' or '?'
-        self.assert_compile(
-            matchtable.c.title.match('somstr'),
-            "MATCH (matchtable.title) AGAINST (%s IN BOOLEAN MODE)" % format)
-
-    @testing.fails_on('mysql+mysqldb', 'uses format')
-    @testing.fails_on('mysql+pymysql', 'uses format')
-    @testing.fails_on('mysql+cymysql', 'uses format')
-    @testing.fails_on('mysql+oursql', 'uses format')
-    @testing.fails_on('mysql+pyodbc', 'uses format')
-    @testing.fails_on('mysql+zxjdbc', 'uses format')
-    def test_expression_pyformat(self):
-        format = '%(title_1)s'
-        self.assert_compile(
-            matchtable.c.title.match('somstr'),
-            "MATCH (matchtable.title) AGAINST (%s IN BOOLEAN MODE)" % format)
-
     def test_simple_match(self):
         results = (matchtable.select().
                    where(matchtable.c.title.match('python')).
@@ -177,3 +158,57 @@ class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
         eq_([1, 3, 5], [r.id for r in results])
 
 
+class AnyAllTest(fixtures.TablesTest):
+    __only_on__ = 'mysql'
+    __backend__ = True
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            'stuff', metadata,
+            Column('id', Integer, primary_key=True),
+            Column('value', Integer)
+        )
+
+    @classmethod
+    def insert_data(cls):
+        stuff = cls.tables.stuff
+        testing.db.execute(
+            stuff.insert(),
+            [
+                {'id': 1, 'value': 1},
+                {'id': 2, 'value': 2},
+                {'id': 3, 'value': 3},
+                {'id': 4, 'value': 4},
+                {'id': 5, 'value': 5},
+            ]
+        )
+
+    def test_any_w_comparator(self):
+        stuff = self.tables.stuff
+        stmt = select([stuff.c.id]).where(
+            stuff.c.value > any_(select([stuff.c.value])))
+
+        eq_(
+            testing.db.execute(stmt).fetchall(),
+            [(2,), (3,), (4,), (5,)]
+        )
+
+    def test_all_w_comparator(self):
+        stuff = self.tables.stuff
+        stmt = select([stuff.c.id]).where(
+            stuff.c.value >= all_(select([stuff.c.value])))
+
+        eq_(
+            testing.db.execute(stmt).fetchall(),
+            [(5,)]
+        )
+
+    def test_any_literal(self):
+        stuff = self.tables.stuff
+        stmt = select([4 == any_(select([stuff.c.value]))])
+
+        is_(
+            testing.db.execute(stmt).scalar(), True
+        )
+
diff --git a/test/dialect/mysql/test_reflection.py b/test/dialect/mysql/test_reflection.py
index 39b39e006..44880c36b 100644
--- a/test/dialect/mysql/test_reflection.py
+++ b/test/dialect/mysql/test_reflection.py
@@ -1,13 +1,196 @@
 # coding: utf-8
 
-from sqlalchemy.testing import eq_
-from sqlalchemy import *
+from sqlalchemy.testing import eq_, is_
+from sqlalchemy import Column, Table, DDL, MetaData, TIMESTAMP, \
+    DefaultClause, String, Integer, Text, UnicodeText, SmallInteger,\
+    NCHAR, LargeBinary, DateTime, select, UniqueConstraint, Unicode,\
+    BigInteger
+from sqlalchemy import event
 from sqlalchemy import sql
+from sqlalchemy import inspect
 from sqlalchemy.dialects.mysql import base as mysql
+from sqlalchemy.dialects.mysql import reflection as _reflection
 from sqlalchemy.testing import fixtures, AssertsExecutionResults
 from sqlalchemy import testing
 
 
+class TypeReflectionTest(fixtures.TestBase):
+    __only_on__ = 'mysql'
+    __backend__ = True
+
+    @testing.provide_metadata
+    def _run_test(self, specs, attributes):
+        columns = [Column('c%i' % (i + 1), t[0]) for i, t in enumerate(specs)]
+
+        # Early 5.0 releases seem to report more "general" for columns
+        # in a view, e.g. char -> varchar, tinyblob -> mediumblob
+        use_views = testing.db.dialect.server_version_info > (5, 0, 10)
+
+        m = self.metadata
+        Table('mysql_types', m, *columns)
+
+        if use_views:
+            event.listen(
+                m, 'after_create',
+                DDL(
+                    'CREATE OR REPLACE VIEW mysql_types_v '
+                    'AS SELECT * from mysql_types')
+            )
+            event.listen(
+                m, 'before_drop',
+                DDL("DROP VIEW IF EXISTS mysql_types_v")
+            )
+        m.create_all()
+
+        m2 = MetaData(testing.db)
+        tables = [
+            Table('mysql_types', m2, autoload=True)
+        ]
+        if use_views:
+            tables.append(Table('mysql_types_v', m2, autoload=True))
+
+        for table in tables:
+            for i, (reflected_col, spec) in enumerate(zip(table.c, specs)):
+                expected_spec = spec[1]
+                reflected_type = reflected_col.type
+                is_(type(reflected_type), type(expected_spec))
+
+                for attr in attributes:
+                    eq_(
+                        getattr(reflected_type, attr),
+                        getattr(expected_spec, attr),
+                        "Column %s: Attribute %s value of %s does not "
+                        "match %s for type %s" % (
+                            "c%i" % (i + 1),
+                            attr,
+                            getattr(reflected_type, attr),
+                            getattr(expected_spec, attr),
+                            spec[0]
+                        )
+                    )
+
+    def test_time_types(self):
+        specs = []
+
+        if testing.requires.mysql_fsp.enabled:
+            fsps = [None, 0, 5]
+        else:
+            fsps = [None]
+
+        for type_ in (mysql.TIMESTAMP, mysql.DATETIME, mysql.TIME):
+            # MySQL defaults fsp to 0, and if 0 does not report it.
+            # we don't actually render 0 right now in DDL but even if we do,
+            # it comes back blank
+            for fsp in fsps:
+                if fsp:
+                    specs.append((type_(fsp=fsp), type_(fsp=fsp)))
+                else:
+                    specs.append((type_(), type_()))
+
+        specs.extend([
+            (TIMESTAMP(), mysql.TIMESTAMP()),
+            (DateTime(), mysql.DATETIME()),
+        ])
+
+        # note 'timezone' should always be None on both
+        self._run_test(specs, ['fsp', 'timezone'])
+
+    def test_year_types(self):
+        specs = [
+            (mysql.YEAR(), mysql.YEAR(display_width=4)),
+            (mysql.YEAR(display_width=2), mysql.YEAR(display_width=2)),
+            (mysql.YEAR(display_width=4), mysql.YEAR(display_width=4)),
+        ]
+
+        self._run_test(specs, ['display_width'])
+
+    def test_string_types(self):
+        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(1)),
+            (mysql.MSChar(3), mysql.MSChar(3)),
+            (NCHAR(2), mysql.MSChar(2)),
+            (mysql.MSNChar(2), mysql.MSChar(2)),
+            (mysql.MSNVarChar(22), mysql.MSString(22),),
+        ]
+        self._run_test(specs, ['length'])
+
+    def test_integer_types(self):
+        specs = []
+        for type_ in [
+                mysql.TINYINT, mysql.SMALLINT,
+                mysql.MEDIUMINT, mysql.INTEGER, mysql.BIGINT]:
+            for display_width in [None, 4, 7]:
+                for unsigned in [False, True]:
+                    for zerofill in [None, True]:
+                        kw = {}
+                        if display_width:
+                            kw['display_width'] = display_width
+                        if unsigned is not None:
+                            kw['unsigned'] = unsigned
+                        if zerofill is not None:
+                            kw['zerofill'] = zerofill
+
+                        zerofill = bool(zerofill)
+                        source_type = type_(**kw)
+
+                        if display_width is None:
+                            display_width = {
+                                mysql.MEDIUMINT: 9,
+                                mysql.SMALLINT: 6,
+                                mysql.TINYINT: 4,
+                                mysql.INTEGER: 11,
+                                mysql.BIGINT: 20
+                            }[type_]
+
+                        if zerofill:
+                            unsigned = True
+
+                        expected_type = type_(
+                            display_width=display_width,
+                            unsigned=unsigned,
+                            zerofill=zerofill
+                        )
+                        specs.append(
+                            (source_type, expected_type)
+                        )
+
+        specs.extend([
+            (SmallInteger(), mysql.SMALLINT(display_width=6)),
+            (Integer(), mysql.INTEGER(display_width=11)),
+            (BigInteger, mysql.BIGINT(display_width=20))
+        ])
+        self._run_test(specs, ['display_width', 'unsigned', 'zerofill'])
+
+    def test_binary_types(self):
+        specs = [
+            (LargeBinary(3), mysql.TINYBLOB(), ),
+            (LargeBinary(), mysql.BLOB()),
+            (mysql.MSBinary(3), mysql.MSBinary(3), ),
+            (mysql.MSVarBinary(3), mysql.MSVarBinary(3)),
+            (mysql.MSTinyBlob(), mysql.MSTinyBlob()),
+            (mysql.MSBlob(), mysql.MSBlob()),
+            (mysql.MSBlob(1234), mysql.MSBlob()),
+            (mysql.MSMediumBlob(), mysql.MSMediumBlob()),
+            (mysql.MSLongBlob(), mysql.MSLongBlob()),
+        ]
+        self._run_test(specs, [])
+
+    @testing.uses_deprecated('Manually quoting ENUM value literals')
+    def test_legacy_enum_types(self):
+
+        specs = [
+            (mysql.ENUM("''","'fleem'"), mysql.ENUM("''","'fleem'")),  # noqa
+        ]
+
+        self._run_test(specs, ['enums'])
+
+
 class ReflectionTest(fixtures.TestBase, AssertsExecutionResults):
 
     __only_on__ = 'mysql'
@@ -75,7 +258,8 @@ class ReflectionTest(fixtures.TestBase, AssertsExecutionResults):
     def test_reflection_with_table_options(self):
         comment = r"""Comment types type speedily ' " \ '' Fun!"""
 
-        def_table = Table('mysql_def', MetaData(testing.db),
+        def_table = Table(
+            'mysql_def', MetaData(testing.db),
             Column('c1', Integer()),
             mysql_engine='MEMORY',
             mysql_comment=comment,
@@ -88,8 +272,9 @@ class ReflectionTest(fixtures.TestBase, AssertsExecutionResults):
 
         def_table.create()
         try:
-            reflected = Table('mysql_def', MetaData(testing.db),
-                          autoload=True)
+            reflected = Table(
+                'mysql_def', MetaData(testing.db),
+                autoload=True)
         finally:
             def_table.drop()
 
@@ -108,15 +293,16 @@ class ReflectionTest(fixtures.TestBase, AssertsExecutionResults):
         assert reflected.kwargs['mysql_connection'] == 'fish'
 
         # This field doesn't seem to be returned by mysql itself.
-        #assert reflected.kwargs['mysql_password'] == 'secret'
+        # assert reflected.kwargs['mysql_password'] == 'secret'
 
         # This is explicitly ignored when reflecting schema.
-        #assert reflected.kwargs['mysql_auto_increment'] == '5'
+        # assert reflected.kwargs['mysql_auto_increment'] == '5'
 
     def test_reflection_on_include_columns(self):
         """Test reflection of include_columns to be sure they respect case."""
 
-        case_table = Table('mysql_case', MetaData(testing.db),
+        case_table = Table(
+            'mysql_case', MetaData(testing.db),
             Column('c1', String(10)),
             Column('C2', String(10)),
             Column('C3', String(10)))
@@ -128,132 +314,68 @@ class ReflectionTest(fixtures.TestBase, AssertsExecutionResults):
             for t in case_table, reflected:
                 assert 'c1' in t.c.keys()
                 assert 'C2' in t.c.keys()
-            reflected2 = Table('mysql_case', MetaData(testing.db),
-                              autoload=True, include_columns=['c1', 'c2'])
+            reflected2 = Table(
+                'mysql_case', MetaData(testing.db),
+                autoload=True, include_columns=['c1', 'c2'])
             assert 'c1' in reflected2.c.keys()
             for c in ['c2', 'C2', 'C3']:
                 assert c not in reflected2.c.keys()
         finally:
             case_table.drop()
 
-    @testing.exclude('mysql', '<', (5, 0, 0), 'early types are squirrely')
-    @testing.uses_deprecated('Using String type with no length')
-    @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'"), ),
-                 ]
-
-        columns = [Column('c%i' % (i + 1), t[0]) for i, t in enumerate(specs)]
-
-        db = testing.db
-        m = MetaData(db)
-        t_table = Table('mysql_types', m, *columns)
-        try:
-            m.create_all()
-
-            m2 = MetaData(db)
-            rt = Table('mysql_types', m2, autoload=True)
-            try:
-                db.execute('CREATE OR REPLACE VIEW mysql_types_v '
-                           'AS SELECT * from mysql_types')
-                rv = Table('mysql_types_v', m2, autoload=True)
-
-                expected = [len(c) > 1 and c[1] or c[0] for c in specs]
-
-                # Early 5.0 releases seem to report more "general" for columns
-                # in a view, e.g. char -> varchar, tinyblob -> mediumblob
-                #
-                # Not sure exactly which point version has the fix.
-                if db.dialect.server_version_info < (5, 0, 11):
-                    tables = rt,
-                else:
-                    tables = rt, rv
-
-                for table in tables:
-                    for i, reflected in enumerate(table.c):
-                        assert isinstance(reflected.type,
-                                type(expected[i])), \
-                            'element %d: %r not instance of %r' % (i,
-                                reflected.type, type(expected[i]))
-            finally:
-                db.execute('DROP VIEW mysql_types_v')
-        finally:
-            m.drop_all()
-
     def test_autoincrement(self):
         meta = MetaData(testing.db)
         try:
             Table('ai_1', meta,
-                  Column('int_y', Integer, primary_key=True),
+                  Column('int_y', Integer, primary_key=True,
+                         autoincrement=True),
                   Column('int_n', Integer, DefaultClause('0'),
                          primary_key=True),
-                         mysql_engine='MyISAM')
+                  mysql_engine='MyISAM')
             Table('ai_2', meta,
-                  Column('int_y', Integer, primary_key=True),
+                  Column('int_y', Integer, primary_key=True,
+                         autoincrement=True),
                   Column('int_n', Integer, DefaultClause('0'),
                          primary_key=True),
-                         mysql_engine='MyISAM')
+                  mysql_engine='MyISAM')
             Table('ai_3', meta,
                   Column('int_n', Integer, DefaultClause('0'),
                          primary_key=True, autoincrement=False),
-                  Column('int_y', Integer, primary_key=True),
-                         mysql_engine='MyISAM')
+                  Column('int_y', Integer, primary_key=True,
+                         autoincrement=True),
+                  mysql_engine='MyISAM')
             Table('ai_4', meta,
                   Column('int_n', Integer, DefaultClause('0'),
                          primary_key=True, autoincrement=False),
                   Column('int_n2', Integer, DefaultClause('0'),
                          primary_key=True, autoincrement=False),
-                         mysql_engine='MyISAM')
+                  mysql_engine='MyISAM')
             Table('ai_5', meta,
-                  Column('int_y', Integer, primary_key=True),
+                  Column('int_y', Integer, primary_key=True,
+                         autoincrement=True),
                   Column('int_n', Integer, DefaultClause('0'),
                          primary_key=True, autoincrement=False),
-                         mysql_engine='MyISAM')
+                  mysql_engine='MyISAM')
             Table('ai_6', meta,
                   Column('o1', String(1), DefaultClause('x'),
                          primary_key=True),
-                  Column('int_y', Integer, primary_key=True),
-                         mysql_engine='MyISAM')
+                  Column('int_y', Integer, primary_key=True,
+                         autoincrement=True),
+                  mysql_engine='MyISAM')
             Table('ai_7', meta,
                   Column('o1', String(1), DefaultClause('x'),
                          primary_key=True),
                   Column('o2', String(1), DefaultClause('x'),
                          primary_key=True),
-                  Column('int_y', Integer, primary_key=True),
-                         mysql_engine='MyISAM')
+                  Column('int_y', Integer, primary_key=True,
+                         autoincrement=True),
+                  mysql_engine='MyISAM')
             Table('ai_8', meta,
                   Column('o1', String(1), DefaultClause('x'),
                          primary_key=True),
                   Column('o2', String(1), DefaultClause('x'),
                          primary_key=True),
-                         mysql_engine='MyISAM')
+                  mysql_engine='MyISAM')
             meta.create_all()
 
             table_names = ['ai_1', 'ai_2', 'ai_3', 'ai_4',
@@ -276,6 +398,37 @@ class ReflectionTest(fixtures.TestBase, AssertsExecutionResults):
         finally:
             meta.drop_all()
 
+    @testing.provide_metadata
+    def test_view_reflection(self):
+        Table('x', self.metadata, Column('a', Integer), Column('b', String(50)))
+        self.metadata.create_all()
+
+        with testing.db.connect() as conn:
+            conn.execute("CREATE VIEW v1 AS SELECT * FROM x")
+            conn.execute(
+                "CREATE ALGORITHM=MERGE VIEW v2 AS SELECT * FROM x")
+            conn.execute(
+                "CREATE ALGORITHM=UNDEFINED VIEW v3 AS SELECT * FROM x")
+            conn.execute(
+                "CREATE DEFINER=CURRENT_USER VIEW v4 AS SELECT * FROM x")
+
+        @event.listens_for(self.metadata, "before_drop")
+        def cleanup(*arg, **kw):
+            with testing.db.connect() as conn:
+                for v in ['v1', 'v2', 'v3', 'v4']:
+                    conn.execute("DROP VIEW %s" % v)
+
+        insp = inspect(testing.db)
+        for v in ['v1', 'v2', 'v3', 'v4']:
+            eq_(
+                [
+                    (col['name'], col['type'].__class__)
+                    for col in insp.get_columns(v)
+                ],
+                [('a', mysql.INTEGER), ('b', mysql.VARCHAR)]
+            )
+
+
     @testing.exclude('mysql', '<', (5, 0, 0), 'no information_schema support')
     def test_system_views(self):
         dialect = testing.db.dialect
@@ -309,7 +462,7 @@ class ReflectionTest(fixtures.TestBase, AssertsExecutionResults):
             ["t TIMESTAMP"],
             ["u TIMESTAMP DEFAULT CURRENT_TIMESTAMP"]
         ]):
-            Table("nn_t%d" % idx, meta) # to allow DROP
+            Table("nn_t%d" % idx, meta)  # to allow DROP
 
             testing.db.execute("""
                 CREATE TABLE nn_t%d (
@@ -380,7 +533,8 @@ class ReflectionTest(fixtures.TestBase, AssertsExecutionResults):
 class RawReflectionTest(fixtures.TestBase):
     def setup(self):
         dialect = mysql.dialect()
-        self.parser = mysql.MySQLTableDefinitionParser(dialect, dialect.identifier_preparer)
+        self.parser = _reflection.MySQLTableDefinitionParser(
+            dialect, dialect.identifier_preparer)
 
     def test_key_reflection(self):
         regex = self.parser._re_key
@@ -391,10 +545,14 @@ class RawReflectionTest(fixtures.TestBase):
         assert regex.match('  PRIMARY KEY (`id`)')
         assert regex.match('  PRIMARY KEY USING BTREE (`id`)')
         assert regex.match('  PRIMARY KEY (`id`) USING BTREE')
-        assert regex.match('  PRIMARY KEY (`id`) USING BTREE KEY_BLOCK_SIZE 16')
-        assert regex.match('  PRIMARY KEY (`id`) USING BTREE KEY_BLOCK_SIZE=16')
-        assert regex.match('  PRIMARY KEY (`id`) USING BTREE KEY_BLOCK_SIZE  = 16')
-        assert not regex.match('  PRIMARY KEY (`id`) USING BTREE KEY_BLOCK_SIZE = = 16')
+        assert regex.match(
+            '  PRIMARY KEY (`id`) USING BTREE KEY_BLOCK_SIZE 16')
+        assert regex.match(
+            '  PRIMARY KEY (`id`) USING BTREE KEY_BLOCK_SIZE=16')
+        assert regex.match(
+            '  PRIMARY KEY (`id`) USING BTREE KEY_BLOCK_SIZE  = 16')
+        assert not regex.match(
+            '  PRIMARY KEY (`id`) USING BTREE KEY_BLOCK_SIZE = = 16')
 
     def test_fk_reflection(self):
         regex = self.parser._re_constraint
diff --git a/test/dialect/mysql/test_types.py b/test/dialect/mysql/test_types.py
index 7c279ffbf..e570e0db1 100644
--- a/test/dialect/mysql/test_types.py
+++ b/test/dialect/mysql/test_types.py
@@ -1,6 +1,6 @@
 # coding: utf-8
 
-from sqlalchemy.testing import eq_, assert_raises, assert_raises_message
+from sqlalchemy.testing import eq_, assert_raises, assert_raises_message, is_
 from sqlalchemy import *
 from sqlalchemy import sql, exc, schema
 from sqlalchemy.util import u
@@ -10,6 +10,7 @@ from sqlalchemy.testing import fixtures, AssertsCompiledSQL, AssertsExecutionRes
 from sqlalchemy import testing
 import datetime
 import decimal
+from sqlalchemy import types as sqltypes
 
 
 class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
@@ -602,6 +603,49 @@ class TypesTest(fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
             eq_(colspec(table.c.y5), 'y5 YEAR(4)')
 
 
+class JSONTest(fixtures.TestBase):
+    __requires__ = ('json_type', )
+    __only_on__ = 'mysql'
+    __backend__ = True
+
+    @testing.provide_metadata
+    def test_reflection(self):
+
+        Table(
+            'mysql_json', self.metadata,
+            Column('foo', mysql.JSON)
+        )
+        self.metadata.create_all()
+
+        reflected = Table('mysql_json', MetaData(), autoload_with=testing.db)
+        is_(reflected.c.foo.type._type_affinity, sqltypes.JSON)
+        assert isinstance(reflected.c.foo.type, mysql.JSON)
+
+    @testing.provide_metadata
+    def test_rudimental_round_trip(self):
+        # note that test_suite has many more JSON round trip tests
+        # using the backend-agnostic JSON type
+
+        mysql_json = Table(
+            'mysql_json', self.metadata,
+            Column('foo', mysql.JSON)
+        )
+        self.metadata.create_all()
+
+        value = {
+            'json': {'foo': 'bar'},
+            'recs': ['one', 'two']
+        }
+
+        with testing.db.connect() as conn:
+            conn.execute(mysql_json.insert(), foo=value)
+
+            eq_(
+                conn.scalar(select([mysql_json.c.foo])),
+                value
+            )
+
+
 class EnumSetTest(
         fixtures.TestBase, AssertsExecutionResults, AssertsCompiledSQL):
 
@@ -932,12 +976,12 @@ class EnumSetTest(
 
         eq_(
             t2.c.value.type.enums[0:2],
-            (u('réveillé'), u('drôle'))  # u'S’il') # eh ?
+            [u('réveillé'), u('drôle')]  # u'S’il') # eh ?
         )
 
         eq_(
             t2.c.value2.type.enums[0:2],
-            (u('réveillé'), u('drôle'))  # u'S’il') # eh ?
+            [u('réveillé'), u('drôle')]  # u'S’il') # eh ?
         )
 
     def test_enum_compile(self):
@@ -975,13 +1019,13 @@ class EnumSetTest(
         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", "'"))
+            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,))