run drop default constraint ahead of type

Fixed issue in SQL Server dialect where the DROP that's automatically
emitted for existing default constraints during an ALTER COLUMN needs to
take place before not just the modification of the column's default, but
also before the column's type is changed.

also vendor sqlalchemy's metadata fixture which otherwise does not
integrate with alembic's version of the connection fixture

Fixes: #1744
Change-Id: Ia756da024297354449f53e9f3254820dec542bef

diff --git a/alembic/ddl/mssql.py b/alembic/ddl/mssql.py
index 50d441f8be1f5590c3ba014e97300ac767c4603b..1d1e3e98b1e7f85deefe9bda54506090cea99f80 100644 (file)
--- a/alembic/ddl/mssql.py
+++ b/alembic/ddl/mssql.py
@@ -140,6 +140,27 @@ class MSSQLImpl(DefaultImpl):
             kw["server_default"] = server_default
             kw["existing_server_default"] = existing_server_default
 
             kw["server_default"] = server_default
             kw["existing_server_default"] = existing_server_default
 

+        # drop existing default constraints before changing type
+        # or default, see issue #1744
+        if (
+            server_default is not False
+            and used_default is False
+            and (
+                existing_server_default is not False or server_default is None
+            )
+        ):
+            self._exec(
+                _ExecDropConstraint(
+                    table_name,
+                    column_name,
+                    "sys.default_constraints",
+                    schema,
+                )
+            )
+
+        # TODO: see why these two alter_columns can't be called
+        # at once.   joining them works but some of the mssql tests
+        # seem to expect something different

         super().alter_column(
             table_name,
             column_name,
         super().alter_column(
             table_name,
             column_name,


@@ -152,15 +173,6 @@ class MSSQLImpl(DefaultImpl):
         )
 
         if server_default is not False and used_default is False:
         )
 
         if server_default is not False and used_default is False:

-            if existing_server_default is not False or server_default is None:
-                self._exec(
-                    _ExecDropConstraint(
-                        table_name,
-                        column_name,
-                        "sys.default_constraints",
-                        schema,
-                    )
-                )

             if server_default is not None:
                 super().alter_column(
                     table_name,
             if server_default is not None:
                 super().alter_column(
                     table_name,

diff --git a/alembic/testing/fixtures.py b/alembic/testing/fixtures.py
index 6d801fcbc1c0252442d1f83e90b858853bbbcb49..62084dccb92e9940376c1e1954dc2846f1137abb 100644 (file)
--- a/alembic/testing/fixtures.py
+++ b/alembic/testing/fixtures.py
@@ -26,6 +26,7 @@ from sqlalchemy.testing.assertions import eq_
 from sqlalchemy.testing.fixtures import FutureEngineMixin
 from sqlalchemy.testing.fixtures import TablesTest as SQLAlchemyTablesTest
 from sqlalchemy.testing.fixtures import TestBase as SQLAlchemyTestBase
 from sqlalchemy.testing.fixtures import FutureEngineMixin
 from sqlalchemy.testing.fixtures import TablesTest as SQLAlchemyTablesTest
 from sqlalchemy.testing.fixtures import TestBase as SQLAlchemyTestBase

+from sqlalchemy.testing.util import drop_all_tables_from_metadata

 
 import alembic
 from .assertions import _get_dialect
 
 import alembic
 from .assertions import _get_dialect


@@ -87,9 +88,41 @@ class TestBase(SQLAlchemyTestBase):
 
     @testing.fixture
     def connection(self):
 
     @testing.fixture
     def connection(self):

+        global _connection_fixture_connection
+

         with config.db.connect() as conn:
         with config.db.connect() as conn:

+            _connection_fixture_connection = conn

             yield conn
 
             yield conn
 

+            _connection_fixture_connection = None
+
+    @config.fixture()
+    def metadata(self, request):
+        """Provide bound MetaData for a single test, dropping afterwards."""
+
+        from sqlalchemy.sql import schema
+
+        metadata = schema.MetaData()
+        request.instance.metadata = metadata
+        yield metadata
+        del request.instance.metadata
+
+        if (
+            _connection_fixture_connection
+            and _connection_fixture_connection.in_transaction()
+        ):
+            trans = _connection_fixture_connection.get_transaction()
+            trans.rollback()
+            with _connection_fixture_connection.begin():
+                drop_all_tables_from_metadata(
+                    metadata, _connection_fixture_connection
+                )
+        else:
+            drop_all_tables_from_metadata(metadata, config.db)
+
+
+_connection_fixture_connection = None
+

 
 class TablesTest(TestBase, SQLAlchemyTablesTest):
     pass
 
 class TablesTest(TestBase, SQLAlchemyTablesTest):
     pass

diff --git a/docs/build/unreleased/1744.rst b/docs/build/unreleased/1744.rst
new file mode 100644 (file)
index 0000000..4abdbe0
--- /dev/null
+++ b/docs/build/unreleased/1744.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, mssql
+    :tickets: 1744
+
+    Fixed issue in SQL Server dialect where the DROP that's automatically
+    emitted for existing default constraints during an ALTER COLUMN needs to
+    take place before not just the modification of the column's default, but
+    also before the column's type is changed.

diff --git a/tests/test_mssql.py b/tests/test_mssql.py
index 70fe36536ad735cf5acc49a6c8104aec239d0a7b..1390f2dfa294ab01078e7764ab8b70dec9f8d9ab 100644 (file)
--- a/tests/test_mssql.py
+++ b/tests/test_mssql.py
@@ -8,8 +8,11 @@ from typing import Dict
 from sqlalchemy import CheckConstraint
 from sqlalchemy import Column
 from sqlalchemy import Computed
 from sqlalchemy import CheckConstraint
 from sqlalchemy import Column
 from sqlalchemy import Computed

+from sqlalchemy import DATE
+from sqlalchemy import DATETIME

 from sqlalchemy import exc
 from sqlalchemy import ForeignKey
 from sqlalchemy import exc
 from sqlalchemy import ForeignKey

+from sqlalchemy import func

 from sqlalchemy import Identity
 from sqlalchemy import inspect
 from sqlalchemy import Integer
 from sqlalchemy import Identity
 from sqlalchemy import inspect
 from sqlalchemy import Integer


@@ -20,6 +23,7 @@ from sqlalchemy import text
 
 from alembic import command
 from alembic import op
 
 from alembic import command
 from alembic import op

+from alembic import testing

 from alembic import util
 from alembic.testing import assert_raises_message
 from alembic.testing import combinations
 from alembic import util
 from alembic.testing import assert_raises_message
 from alembic.testing import combinations


@@ -576,3 +580,28 @@ class RoundTripTest(TestBase):
 
     # don't know if a default constraint can be explicitly named, but
     # the path is the same as the check constraint, so it should be good
 
     # don't know if a default constraint can be explicitly named, but
     # the path is the same as the check constraint, so it should be good

+
+    @testing.variation("op", ["drop", "alter"])
+    def test_issue_1744(self, ops_context, connection, metadata, op):
+        access = Table(
+            "access",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            Column("created_at", DATE, server_default=func.getdate()),
+        )
+        access.create(connection)
+
+        if op.alter:
+            ops_context.alter_column(
+                "access",
+                "created_at",
+                existing_type=DATETIME(),
+                type_=DATETIME(timezone=True),
+                server_default=func.getdate(),
+                existing_nullable=False,
+                existing_server_default=text("(getdate())"),
+            )
+        elif op.drop:
+            ops_context.drop_column(
+                "access", "created_at", mssql_drop_default=True
+            )