fix(autogen): compare identity and computed defaults when compare_server_default is false

alembic/autogenerate/compare/server_defaults.py

@@ -194,9 +194,6 @@ def _user_compare_server_default(
 
     migration_context = autogen_context.migration_context
 
-    if migration_context._user_compare_server_default is False:
-        return PriorityDispatchResult.STOP
-
     if not callable(migration_context._user_compare_server_default):
         return PriorityDispatchResult.CONTINUE
 
@@ -266,6 +263,9 @@ def _dialect_impl_compare_server_default(
 
     migration_context = autogen_context.migration_context
 
+    if migration_context._user_compare_server_default is False:
+        return PriorityDispatchResult.STOP
+
     rendered_metadata_default = _render_server_default_for_compare(
         metadata_default, autogen_context
     )

alembic/testing/suite/test_autogen_identity.py

@@ -86,6 +86,25 @@ def test_no_change_identity_column(self):
 
         eq_(diffs, [])
 
+    def test_identity_compare_server_default_false(self):
+        m1 = MetaData()
+        m2 = MetaData()
+
+        Table("user", m1, Column("id", Integer, sa.Identity(start=2)))
+        Table("user", m2, Column("id", Integer, sa.Identity(start=5)))
+
+        diffs = self._fixture(m1, m2, opts={"compare_server_default": False})
+        eq_(len(diffs[0]), 1)
+        diffs = diffs[0][0]
+        eq_(diffs[0], "modify_default")
+        eq_(diffs[2], "user")
+        eq_(diffs[3], "id")
+        old = diffs[5]
+        new = diffs[6]
+
+        is_true(isinstance(old, sa.Identity))
+        is_true(isinstance(new, sa.Identity))
+
     def test_dialect_kwargs_changes(self):
         m1 = MetaData()
         m2 = MetaData()

docs/build/unreleased/1877.rst

@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, autogenerate
+    :tickets: 1877
+
+    Fixed regression where ``compare_metadata`` skipped comparing identity
+    and computed default changes when ``compare_server_default`` was not
+    configured (the default setting of ``False``). The comparator dispatch
+    prematurely halted comparison rather than deferring the plain default
+    comparison skip to the dialect implementation.

tests/test_autogen_diffs.py

@@ -943,6 +943,88 @@ def my_compare_server_default(
         else:
             assert not diff
 
+    def test_computed_default_when_compare_server_default_false(
+        self, connection, metadata
+    ):
+        import warnings
+        from sqlalchemy import Computed
+
+        t1 = Table(
+            "t1",
+            metadata,
+            Column("id", Integer, primary_key=True),
+            Column("x", Integer),
+            Column("c", Integer, Computed("x * 2")),
+        )
+        t1.create(connection)
+
+        new_metadata = MetaData()
+        Table(
+            "t1",
+            new_metadata,
+            Column("id", Integer, primary_key=True),
+            Column("x", Integer),
+            Column("c", Integer, Computed("x * 3")),
+        )
+
+        mc = MigrationContext.configure(connection)
+
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+            api.compare_metadata(mc, new_metadata)
+
+        eq_(len(w), 1)
+        assert "Computed default on t1.c cannot be modified" in str(
+            w[0].message
+        )
+
+    def test_identity_default_when_compare_server_default_false(
+        self, connection, metadata
+    ):
+        from sqlalchemy import event
+        from sqlalchemy import Identity
+
+        t1 = Table(
+            "t1",
+            metadata,
+            Column("id", Integer, primary_key=True),
+        )
+        t1.create(connection)
+
+        def set_identity(inspector, table, column_info):
+            if column_info["name"] == "id":
+                column_info["identity"] = {"start": 1, "increment": 1}
+
+        event.listen(Table, "column_reflect", set_identity)
+        try:
+            new_metadata = MetaData()
+            Table(
+                "t1",
+                new_metadata,
+                Column(
+                    "id",
+                    Integer,
+                    Identity(start=1, increment=5),
+                    primary_key=True,
+                ),
+            )
+
+            mc = MigrationContext.configure(connection)
+
+            diff = api.compare_metadata(mc, new_metadata)
+            eq_(len(diff), 1)
+            eq_(diff[0][0][0], "modify_default")
+            eq_(diff[0][0][2], "t1")
+            eq_(diff[0][0][3], "id")
+            old = diff[0][0][5]
+            new = diff[0][0][6]
+            assert isinstance(old, Identity)
+            assert isinstance(new, Identity)
+            eq_(old.increment, 1)
+            eq_(new.increment, 5)
+        finally:
+            event.remove(Table, "column_reflect", set_identity)
+
 
 class CompareMetadataToInspectorTest(TestBase):
     __backend__ = True

tests/test_oracle.py

@@ -445,3 +445,35 @@ def test_identity_on_null(self):
 
         is_true(isinstance(old, sa.Identity))
         is_true(isinstance(new, sa.Identity))
+
+    def test_identity_compare_server_default_false(self):
+        m1 = MetaData()
+        m2 = MetaData()
+
+        Table(
+            "user",
+            m1,
+            Column("id", Integer, Identity(start=2)),
+            Column("other", sa.Text),
+        )
+
+        Table(
+            "user",
+            m2,
+            Column("id", Integer, Identity(start=5)),
+            Column("other", sa.Text),
+        )
+
+        diffs = self._fixture(
+            m1, m2, opts={"compare_server_default": False}
+        )
+        eq_(len(diffs[0]), 1)
+        diffs = diffs[0][0]
+        eq_(diffs[0], "modify_default")
+        eq_(diffs[2], "user")
+        eq_(diffs[3], "id")
+        old = diffs[5]
+        new = diffs[6]
+
+        is_true(isinstance(old, sa.Identity))
+        is_true(isinstance(new, sa.Identity))