Consolidate duplicated is_admin_on_provider fn

Author: Swatinem

api/internal/tests/test_permissions.py

@@ -8,7 +8,7 @@
     GetAdminErrorProviderAdapter,
     GetAdminProviderAdapter,
 )
-from api.shared.permissions import RepositoryPermissionsService, UserIsAdminPermissions
+from api.shared.permissions import RepositoryPermissionsService, UserIsAdminPermissions, is_admin_on_provider
 
 
 class MockedPermissionsAdapter:
@@ -178,7 +178,7 @@ def test_is_admin_on_provider_invokes_torngit_adapter_when_user_not_in_admin_arr
         user = OwnerFactory()
 
         mocked_get_adapter.return_value = GetAdminProviderAdapter()
-        self.permissions_class._is_admin_on_provider(user, org)
+        is_admin_on_provider(user, org)
         assert mocked_get_adapter.return_value.last_call_args == {
             "username": user.username,
             "service_id": user.service_id,
@@ -192,4 +192,4 @@ def test_is_admin_on_provider_handles_torngit_exception(self, mock_get_provider)
         user = OwnerFactory()
 
         with self.assertRaises(APIException):
-            self.permissions_class._is_admin_on_provider(user, org)
+            is_admin_on_provider(user, org)

api/shared/permissions.py

@@ -140,6 +140,21 @@ def has_permission(self, request: HttpRequest, view: Any) -> bool:
         return True
 
 
+@torngit_safe
+def is_admin_on_provider(current_user: Owner, owner: Owner) -> bool:
+    torngit_provider_adapter = get_provider(
+        owner.service,
+        {
+            **get_generic_adapter_params(current_user, owner.service),
+            "owner": {"username": owner.username, "service_id": owner.service_id},
+        },
+    )
+
+    return async_to_sync(torngit_provider_adapter.get_is_admin)(
+        user={"username": current_user.username, "service_id": current_user.service_id}
+    )
+
+
 class UserIsAdminPermissions(BasePermission):
     """
     Permissions class for asserting the user is an admin of the 'owner'
@@ -158,29 +173,10 @@ def has_permission(self, request: HttpRequest, view: Any) -> bool:
                 and request.current_owner
                 and (
                     view.owner.is_admin(request.current_owner)
-                    or self._is_admin_on_provider(request.current_owner, view.owner)
+                    or is_admin_on_provider(request.current_owner, view.owner)
                 )
             )
 
-    @torngit_safe
-    def _is_admin_on_provider(self, user: Owner, owner: Owner) -> bool:
-        torngit_provider_adapter = get_provider(
-            owner.service,
-            {
-                **get_generic_adapter_params(user, owner.service),
-                **{
-                    "owner": {
-                        "username": owner.username,
-                        "service_id": owner.service_id,
-                    }
-                },
-            },
-        )
-
-        return async_to_sync(torngit_provider_adapter.get_is_admin)(
-            user={"username": user.username, "service_id": user.service_id}
-        )
-
 
 class MemberOfOrgPermissions(BasePermission):
     """

codecov_auth/commands/owner/interactors/get_is_current_user_an_admin.py

@@ -1,32 +1,9 @@
-from asgiref.sync import async_to_sync, sync_to_async
+from asgiref.sync import sync_to_async
 from django.conf import settings
 
 import services.self_hosted as self_hosted
+from api.shared.permissions import is_admin_on_provider
 from codecov.commands.base import BaseInteractor
-from services.decorators import torngit_safe
-from services.repo_providers import get_generic_adapter_params, get_provider
-
-
-@torngit_safe
-@sync_to_async
-def _is_admin_on_provider(owner, current_user):
-    torngit_provider_adapter = get_provider(
-        owner.service,
-        {
-            **get_generic_adapter_params(current_user, owner.service),
-            **{
-                "owner": {
-                    "username": owner.username,
-                    "service_id": owner.service_id,
-                }
-            },
-        },
-    )
-
-    isAdmin = async_to_sync(torngit_provider_adapter.get_is_admin)(
-        user={"username": current_user.username, "service_id": current_user.service_id}
-    )
-    return isAdmin
 
 
 class GetIsCurrentUserAnAdminInteractor(BaseInteractor):
@@ -44,7 +21,7 @@ def execute(self, owner, current_owner):
                 return True
             else:
                 try:
-                    isAdmin = async_to_sync(_is_admin_on_provider)(owner, current_owner)
+                    isAdmin = is_admin_on_provider(current_owner, owner)
                     if isAdmin:
                         # save admin provider in admins list
                         owner.add_admin(current_owner)

codecov_auth/commands/owner/interactors/tests/test_get_is_current_user_an_admin.py

@@ -7,10 +7,9 @@
     OwnerFactory,
 )
 
-from ..get_is_current_user_an_admin import (
-    GetIsCurrentUserAnAdminInteractor,
-    _is_admin_on_provider,
-)
+from api.shared.permissions import is_admin_on_provider
+
+from ..get_is_current_user_an_admin import GetIsCurrentUserAnAdminInteractor
 
 
 class GetIsCurrentUserAnAdminInteractorTest(TestCase):
@@ -43,22 +42,18 @@ def test_user_not_a_provider_admin(self):
         )(owner, current_user)
         assert isAdmin == False
 
-    @patch(
-        "codecov_auth.commands.owner.interactors.get_is_current_user_an_admin.get_provider"
-    )
+    @patch("api.shared.permissions.get_provider")
     def test_is_admin_on_provider_invokes_torngit_adapter(self, mocked_get_adapter):
         current_user = OwnerFactory(ownerid=3)
         owner = self.owner_has_no_admins
         mocked_get_adapter.return_value = GetAdminProviderAdapter()
-        async_to_sync(_is_admin_on_provider)(owner, current_user)
+        is_admin_on_provider(current_user, owner)
         assert mocked_get_adapter.return_value.last_call_args == {
             "username": current_user.username,
             "service_id": current_user.service_id,
         }
 
-    @patch(
-        "codecov_auth.commands.owner.interactors.get_is_current_user_an_admin.get_provider"
-    )
+    @patch("api.shared.permissions.get_provider")
     def test_is_admin_in_org_not_on_provider(self, mocked_get_adapter):
         current_user = OwnerFactory(ownerid=2)
         owner = self.owner_has_admins
@@ -68,9 +63,7 @@ def test_is_admin_in_org_not_on_provider(self, mocked_get_adapter):
         )(owner, current_user)
         assert isAdmin == True
 
-    @patch(
-        "codecov_auth.commands.owner.interactors.get_is_current_user_an_admin.get_provider"
-    )
+    @patch("api.shared.permissions.get_provider")
     def test_is_admin_on_provider(self, mocked_get_adapter):
         current_user = OwnerFactory(ownerid=3)
         owner = self.owner_has_no_admins
@@ -81,9 +74,7 @@ def test_is_admin_on_provider(self, mocked_get_adapter):
         assert current_user.ownerid in owner.admins
         assert isAdmin == True
 
-    @patch(
-        "codecov_auth.commands.owner.interactors.get_is_current_user_an_admin.get_provider"
-    )
+    @patch("api.shared.permissions.get_provider")
     def test_is_admin_on_provider_only_once(self, mocked_get_adapter):
         # Ensure duplicate ownerids won't be saved in admins upon multiple calls
         current_user = OwnerFactory(ownerid=3)
@@ -98,9 +89,7 @@ def test_is_admin_on_provider_only_once(self, mocked_get_adapter):
         assert owner.admins == [3]
         assert current_user.ownerid in owner.admins
 
-    @patch(
-        "codecov_auth.commands.owner.interactors.get_is_current_user_an_admin.get_provider"
-    )
+    @patch("api.shared.permissions.get_provider")
     def test_admin_on_provider_initially_is_null(self, mocked_get_adapter):
         # Owner model defaults admins to null, check can handle first update
         current_user = OwnerFactory(ownerid=3)
@@ -113,9 +102,7 @@ def test_admin_on_provider_initially_is_null(self, mocked_get_adapter):
         assert current_user.ownerid in owner.admins
         assert isAdmin == True
 
-    @patch(
-        "codecov_auth.commands.owner.interactors.get_is_current_user_an_admin.get_provider"
-    )
+    @patch("api.shared.permissions.get_provider")
     def test_is_admin_not_in_org_or_on_provider(self, mocked_get_adapter):
         current_user = OwnerFactory(ownerid=3)
         owner = self.owner_has_no_admins