When testing unicode for MySQL, ensure the connection is utf8

diff --git a/test/sql/unicode.py b/test/sql/unicode.py
index 00bcff7c767ab8f042d0c7f208855397bdb8c588..f47ac97f53b29d31611f02cd3dd55f4fbf59dc58 100644 (file)
--- a/test/sql/unicode.py
+++ b/test/sql/unicode.py
@@ -9,17 +9,19 @@ from testbase import Table, Column
 
 class UnicodeSchemaTest(testbase.PersistTest):
     def setUpAll(self):
-        global metadata, t1, t2
-        metadata = MetaData(testbase.db)
+        global unicode_bind, metadata, t1, t2
+
+        unicode_bind = self._unicode_bind()
+
+        metadata = MetaData(unicode_bind)
         t1 = Table('unitable1', metadata,
             Column(u'méil', Integer, primary_key=True),
-            Column(u'測試', Integer),
+            Column(u'\u6e2c\u8a66', Integer),
 
             )
         t2 = Table(u'unitéble2', metadata,
             Column(u'méil', Integer, primary_key=True, key="a"),
-            Column(u'測試', Integer, ForeignKey(u'unitable1.méil'), key="b"),
-
+            Column(u'\u6e2c\u8a66', Integer, ForeignKey(u'unitable1.méil'), key="b"),
             )
         metadata.create_all()
 
@@ -28,24 +30,46 @@ class UnicodeSchemaTest(testbase.PersistTest):
         t1.delete().execute()
         
     def tearDownAll(self):
+        global unicode_bind
         metadata.drop_all()
+        del unicode_bind
+
+    def _unicode_bind(self):
+        if testbase.db.name != 'mysql':
+            return testbase.db
+        else:
+            # most mysql installations don't default to utf8 connections
+            version = testbase.db.dialect.get_version_info(testbase.db)
+            if version < (4, 1):
+                raise AssertionError("Unicode not supported on MySQL < 4.1")
+
+            c = testbase.db.connect()
+            if not hasattr(c.connection.connection, 'set_character_set'):
+                raise AssertionError(
+                    "Unicode not supported on this MySQL-python version")
+            else: 
+                c.connection.set_character_set('utf8')
+                c.detach()
+
+            return c
         
     def test_insert(self):
-        t1.insert().execute({u'méil':1, u'測試':5})
+        t1.insert().execute({u'méil':1, u'\u6e2c\u8a66':5})
         t2.insert().execute({'a':1, 'b':1})
         
         assert t1.select().execute().fetchall() == [(1, 5)]
         assert t2.select().execute().fetchall() == [(1, 1)]
     
     def test_reflect(self):
-        t1.insert().execute({u'méil':2, u'測試':7})
+        t1.insert().execute({u'méil':2, u'\u6e2c\u8a66':7})
         t2.insert().execute({'a':2, 'b':2})
 
-        meta = MetaData(testbase.db)
+        meta = MetaData(unicode_bind)
         tt1 = Table(t1.name, meta, autoload=True)
         tt2 = Table(t2.name, meta, autoload=True)
-        tt1.insert().execute({u'méil':1, u'測試':5})
-        tt2.insert().execute({u'méil':1, u'測試':1})
+
+        tt1.insert().execute({u'méil':1, u'\u6e2c\u8a66':5})
+        tt2.insert().execute({u'méil':1, u'\u6e2c\u8a66':1})
 
         assert tt1.select(order_by=desc(u'méil')).execute().fetchall() == [(2, 7), (1, 5)]
         assert tt2.select(order_by=desc(u'méil')).execute().fetchall() == [(2, 2), (1, 1)]
@@ -59,7 +83,7 @@ class UnicodeSchemaTest(testbase.PersistTest):
         mapper(A, t1, properties={
             't2s':relation(B),
             'a':t1.c[u'méil'],
-            'b':t1.c[u'測試']
+            'b':t1.c[u'\u6e2c\u8a66']
         })
         mapper(B, t2)
         sess = create_session()