Added `set_types` to util, a tuple of available set implementations.

Added BIT and SET ([ticket:674])- all mysql data types are now covered!
Fix for YEAR DDL generation, also no longer a concatenable type.
Expanded docs for some mysql column esoterica.

1 files changed, 263 insertions, 16 deletions

diff --git a/test/dialect/mysql.py b/test/dialect/mysql.py
index 014d98c7e..fcf1de449 100644
--- a/test/dialect/mysql.py
+++ b/test/dialect/mysql.py
@@ -1,4 +1,5 @@
 import testbase
+import sets
 from sqlalchemy import *
 from sqlalchemy.databases import mysql
 from testlib import *
@@ -157,8 +158,8 @@ class TypesTest(AssertMixin):
         
         for col in numeric_table.c:
             index = int(col.name[1:])
-            self.assertEquals(gen.get_column_specification(col),
-                              "%s %s" % (col.name, columns[index][3]))
+            self.assert_eq(gen.get_column_specification(col),
+                           "%s %s" % (col.name, columns[index][3]))
 
         try:
             numeric_table.create(checkfirst=True)
@@ -170,8 +171,7 @@ class TypesTest(AssertMixin):
     @testing.supported('mysql')
     @testing.exclude('mysql', '<', (4, 1, 1))
     def test_charset(self):
-        """Exercise CHARACTER SET and COLLATE-related options on string-type
-        columns."""
+        """Exercise CHARACTER SET and COLLATE-ish options on string types."""
 
         columns = [
             (mysql.MSChar, [1], {},
@@ -242,8 +242,8 @@ class TypesTest(AssertMixin):
         
         for col in charset_table.c:
             index = int(col.name[1:])
-            self.assertEquals(gen.get_column_specification(col),
-                              "%s %s" % (col.name, columns[index][3]))
+            self.assert_eq(gen.get_column_specification(col),
+                           "%s %s" % (col.name, columns[index][3]))
 
         try:
             charset_table.create(checkfirst=True)
@@ -253,21 +253,258 @@ class TypesTest(AssertMixin):
         charset_table.drop()
 
     @testing.supported('mysql')
+    @testing.exclude('mysql', '<', (5, 0, 5))
+    def test_bit_50(self):
+        """Exercise BIT types on 5.0+ (not valid for all engine types)"""
+        
+        meta = MetaData(testbase.db)
+        bit_table = Table('mysql_bits', meta,
+                          Column('b1', mysql.MSBit),
+                          Column('b2', mysql.MSBit()),
+                          Column('b3', mysql.MSBit(), nullable=False),
+                          Column('b4', mysql.MSBit(1)),
+                          Column('b5', mysql.MSBit(8)),
+                          Column('b6', mysql.MSBit(32)),
+                          Column('b7', mysql.MSBit(63)),
+                          Column('b8', mysql.MSBit(64)))
+
+        self.assert_eq(colspec(bit_table.c.b1), 'b1 BIT')
+        self.assert_eq(colspec(bit_table.c.b2), 'b2 BIT')
+        self.assert_eq(colspec(bit_table.c.b3), 'b3 BIT NOT NULL')
+        self.assert_eq(colspec(bit_table.c.b4), 'b4 BIT(1)')
+        self.assert_eq(colspec(bit_table.c.b5), 'b5 BIT(8)')
+        self.assert_eq(colspec(bit_table.c.b6), 'b6 BIT(32)')
+        self.assert_eq(colspec(bit_table.c.b7), 'b7 BIT(63)')
+        self.assert_eq(colspec(bit_table.c.b8), 'b8 BIT(64)')
+
+        try:
+            meta.create_all()
+
+            meta2 = MetaData(testbase.db)
+            reflected = Table('mysql_bits', meta2, autoload=True)
+
+            for table in bit_table, reflected:
+
+                def roundtrip(store, expected=None):
+                    expected = expected or store
+                    table.insert(store).execute()
+                    row = list(table.select().execute())[0]
+                    try:
+                        self.assert_(list(row) == expected)
+                    except:
+                        print "Storing %s" % store
+                        print "Expected %s" % expected
+                        print "Found %s" % list(row)
+                        raise
+                    table.delete().execute()
+
+                roundtrip([0] * 8)
+                roundtrip([None, None, 0, None, None, None, None, None])
+                roundtrip([1] * 8)
+                roundtrip([sql.text("b'1'")] * 8, [1] * 8)
+
+                i = 255
+                roundtrip([0, 0, 0, 0, i, i, i, i])
+                i = 2**32 - 1
+                roundtrip([0, 0, 0, 0, 0, i, i, i])
+                i = 2**63 - 1
+                roundtrip([0, 0, 0, 0, 0, 0, i, i])
+                i = 2**64 - 1
+                roundtrip([0, 0, 0, 0, 0, 0, 0, i])
+        finally:
+            meta.drop_all()
+
+    @testing.supported('mysql')
+    def test_boolean(self):
+        """Test BOOL/TINYINT(1) compatability and reflection."""
+
+        meta = MetaData(testbase.db)
+        bool_table = Table('mysql_bool', meta,
+                           Column('b1', BOOLEAN),
+                           Column('b2', mysql.MSBoolean),
+                           Column('b3', mysql.MSTinyInteger(1)),
+                           Column('b4', mysql.MSTinyInteger))
+
+        self.assert_eq(colspec(bool_table.c.b1), 'b1 BOOL')
+        self.assert_eq(colspec(bool_table.c.b2), 'b2 BOOL')
+        self.assert_eq(colspec(bool_table.c.b3), 'b3 TINYINT(1)')
+        self.assert_eq(colspec(bool_table.c.b4), 'b4 TINYINT')
+
+        try:
+            meta.create_all()
+
+            table = bool_table
+            def roundtrip(store, expected=None):
+                expected = expected or store
+                table.insert(store).execute()
+                row = list(table.select().execute())[0]
+                try:
+                    self.assert_(list(row) == expected)
+                    for i, val in enumerate(expected):
+                        if isinstance(val, bool):
+                            self.assert_(val is row[i])
+                except:
+                    print "Storing %s" % store
+                    print "Expected %s" % expected
+                    print "Found %s" % list(row)
+                    raise
+                table.delete().execute()
+
+
+            roundtrip([None, None, None, None])
+            roundtrip([True, True, 1, 1])
+            roundtrip([False, False, 0, 0])
+            roundtrip([True, True, True, True], [True, True, 1, 1])
+            roundtrip([False, False, 0, 0], [False, False, 0, 0])
+
+            meta2 = MetaData(testbase.db)
+            # replace with reflected
+            table = Table('mysql_bool', meta2, autoload=True)
+            self.assert_eq(colspec(table.c.b3), 'b3 BOOL')
+
+            roundtrip([None, None, None, None])
+            roundtrip([True, True, 1, 1], [True, True, True, 1])
+            roundtrip([False, False, 0, 0], [False, False, False, 0])
+            roundtrip([True, True, True, True], [True, True, True, 1])
+            roundtrip([False, False, 0, 0], [False, False, False, 0])
+        finally:
+            meta.drop_all()
+
+    @testing.supported('mysql')
+    @testing.exclude('mysql', '<', (4, 1, 0))
+    def test_timestamp(self):
+        """Exercise funky TIMESTAMP default syntax."""
+    
+        meta = MetaData(testbase.db)
+
+        try:
+            columns = [
+                ([TIMESTAMP],
+                 'TIMESTAMP'),
+                ([mysql.MSTimeStamp],
+                 'TIMESTAMP'),
+                ([mysql.MSTimeStamp,
+                  PassiveDefault(sql.text('CURRENT_TIMESTAMP'))],
+                 "TIMESTAMP DEFAULT CURRENT_TIMESTAMP"),
+                ([mysql.MSTimeStamp,
+                  PassiveDefault(sql.text("'1999-09-09 09:09:09'"))],
+                 "TIMESTAMP DEFAULT '1999-09-09 09:09:09'"),
+                ([mysql.MSTimeStamp,
+                  PassiveDefault(sql.text("'1999-09-09 09:09:09' "
+                                          "ON UPDATE CURRENT_TIMESTAMP"))],
+                 "TIMESTAMP DEFAULT '1999-09-09 09:09:09' "
+                 "ON UPDATE CURRENT_TIMESTAMP"),
+                ([mysql.MSTimeStamp,
+                  PassiveDefault(sql.text("CURRENT_TIMESTAMP "
+                                          "ON UPDATE CURRENT_TIMESTAMP"))],
+                 "TIMESTAMP DEFAULT CURRENT_TIMESTAMP "
+                 "ON UPDATE CURRENT_TIMESTAMP"),
+                ]
+            for idx, (spec, expected) in enumerate(columns):
+                t = Table('mysql_ts%s' % idx, meta,
+                          Column('id', Integer, primary_key=True),
+                          Column('t', *spec))
+                self.assert_eq(colspec(t.c.t), "t %s" % expected)
+                t.create()
+                r = Table('mysql_ts%s' % idx, MetaData(testbase.db),
+                          autoload=True)
+                if len(spec) > 1:
+                    self.assert_(r.c.t is not None)
+        finally:
+            meta.drop_all()
+
+    @testing.supported('mysql')
+    def test_year(self):
+        """Exercise YEAR."""
+
+        meta = MetaData(testbase.db)
+        year_table = Table('mysql_year', meta,
+                           Column('y1', mysql.MSYear),
+                           Column('y2', mysql.MSYear),
+                           Column('y3', mysql.MSYear),
+                           Column('y4', mysql.MSYear),
+                           Column('y5', mysql.MSYear))
+
+        try:
+            year_table.create()
+            reflected = Table('mysql_year', MetaData(testbase.db),
+                              autoload=True)
+
+            for table in year_table, reflected:
+                table.insert(['1950', '50', None, 50, 1950]).execute()
+                row = list(table.select().execute())[0]
+                self.assert_eq(list(row), [1950, 2050, None, 2050, 1950])
+                table.delete().execute()
+        finally:
+            meta.drop_all()
+        
+
+    @testing.supported('mysql')
+    def test_set(self):
+        """Exercise the SET type."""
+
+        meta = MetaData(testbase.db)
+        set_table = Table('mysql_set', meta,
+                          Column('s1', mysql.MSSet('"dq"', "'sq'")),
+                          Column('s2', mysql.MSSet("'a'")),
+                          Column('s3', mysql.MSSet("'5'", "'7'", "'9'")))
+
+        self.assert_eq(colspec(set_table.c.s1), """s1 SET("dq",'sq')""")
+        self.assert_eq(colspec(set_table.c.s2), "s2 SET('a')")
+        self.assert_eq(colspec(set_table.c.s3), "s3 SET('5','7','9')")
+
+        try:
+            set_table.create()
+            reflected = Table('mysql_set', MetaData(testbase.db),
+                              autoload=True)
+
+            for table in set_table, reflected:
+                def roundtrip(store, expected=None):
+                    expected = expected or store
+                    table.insert(store).execute()
+                    row = list(table.select().execute())[0]
+                    try:
+                        self.assert_(list(row) == expected)
+                    except:
+                        print "Storing %s" % store
+                        print "Expected %s" % expected
+                        print "Found %s" % list(row)
+                        raise
+                    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'])])
+        finally:
+            meta.drop_all()
+
+    @testing.supported('mysql')
     def test_enum(self):
-        "Exercise the ENUM type"
+        """Exercise the ENUM type."""
         
         db = testbase.db
         enum_table = Table('mysql_enum', MetaData(testbase.db),
             Column('e1', mysql.MSEnum('"a"', "'b'")),
-            Column('e2', mysql.MSEnum('"a"', "'b'"), nullable=False),
+            Column('e2', mysql.MSEnum('"a"', "'b'"),
+                   nullable=False),
             Column('e3', mysql.MSEnum('"a"', "'b'", strict=True)),
-            Column('e4', mysql.MSEnum('"a"', "'b'", strict=True), nullable=False))
-        spec = lambda c: db.dialect.schemagenerator(db, None, None).get_column_specification(c)
-
-        self.assertEqual(spec(enum_table.c.e1), """e1 ENUM("a",'b')""")
-        self.assertEqual(spec(enum_table.c.e2), """e2 ENUM("a",'b') NOT NULL""")
-        self.assertEqual(spec(enum_table.c.e3), """e3 ENUM("a",'b')""")
-        self.assertEqual(spec(enum_table.c.e4), """e4 ENUM("a",'b') NOT NULL""")
+            Column('e4', mysql.MSEnum('"a"', "'b'", strict=True),
+                   nullable=False))
+
+        self.assert_eq(colspec(enum_table.c.e1),
+                       """e1 ENUM("a",'b')""")
+        self.assert_eq(colspec(enum_table.c.e2),
+                       """e2 ENUM("a",'b') NOT NULL""")
+        self.assert_eq(colspec(enum_table.c.e3),
+                       """e3 ENUM("a",'b')""")
+        self.assert_eq(colspec(enum_table.c.e4),
+                       """e4 ENUM("a",'b') NOT NULL""")
         enum_table.drop(checkfirst=True)
         enum_table.create()
 
@@ -313,7 +550,7 @@ class TypesTest(AssertMixin):
                 e.append(tuple([convert(c) for c in row]))
             expected = e
 
-        self.assertEqual(res, expected)
+        self.assert_eq(res, expected)
         enum_table.drop()
 
     @testing.supported('mysql')
@@ -363,6 +600,16 @@ class TypesTest(AssertMixin):
 
         m.drop_all()
 
+    def assert_eq(self, got, wanted):
+        if got != wanted:
+            print "Expected %s" % wanted
+            print "Found %s" % got
+        self.assertEqual(got, wanted)
+    
+
+def colspec(c):
+    return testbase.db.dialect.schemagenerator(
+        testbase.db, None, None).get_column_specification(c)
 
 if __name__ == "__main__":
     testbase.main()