Ensure beta-galaxy users can delete and deprecate their collections

CHANGES/2632.bugfix

@@ -0,0 +1 @@
+Ensure beta-galaxy users can delete and deprecate their collections

galaxy_ng/app/access_control/access_policy.py

@@ -259,6 +259,27 @@ def v3_can_view_repo_content(self, request, view, action):
 
         return True
 
+    def v3_can_destroy_collections(self, request, view, action):
+        SOCIAL_AUTH_GITHUB_KEY = settings.get("SOCIAL_AUTH_GITHUB_KEY", default=None)
+        SOCIAL_AUTH_GITHUB_SECRET = settings.get("SOCIAL_AUTH_GITHUB_SECRET", default=None)
+        is_social_auth = all([SOCIAL_AUTH_GITHUB_KEY, SOCIAL_AUTH_GITHUB_SECRET])
+
+        user = request.user
+        perm = "ansible.delete_collection"
+        social_perm = "galaxy.change_namespace"
+        collection = view.get_object()
+        namespace = models.Namespace.objects.get(name=collection.namespace)
+
+        if not is_social_auth and user.has_perm(perm) and self.v3_can_view_repo_content(
+            request,
+            view,
+            action,
+        ):
+            return True
+        elif is_social_auth and user.has_perm(social_perm, namespace):
+            return True
+        return False
+
     def has_ansible_repo_perms(self, request, view, action, permission):
         """
         Check if the user has model or object-level permissions

galaxy_ng/app/access_control/statements/standalone.py

@@ -35,7 +35,7 @@
         "action": "destroy",
         "principal": "authenticated",
         "effect": "allow",
-        "condition": ["has_model_perms:ansible.delete_collection", "v3_can_view_repo_content"],
+        "condition": ["v3_can_destroy_collections"],
     },
     {
         "action": ["download"],

galaxy_ng/tests/integration/api/test_community.py

@@ -12,6 +12,7 @@
     get_client,
     SocialGithubClient,
     create_user,
+    wait_for_task,
 )
 from ..utils.legacy import (
     clean_all_roles,
@@ -219,6 +220,121 @@ def test_social_auth_creates_v3_namespace(ansible_config):
         )
 
 
+@pytest.mark.deployment_community
+def test_social_auth_delete_collection(ansible_config):
+
+    cleanup_social_user('gh01', ansible_config)
+
+    cfg = ansible_config('github_user_1')
+    with SocialGithubClient(config=cfg) as client:
+
+        resp = client.get('v3/namespaces/?name=gh01')
+        result = resp.json()
+        assert result['meta']['count'] == 1
+        assert result['data'][0]['name'] == 'gh01'
+
+        # make a collection
+        namespace = 'gh01'
+        name = 'mystuff'
+        collection = build_collection(
+            use_orionutils=False,
+            namespace=namespace,
+            name=name,
+            version='1.0.0'
+        )
+
+        # verify the user can publish to the namespace ...
+        ansible_galaxy(
+            f"collection publish {collection.filename}",
+            ansible_config=ansible_config("github_user_1"),
+            force_token=True,
+            token=client.get_hub_token(),
+        )
+
+        exists_resp = client.get(
+            f"v3/plugin/ansible/content/published/collections/index/{namespace}/{name}/"
+        )
+        assert exists_resp.status_code == 200
+
+        del_resp = client.delete(
+            f"v3/plugin/ansible/content/published/collections/index/{namespace}/{name}/"
+        )
+        assert del_resp.status_code == 202
+        del_results = del_resp.json()
+        assert isinstance(del_results, dict)
+        assert del_results.get('task') is not None
+
+    admin_config = ansible_config("admin")
+    admin_client = get_client(
+        config=admin_config,
+        request_token=False,
+        require_auth=True
+    )
+    wait_for_task(resp=del_results, api_client=admin_client)
+
+
+@pytest.mark.deployment_community
+def test_social_auth_deprecate_collection(ansible_config):
+
+    cleanup_social_user('gh01', ansible_config)
+
+    cfg = ansible_config('github_user_1')
+    with SocialGithubClient(config=cfg) as client:
+
+        resp = client.get('v3/namespaces/?name=gh01')
+        result = resp.json()
+        assert result['meta']['count'] == 1
+        assert result['data'][0]['name'] == 'gh01'
+
+        # make a collection
+        namespace = 'gh01'
+        name = 'mystuff'
+        collection = build_collection(
+            use_orionutils=False,
+            namespace=namespace,
+            name=name,
+            version='1.0.0'
+        )
+
+        # verify the user can publish to the namespace ...
+        ansible_galaxy(
+            f"collection publish {collection.filename}",
+            ansible_config=ansible_config("github_user_1"),
+            force_token=True,
+            token=client.get_hub_token(),
+        )
+
+        exists_resp = client.get(
+            f"v3/plugin/ansible/content/published/collections/index/{namespace}/{name}/"
+        )
+        assert exists_resp.status_code == 200
+
+        dep_resp = client.patch(
+            f"v3/plugin/ansible/content/published/collections/index/{namespace}/{name}/",
+            data={"deprecated": True},
+        )
+        assert dep_resp.status_code == 202
+        dep_results = dep_resp.json()
+        assert isinstance(dep_results, dict)
+        assert dep_results.get('task') is not None
+
+    admin_config = ansible_config("admin")
+    admin_client = get_client(
+        config=admin_config,
+        request_token=False,
+        require_auth=True
+    )
+    wait_for_task(resp=dep_results, api_client=admin_client)
+
+    with SocialGithubClient(config=cfg) as client:
+        verify_resp = client.get(
+            f"v3/plugin/ansible/content/published/collections/index/{namespace}/{name}/"
+        )
+        assert verify_resp.status_code == 200
+        verify_results = verify_resp.json()
+        assert verify_results["deprecated"] is True
+
+
 @pytest.mark.deployment_community
 def test_social_auth_creates_legacynamespace(ansible_config):
 

galaxy_ng/tests/integration/utils/client_social_github.py

@@ -201,6 +201,30 @@ def put(
         resp = self._rs.put(this_url, headers=pheaders, json=data)
         return resp
 
+    def patch(
+        self,
+        relative_url: str = None,
+        absolute_url: str = None,
+        data=None
+    ) -> requests.models.Response:
+
+        pheaders = {
+            'Accept': 'application/json',
+            'X-CSRFToken': self.csrftoken,
+            'Cookie': f'csrftoken={self.csrftoken}; sessionid={self.sessionid}'
+        }
+
+        this_url = None
+        if absolute_url:
+            uri = urlparse(self.baseurl)
+            this_url = f"{uri.scheme}://{uri.netloc}{absolute_url}"
+        else:
+            this_url = self.baseurl + relative_url
+
+        # call patch
+        resp = self._rs.patch(this_url, headers=pheaders, json=data)
+        return resp
+
     def post(
         self,
         relative_url: str = None,