Fix downgrade when normalized down revisions have overlap via depends_on

alembic/runtime/migration.py

@@ -1168,7 +1168,18 @@ def _unmerge_to_revisions(self, heads: Set[str]) -> Tuple[str, ...]:
             }
             return tuple(set(self.to_revisions).difference(ancestors))
         else:
-            return self.to_revisions
+            # for each revision we plan to return, compute its ancestors
+            # (excluding self), and remove those from the final output since
+            # they are already accounted for.
+            ancestors = {
+                r.revision
+                for to_revision in self.to_revisions
+                for r in self.revision_map._get_ancestor_nodes(
+                    self.revision_map.get_revisions(to_revision), check=False
+                )
+                if r.revision != to_revision
+            }
+            return tuple(set(self.to_revisions).difference(ancestors))
 
     def unmerge_branch_idents(
         self, heads: Set[str]

tests/test_version_traversal.py

@@ -1176,6 +1176,67 @@ def test_dependencies_are_normalized(self):
         )
 
 
+class DependsOnBranchTestFive(MigrationTest):
+    @classmethod
+    def setup_class(cls):
+        """
+        issue #1373
+
+        Structure::
+
+            <base> -> a1 ------+
+                      ^        |
+                      |        +-> bmerge
+                      |        |
+                      +-- b1 --+
+        """
+        cls.env = env = staging_env()
+        cls.a1 = env.generate_revision("a1", "->a1")
+        cls.b1 = env.generate_revision(
+            "b1", "->b1", head="base", depends_on="a1"
+        )
+        cls.bmerge = env.generate_revision(
+            "bmerge", "bmerge", head=[cls.a1.revision, cls.b1.revision]
+        )
+
+    @classmethod
+    def teardown_class(cls):
+        clear_staging_env()
+
+    def test_downgrade_to_depends_on(self):
+        # Upgrade from a1 to b1 just has heads={"b1"}.
+        self._assert_upgrade(
+            self.b1.revision,
+            self.a1.revision,
+            [self.up_(self.b1)],
+            {self.b1.revision},
+        )
+
+        # Upgrade from b1 to bmerge just has {"bmerge"}.
+        self._assert_upgrade(
+            self.bmerge.revision,
+            self.b1.revision,
+            [self.up_(self.bmerge)],
+            {self.bmerge.revision},
+        )
+
+        # Downgrading from bmerge to a1 should return back to heads={"b1"}.
+        self._assert_downgrade(
+            self.a1.revision,
+            self.bmerge.revision,
+            [self.down_(self.bmerge)],
+            {self.b1.revision},
+        )
+
+        # Downgrading from bmerge to b1 also returns back to heads={"b1"}.
+        self._assert_downgrade(
+            self.b1.revision,
+            self.bmerge.revision,
+            [self.down_(self.bmerge)],
+            {self.b1.revision},
+        )
+
+
 class DependsOnBranchLabelTest(MigrationTest):
     @classmethod
     def setup_class(cls):