narrow rangenotancestor to exclude target heads already present

Fixed regression in new revisioning traversal where "alembic downgrade
base" would fail if the database itself were clean and unversioned;
additionally repairs the case where downgrade would fail if attempting
to downgrade to the current head that is already present.

Change-Id: I581cab5a17f69d82e41eab147ceb27a38fe4ce1d
Fixes: sqlalchemy#838

alembic/script/revision.py

@@ -1226,6 +1226,7 @@ def _collect_downgrade_revisions(
         active_revisions = set(
             self._get_ancestor_nodes(heads, include_dependencies=True)
         )
+
         # Emit revisions to drop in reverse topological sorted order.
         downgrade_revisions.intersection_update(active_revisions)
 
@@ -1235,8 +1236,13 @@ def _collect_downgrade_revisions(
                 active_revisions.difference(self._get_ancestor_nodes(roots))
             )
 
-        if not downgrade_revisions:
+        if (
+            target_revision is not None
+            and not downgrade_revisions
+            and target_revision not in heads
+        ):
             # Empty intersection: target revs are not present.
+
             raise RangeNotAncestorError("Nothing to drop", upper)
 
         return downgrade_revisions, heads

docs/build/unreleased/838.rst

@@ -0,0 +1,8 @@
+.. change::
+    :tags: bug, versioning, regression
+    :tickets: 838
+
+    Fixed regression in new revisioning traversal where "alembic downgrade
+    base" would fail if the database itself were clean and unversioned;
+    additionally repairs the case where downgrade would fail if attempting
+    to downgrade to the current head that is already present.

tests/test_version_traversal.py

@@ -50,6 +50,14 @@ def setup_class(cls):
     def teardown_class(cls):
         clear_staging_env()
 
+    def test_downgrade_base_no_version(self):
+        self._assert_downgrade("base", [], [], set())
+
+    def test_downgrade_to_existing(self):
+        self._assert_downgrade(
+            self.a.revision, [self.a.revision], [], {self.a.revision}
+        )
+
     def test_upgrade_path(self):
         self._assert_upgrade(
             self.e.revision,