Allow deleting collection versions when another version of the collec…

…tion satisfies requirements.

fixes: #933
(cherry picked from commit c1bea33)

CHANGES/933.bugfix

@@ -0,0 +1 @@
+Allow deleting collection versions when another version of the collection satisfies requirements.

pulp_ansible/app/galaxy/v3/views.py

@@ -419,14 +419,27 @@ def get_collection_dependents(parent):
     )
 
 
-def get_dependents(parent):
+def get_unique_dependents(parent):
     """Given a parent collection version, return a list of collection versions that depend on it."""
     key = f"{parent.namespace}.{parent.name}"
+
+    this_version = semantic_version.Version(parent.version)
+
+    # Other versions contain a set of all versions of this collection aside from the version
+    # that is being deleted.
+    other_versions = []
+    for v in parent.collection.versions.exclude(version=parent.version):
+        other_versions.append(semantic_version.Version(v.version))
+
     dependents = []
     for child in CollectionVersion.objects.filter(dependencies__has_key=key):
         spec = semantic_version.SimpleSpec(child.dependencies[key])
-        if spec.match(semantic_version.Version(parent.version)):
+
+        # If this collection matches the parent collections version and there are no other
+        # collection versions that can satisfy the requirement, add it to the list of dependants.
+        if spec.match(this_version) and not spec.select(other_versions):
             dependents.append(child)
+
     return dependents
 
 
@@ -629,7 +642,7 @@ def destroy(self, request: Request, *args, **kwargs) -> Response:
         collection_version = self.get_object()
 
         # dependency check
-        dependents = get_dependents(collection_version)
+        dependents = get_unique_dependents(collection_version)
         if dependents:
             return Response(
                 {

pulp_ansible/tests/functional/api/collection/v3/test_deletion.py

@@ -48,16 +48,13 @@ def test_collection_deletion(self):
 
         assert versions.meta.count == 0
 
-        try:
+        with self.assertRaises(ApiException) as e:
             self.collections_v3api.read(
                 path=self.distribution.base_path,
                 name=self.collection_name,
                 namespace=self.collection_namespace,
             )
 
-            # Fail if 404 isn't raised
-            assert False
-        except ApiException as e:
             assert e.status == 404
 
     def test_collection_version_deletion(self):
@@ -84,17 +81,13 @@ def test_collection_version_deletion(self):
 
         monitor_task(resp.task)
 
-        try:
+        with self.assertRaises(ApiException) as e:
             self.collections_versions_v3api.read(
                 path=self.distribution.base_path,
                 name=self.collection_name,
                 namespace=self.collection_namespace,
                 version=to_delete,
             )
-
-            # Fail if 404 isn't raised
-            assert False
-        except ApiException as e:
             assert e.status == 404
 
         # Verify that the collection still exists
@@ -132,16 +125,12 @@ def test_collection_version_deletion(self):
 
         # With all the versions deleted, verify that the collection has also
         # been deleted
-        try:
+        with self.assertRaises(ApiException) as e:
             self.collections_v3api.read(
                 path=self.distribution.base_path,
                 name=self.collection_name,
                 namespace=self.collection_namespace,
             )
-
-            # Fail if 404 isn't raised
-            assert False
-        except ApiException as e:
             assert e.status == 404
 
     def test_invalid_deletion(self):
@@ -155,33 +144,24 @@ def test_invalid_deletion(self):
         err_msg = f"{dependent_collection['namespace']}.{dependent_collection['name']} 1.0.0"
 
         # Verify entire collection can't be deleted
-        try:
+        with self.assertRaises(ApiException) as e:
             self.collections_v3api.delete(
                 path=self.distribution.base_path,
                 name=self.collection_name,
                 namespace=self.collection_namespace,
             )
 
-            # Fail if 400 isn't raised
-            assert False
-        except ApiException as e:
-            assert e.status == 400
-
             # check error message includes collection that's blocking delete
             assert err_msg in e.body
 
         # Verify specific version that's used can't be deleted
-        try:
+        with self.assertRaises(ApiException) as e:
             self.collections_versions_v3api.delete(
                 path=self.distribution.base_path,
                 name=self.collection_name,
                 namespace=self.collection_namespace,
                 version=dependent_version,
             )
-
-            # Fail if 400 isn't raised
-            assert False
-        except ApiException as e:
             assert e.status == 400
 
             # check error message includes collection that's blocking delete
@@ -246,3 +226,36 @@ def test_delete_signed_content(self):
         latest_version = self.repo_version_api.read(repo.latest_version_href)
 
         assert len(latest_version.content_summary.present) == 0
+
+    def test_version_deletion_with_range_of_versions(self):
+        """Verify collections can be deleted when another version satisfies requirements."""
+        # Create a collection that depends on any version of an existing collection
+        gen_collection_in_distribution(
+            self.distribution.base_path,
+            dependencies={f"{self.collection_namespace}.{self.collection_name}": "*"},
+        )
+
+        to_delete = self.collection_versions.pop()
+
+        # Verify the collection version can be deleted as long as there is one version
+        # left that satisfies the requirements.
+        resp = self.collections_versions_v3api.delete(
+            path=self.distribution.base_path,
+            name=self.collection_name,
+            namespace=self.collection_namespace,
+            version=to_delete,
+        )
+
+        resp = monitor_task(resp.task)
+
+        # Verify that the last version of the collection can't be deleted
+
+        with self.assertRaises(ApiException) as e:
+            self.collections_versions_v3api.delete(
+                path=self.distribution.base_path,
+                name=self.collection_name,
+                namespace=self.collection_namespace,
+                version=self.collection_versions[0],
+            )
+
+            assert e.status == 400