cast() types in batch only if type_affinity is different

Batch mode will not use CAST() to copy data if type_ is given, however
the basic type affinity matches that of the existing type.  This to
avoid SQLite's CAST of TIMESTAMP which results in truncation of the
data, in those cases where the user needs to add redundant type_ for
other reasons.

Change-Id: I20f7b399cd3cd7740d67ff7d624aa1da874ebc71
Fixes: #391

1 files changed, 31 insertions, 6 deletions

diff --git a/tests/test_batch.py b/tests/test_batch.py
index 7dad62f..3124baa 100644
--- a/tests/test_batch.py
+++ b/tests/test_batch.py
@@ -13,9 +13,9 @@ from alembic.runtime.migration import MigrationContext
 
 from sqlalchemy import Integer, Table, Column, String, MetaData, ForeignKey, \
     UniqueConstraint, ForeignKeyConstraint, Index, Boolean, CheckConstraint, \
-    Enum
+    Enum, DateTime
 from sqlalchemy.engine.reflection import Inspector
-from sqlalchemy.sql import column, text
+from sqlalchemy.sql import column, text, select
 from sqlalchemy.schema import CreateTable, CreateIndex
 from sqlalchemy import exc
 
@@ -215,6 +215,7 @@ class BatchApplyTest(TestBase):
             impl.new_table.c[name].name
             for name in colnames
             if name in impl.table.c])
+
         args['tname_colnames'] = ", ".join(
             "CAST(%(schema)stname.%(name)s AS %(type)s) AS anon_1" % {
                 'schema': args['schema'],
@@ -243,9 +244,9 @@ class BatchApplyTest(TestBase):
 
     def test_change_type(self):
         impl = self._simple_fixture()
-        impl.alter_column('tname', 'x', type_=Integer)
+        impl.alter_column('tname', 'x', type_=String)
         new_table = self._assert_impl(impl)
-        assert new_table.c.x.type._type_affinity is Integer
+        assert new_table.c.x.type._type_affinity is String
 
     def test_rename_col(self):
         impl = self._simple_fixture()
@@ -751,7 +752,6 @@ class CopyFromTest(TestBase):
         with self.op.batch_alter_table(
                 "foo", copy_from=self.table) as batch_op:
             batch_op.alter_column('data', type_=Integer)
-
         context.assert_(
             'CREATE TABLE _alembic_batch_temp (id INTEGER NOT NULL, '
             'data INTEGER, x INTEGER, PRIMARY KEY (id))',
@@ -889,7 +889,7 @@ class CopyFromTest(TestBase):
             'CREATE TABLE _alembic_batch_temp (id INTEGER NOT NULL, '
             'data VARCHAR, x INTEGER, PRIMARY KEY (id))',
             'INSERT INTO _alembic_batch_temp (id, data, x) SELECT foo.id, '
-            'CAST(foo.data AS VARCHAR) AS anon_1, foo.x FROM foo',
+            'foo.data, foo.x FROM foo',
             'DROP TABLE foo',
             'ALTER TABLE _alembic_batch_temp RENAME TO foo'
         )
@@ -970,6 +970,14 @@ class BatchRoundTripTest(TestBase):
         )
         t.create(self.conn)
 
+    def _timestamp_fixture(self):
+        t = Table(
+            'hasts', self.metadata,
+            Column('x', DateTime()),
+        )
+        t.create(self.conn)
+        return t
+
     def _int_to_boolean_fixture(self):
         t = Table(
             'hasbool', self.metadata,
@@ -993,6 +1001,23 @@ class BatchRoundTripTest(TestBase):
             [Integer]
         )
 
+    def test_no_net_change_timestamp(self):
+        t = self._timestamp_fixture()
+
+        import datetime
+        self.conn.execute(
+            t.insert(),
+            {"x": datetime.datetime(2012, 5, 18, 15, 32, 5)}
+        )
+
+        with self.op.batch_alter_table("hasts") as batch_op:
+            batch_op.alter_column("x", type_=DateTime())
+
+        eq_(
+            self.conn.execute(select([t.c.x])).fetchall(),
+            [(datetime.datetime(2012, 5, 18, 15, 32, 5),)]
+        )
+
     def test_drop_col_schematype(self):
         self._boolean_fixture()
         with self.op.batch_alter_table(