Refactor PackageListing ensure_x functions (TS-2580)

django/thunderstore/api/cyberstorm/services/package_listing.py

@@ -11,7 +11,9 @@
 def update_categories(
     agent: UserType, categories: list, listing: PackageListing
 ) -> None:
-    listing.ensure_update_categories_permission(agent)
+    validation_result = listing.validate_update_categories_permissions(agent)
+    validation_result.raise_if_invalid()
+
     listing.update_categories(agent=agent, categories=categories)
 
     get_package_listing_or_404.clear_cache_with_args(

django/thunderstore/community/models/package_listing.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, List, Optional
+from typing import TYPE_CHECKING, List, Optional, Tuple
 
 from django.conf import settings
 from django.core.exceptions import ValidationError
@@ -11,14 +11,12 @@
 from thunderstore.cache.enums import CacheBustCondition
 from thunderstore.cache.tasks import invalidate_cache_on_commit_async
 from thunderstore.community.consts import PackageListingReviewStatus
-from thunderstore.core.exceptions import PermissionValidationError
 from thunderstore.core.mixins import AdminLinkMixin, TimestampMixin
 from thunderstore.core.types import UserType
-from thunderstore.core.utils import check_validity
 from thunderstore.frontend.url_reverse import get_community_url_reverse_args
 from thunderstore.permissions.mixins import VisibilityMixin
 from thunderstore.permissions.models.visibility import VisibilityFlagsQuerySet
-from thunderstore.permissions.utils import validate_user
+from thunderstore.permissions.utils import PermissionResult, check_user_permissions
 from thunderstore.webhooks.audit import (
     AuditAction,
     AuditEvent,
@@ -321,7 +319,9 @@ def is_rejected(self):
         return self.review_status == PackageListingReviewStatus.rejected
 
     def update_categories(self, agent: UserType, categories: List["PackageCategory"]):
-        self.ensure_update_categories_permission(agent)
+        if not self.can_update_categories_permission(agent):
+            raise PermissionError()
+
         for category in categories:
             if category.community_id != self.community_id:
                 raise ValidationError(
@@ -336,33 +336,51 @@ def is_unavailable(self):
     def can_be_moderated_by_user(self, user: Optional[UserType]) -> bool:
         return self.community.can_user_manage_packages(user)
 
-    def ensure_user_can_manage_listing(self, user: Optional[UserType]) -> None:
-        user = validate_user(user)
+    def validate_user_can_manage_listing(
+        self, user: Optional[UserType]
+    ) -> PermissionResult:
+
+        result = check_user_permissions(user)
+        if not result.is_valid:
+            return result
+
         is_allowed = self.can_be_moderated_by_user(
             user
         ) or self.package.owner.can_user_manage_packages(user)
+
         if not is_allowed:
-            raise PermissionValidationError("Must have listing management permission")
+            return PermissionResult(error="Must have listing management permission")
+
+        return PermissionResult()
+
+    def validate_update_categories_permissions(
+        self, user: Optional[UserType]
+    ) -> PermissionResult:
+
+        result: PermissionResult = check_user_permissions(user)
 
-    def ensure_update_categories_permission(self, user: Optional[UserType]) -> None:
-        user = validate_user(user)
-        is_allowed = (
+        if not result.is_valid:
+            return result
+
+        if not (
             self.can_be_moderated_by_user(user)
             or self.package.owner.can_user_manage_packages(user)
             or self.community.can_user_manage_categories(user)
-        )
-        if not is_allowed:
-            raise PermissionValidationError(
-                "User is missing necessary roles or permissions"
-            )
+        ):
+            error = "User is missing necessary roles or permissions"
+            return PermissionResult(error=error)
+
+        return PermissionResult()
 
-    def check_update_categories_permission(self, user: Optional[UserType]) -> bool:
-        return check_validity(lambda: self.ensure_update_categories_permission(user))
+    def can_update_categories_permission(self, user: Optional[UserType]) -> bool:
+        return self.validate_update_categories_permissions(user).is_valid
 
     def can_user_manage_approval_status(self, user: Optional[UserType]) -> bool:
         return self.can_be_moderated_by_user(user)
 
-    def ensure_can_be_viewed_by_user(self, user: Optional[UserType]) -> None:
+    def validate_can_be_viewed_by_user(
+        self, user: Optional[UserType]
+    ) -> PermissionResult:
         def get_has_perms() -> bool:
             return (
                 user is not None
@@ -378,16 +396,18 @@ def get_has_perms() -> bool:
                 self.review_status != PackageListingReviewStatus.approved
                 and not get_has_perms()
             ):
-                raise PermissionValidationError("Insufficient permissions to view")
+                return PermissionResult(error="Insufficient permissions to view")
         else:
             if (
                 self.review_status == PackageListingReviewStatus.rejected
                 and not get_has_perms()
             ):
-                raise PermissionValidationError("Insufficient permissions to view")
+                return PermissionResult(error="Insufficient permissions to view")
+
+        return PermissionResult()
 
     def can_be_viewed_by_user(self, user: Optional[UserType]) -> bool:
-        return check_validity(lambda: self.ensure_can_be_viewed_by_user(user))
+        return self.validate_can_be_viewed_by_user(user).is_valid
 
     def is_visible_to_user(self, user: Optional[UserType]) -> bool:
         if not self.visibility:

django/thunderstore/community/tests/test_package_listing.py

@@ -184,7 +184,7 @@ def test_package_listing_is_rejected(
 @pytest.mark.parametrize("user_type", TestUserTypes.options())
 @pytest.mark.parametrize("team_role", TeamMemberRole.options() + [None])
 @pytest.mark.parametrize("community_role", CommunityMemberRole.options() + [None])
-def test_package_listing_ensure_can_be_viewed_by_user(
+def test_package_listing_validate_can_be_viewed_by_user(
     active_package_listing: PackageListing,
     require_approval: bool,
     review_status: str,
@@ -214,52 +214,48 @@ def test_package_listing_ensure_can_be_viewed_by_user(
         )
 
     result = listing.can_be_viewed_by_user(user)
-    errors = []
     expected_error = "Insufficient permissions to view"
-    try:
-        listing.ensure_can_be_viewed_by_user(user)
-    except ValidationError as e:
-        errors = e.messages
+    validation_result = listing.validate_can_be_viewed_by_user(user)
 
     if require_approval:
         if review_status == PackageListingReviewStatus.approved:
             assert result is True
-            assert not errors
+            assert not validation_result.error
         elif user is None:
             assert result is False
-            assert expected_error in errors
+            assert expected_error == validation_result.error
         elif not user.is_authenticated:
             assert result is False
-            assert expected_error in errors
+            assert expected_error == validation_result.error
         elif community.can_user_manage_packages(user):
             assert result is True
-            assert not errors
+            assert not validation_result.error
         elif listing.package.owner.can_user_access(user):
             assert result is True
-            assert not errors
+            assert not validation_result.error
     else:
         if review_status != PackageListingReviewStatus.rejected:
             assert result is True
-            assert not errors
+            assert not validation_result.error
         elif user is None:
             assert result is False
-            assert expected_error in errors
+            assert expected_error == validation_result.error
         elif not user.is_authenticated:
             assert result is False
-            assert expected_error in errors
+            assert expected_error == validation_result.error
         elif community.can_user_manage_packages(user):
             assert result is True
-            assert not errors
+            assert not validation_result.error
         elif listing.package.owner.can_user_access(user):
             assert result is True
-            assert not errors
+            assert not validation_result.error
 
 
 @pytest.mark.django_db
 @pytest.mark.parametrize("user_type", TestUserTypes.options())
 @pytest.mark.parametrize("team_role", TeamMemberRole.options() + [None])
 @pytest.mark.parametrize("community_role", CommunityMemberRole.options() + [None])
-def test_package_listing_ensure_update_categories_permission(
+def test_package_listing_validate_update_categories_permission(
     active_package_listing: PackageListing,
     user_type: str,
     community_role: str,
@@ -282,12 +278,8 @@ def test_package_listing_ensure_update_categories_permission(
             role=team_role,
         )
 
-    result = listing.check_update_categories_permission(user)
-    errors = []
-    try:
-        listing.ensure_update_categories_permission(user)
-    except ValidationError as e:
-        errors = e.messages
+    can_update = listing.can_update_categories_permission(user)
+    validation_result = listing.validate_update_categories_permissions(user)
 
     has_perms = any(
         (
@@ -313,33 +305,25 @@ def test_package_listing_ensure_update_categories_permission(
     expected_error = error_map[user_type]
 
     if expected_error:
-        assert result is False
-        assert len(errors) == 1
-        assert errors[0] == expected_error
+        assert can_update is False
+        assert validation_result.error == expected_error
     else:
-        assert result is True
-        assert not errors
+        assert can_update is True
+        assert validation_result.error is None
 
 
 @pytest.mark.django_db
 def test_package_listing_update_categories(
     active_package_listing: PackageListing,
     package_category: PackageCategory,
     team_owner: TeamMember,
-    mocker,
 ):
+    user = team_owner.user
+
     assert package_category.community == active_package_listing.community
     assert active_package_listing.package.owner == team_owner.team
     assert active_package_listing.categories.count() == 0
-    mocked_permission_check = mocker.patch.object(
-        active_package_listing,
-        "ensure_update_categories_permission",
-    )
-    active_package_listing.update_categories(
-        agent=team_owner.user,
-        categories=[package_category],
-    )
-    mocked_permission_check.assert_called_with(team_owner.user)
+    active_package_listing.update_categories(agent=user, categories=[package_category])
     assert package_category in active_package_listing.categories.all()
 
     invalid_category = PackageCategory.objects.create(
@@ -353,8 +337,7 @@ def test_package_listing_update_categories(
         match="Community mismatch between package listing and category",
     ):
         active_package_listing.update_categories(
-            agent=team_owner.user,
-            categories=[invalid_category],
+            agent=user, categories=[invalid_category]
         )
 
 
@@ -437,6 +420,32 @@ def test_package_listing_has_mod_manager_support(mod_manager_support: bool) -> N
     assert package_listing.has_mod_manager_support == mod_manager_support
 
 
+@pytest.mark.django_db
+@pytest.mark.parametrize("user_type", TestUserTypes.options())
+def test_package_listing_validate_user_can_manage_listing(
+    user_type,
+    active_package_listing,
+) -> None:
+    user = TestUserTypes.get_user_by_type(user_type)
+
+    valid_user_type_map = {
+        TestUserTypes.no_user: ("Must be authenticated", True),
+        TestUserTypes.unauthenticated: ("Must be authenticated", True),
+        TestUserTypes.regular_user: ("Must have listing management permission", True),
+        TestUserTypes.deactivated_user: ("User has been deactivated", False),
+        TestUserTypes.service_account: (
+            "Service accounts are unable to perform this action",
+            True,
+        ),
+        TestUserTypes.site_admin: (None, True),
+        TestUserTypes.superuser: (None, True),
+    }
+
+    validation_result = active_package_listing.validate_user_can_manage_listing(user)
+    assert validation_result.error == valid_user_type_map[user_type][0]
+    assert validation_result.is_public == valid_user_type_map[user_type][1]
+
+
 # TODO: Re-enable once visibility system fixed
 # @pytest.mark.django_db
 # def test_package_listing_visibility_inherits_package_is_active(

django/thunderstore/permissions/tests/test_validate_user.py

@@ -1,9 +1,14 @@
 import pytest
 from django.contrib.auth.models import AnonymousUser
-from django.core.exceptions import ValidationError
+from django.core.exceptions import PermissionDenied, ValidationError
 
 from thunderstore.account.models import ServiceAccount
-from thunderstore.permissions.utils import validate_user
+from thunderstore.core.exceptions import PermissionValidationError
+from thunderstore.permissions.utils import (
+    PermissionResult,
+    check_user_permissions,
+    validate_user,
+)
 
 
 @pytest.mark.parametrize("fake_user", (None, AnonymousUser()))
@@ -35,3 +40,71 @@ def test_permissions_validate_user_serviceaccount_disallowed(
         ValidationError, match="Service accounts are unable to perform this action"
     ):
         validate_user(service_account.user)
+
+
+@pytest.mark.parametrize("fake_user", (None, AnonymousUser()))
+def test_check_user_permissions_not_signed_in(fake_user):
+    result = check_user_permissions(fake_user)
+    assert not result.is_valid
+    assert result.error == "Must be authenticated"
+    assert result.is_public is True
+
+
+@pytest.mark.django_db
+def test_check_user_permissions_inactive(user):
+    user.is_active = False
+    user.save()
+
+    result = check_user_permissions(user)
+    assert not result.is_valid
+    assert result.error == "User has been deactivated"
+    assert result.is_public is False
+
+
+@pytest.mark.django_db
+def test_check_user_permissions_serviceaccount_allowed(service_account):
+    result = check_user_permissions(service_account.user, allow_serviceaccount=True)
+    assert result.is_valid
+    assert result.error is None
+    assert result.is_public is True
+
+
+@pytest.mark.django_db
+def test_check_user_permissions_serviceaccount_disallowed(service_account):
+    result = check_user_permissions(service_account.user, allow_serviceaccount=False)
+    assert not result.is_valid
+    assert result.error == "Service accounts are unable to perform this action"
+    assert result.is_public is True
+
+
+def test_permission_result_valid_by_default():
+    result = PermissionResult()
+    assert result.is_valid
+    assert result.error is None
+    assert result.is_public is True
+
+
+def test_permission_result_invalid_sets_error():
+    result = PermissionResult(error="Something went wrong")
+    assert not result.is_valid
+    assert result.error == "Something went wrong"
+    assert result.is_public is True
+
+
+def test_permission_result_raise_if_invalid_valid_case_does_not_raise():
+    result = PermissionResult()
+    result.raise_if_invalid()  # Should not raise an error
+
+
+@pytest.mark.parametrize("is_public", (True, False))
+def test_permission_result_raise_if_invalid_invalid_case_raises(is_public):
+    result = PermissionResult(error="Forbidden", is_public=is_public)
+    with pytest.raises(PermissionValidationError, match="Forbidden") as exc_info:
+        result.raise_if_invalid()
+    assert exc_info.value.is_public is is_public
+
+
+def test_permission_result_raise_if_invalid_with_custom_exception():
+    result = PermissionResult(error="SomeOtherError")
+    with pytest.raises(ValidationError, match="SomeOtherError"):
+        result.raise_if_invalid(exc_cls=ValidationError)