WIP: Permission framework migration

Convert more code from team services to the experimental permission framework

Author: MythicManiac

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

@@ -1,5 +1,5 @@
 import dataclasses
-from typing import Any, Generator, List, Optional, Sequence
+from typing import List, Optional, Sequence, Union
 
 from django.core.exceptions import ValidationError
 from django.db import transaction
@@ -12,109 +12,195 @@
 from thunderstore.repository.models.team import TeamMemberRole
 
 
+class CheckException(ValidationError):
+    pass
+
+
 @dataclasses.dataclass
 class CheckResult:
-    check_pass: bool
+    success: bool
     errors: Optional[List[Exception]]
 
+    def as_exception(self) -> Union[CheckException, ValidationError, Exception]:
+        if len(self.errors) == 1:
+            return self.errors[0]
+        else:
+            return CheckException(message=self.errors)
 
-class OpCheck:
-    """OperationCheck"""
 
-    def run_check(self, _: UserType) -> CheckResult:
+class OpCheck:
+    def run_check(self, agent: UserType) -> CheckResult:
         return CheckResult(
-            check_pass=False, errors=[NotImplementedError("Check not implemented")]
+            success=False, errors=[ValidationError("Check not implemented")]
         )
 
 
+@dataclasses.dataclass
 class TeamOpCheck(OpCheck):
     team: Team
 
-    def __init__(self, team: Team):
-        self.team = team
 
+class CheckTeamAccessPermission(TeamOpCheck):
+    def run_check(self, agent: UserType) -> CheckResult:
+        try:
+            self.team.ensure_user_can_access(agent)
+            return CheckResult(success=True, errors=None)
+        except Exception as e:
+            return CheckResult(success=False, errors=[e])
+
+
+class CheckTeamDisbandPermission(TeamOpCheck):
+    def run_check(self, agent: UserType) -> CheckResult:
+        try:
+            self.team.ensure_user_can_disband(agent)
+            return CheckResult(success=True, errors=None)
+        except Exception as e:
+            return CheckResult(success=False, errors=[e])
 
-class TeamAccessCheck(TeamOpCheck):
-    def run_check(self, user: UserType) -> CheckResult:
+
+class CheckTeamEditPermission(TeamOpCheck):
+    def run_check(self, agent: UserType) -> CheckResult:
         try:
-            self.team.ensure_user_can_access(user)
-            return CheckResult(check_pass=True, errors=None)
+            self.team.ensure_user_can_edit_info(agent)
+            return CheckResult(success=True, errors=None)
         except Exception as e:
-            return CheckResult(check_pass=False, errors=[e])
+            return CheckResult(success=False, errors=[e])
+
+
+class CheckTeamMemberManagePermission(TeamOpCheck):
+    def run_check(self, agent: UserType) -> CheckResult:
+        try:
+            self.team.ensure_user_can_manage_members(agent)
+            return CheckResult(success=True, errors=None)
+        except Exception as e:
+            return CheckResult(success=False, errors=[e])
+
 
+@dataclasses.dataclass
+class CheckTeamMemberCanBeRemoved(OpCheck):
+    member: TeamMember
 
-class TeamDisbandCheck(TeamOpCheck):
-    def run_check(self, user: UserType) -> CheckResult:
+    def run_check(self, _: UserType) -> CheckResult:
         try:
-            self.team.ensure_user_can_disband(user)
-            return CheckResult(check_pass=True, errors=None)
+            self.member.team.ensure_member_can_be_removed(self.member)
+            return CheckResult(success=True, errors=None)
         except Exception as e:
-            return CheckResult(check_pass=False, errors=[e])
+            return CheckResult(success=False, errors=[e])
 
 
-def _run_checks(
-    agent: UserType, checks: Sequence[OpCheck]
-) -> Generator[CheckResult, Any, None]:
-    return (check.run_check(agent) for check in checks)
+@dataclasses.dataclass
+class CheckTeamNameFree(OpCheck):
+    team_name: str
 
+    def run_check(self, agent: UserType) -> CheckResult:
+        if Team.objects.filter(name__iexact=self.team_name).exists():
+            return CheckResult(
+                success=False,
+                errors=[ValidationError("Team with this name already exists")],
+            )
+        else:
+            return CheckResult(success=True, errors=None)
 
-def run_checks(agent: UserType, checks: Sequence[OpCheck]) -> CheckResult:
-    results = _run_checks(agent, checks)
+
+@dataclasses.dataclass
+class CheckNamespaceNameFree(OpCheck):
+    namespace_name: str
+
+    def run_check(self, agent: UserType) -> CheckResult:
+        if Namespace.objects.filter(name__iexact=self.namespace_name).exists():
+            return CheckResult(
+                success=False,
+                errors=[ValidationError("Namespace with this name already exists")],
+            )
+        else:
+            return CheckResult(success=True, errors=None)
+
+
+class CheckUserIsAuthenticated(OpCheck):
+    def run_check(self, agent: UserType) -> CheckResult:
+        if not agent or not agent.is_authenticated or not agent.is_active:
+            return CheckResult(
+                success=False,
+                errors=[PermissionValidationError("User must be authenticated")],
+            )
+        else:
+            return CheckResult(success=True, errors=None)
+
+
+class CheckUserIsNotServiceAccount(OpCheck):
+    def run_check(self, agent: UserType) -> CheckResult:
+        if getattr(agent, "service_account", None) is not None:
+            raise PermissionValidationError(
+                "Service accounts cannot perform this action"
+            )
+
+
+def run_checks(agent: UserType, checks: Sequence[Union[OpCheck, None]]) -> CheckResult:
+    results = (check.run_check(agent) for check in checks if check)
     errors = []
     check_pass = True
     for entry in results:
         if entry.errors:
             errors += entry.errors
-        if not entry.check_pass:
+        if not entry.success:
             check_pass = False
-    return CheckResult(check_pass=check_pass, errors=errors)
+    return CheckResult(success=check_pass, errors=errors)
 
 
 @transaction.atomic
 def disband_team(agent: UserType, team: Team) -> None:
     checks = [
-        TeamAccessCheck(team),
-        TeamDisbandCheck(team),
+        CheckTeamAccessPermission(team=team),
+        CheckTeamDisbandPermission(team=team),
     ]
-    if (check_result := run_checks(agent, checks)).check_pass:
+    if (check_result := run_checks(agent, checks)).success:
         team.delete()
     else:
-        raise check_result.errors
+        raise check_result.as_exception()
 
 
 @transaction.atomic
 def create_team(agent: UserType, team_name: str) -> Team:
-    if not agent or not agent.is_authenticated or not agent.is_active:
-        raise PermissionValidationError("Must be authenticated to create teams")
-    if getattr(agent, "service_account", None) is not None:
-        raise PermissionValidationError("Service accounts cannot create teams")
-    if Team.objects.filter(name__iexact=team_name).exists():
-        raise ValidationError("Team with this name already exists")
-    if Namespace.objects.filter(name__iexact=team_name).exists():
-        raise ValidationError("Namespace with this name already exists")
-
-    team = Team.create(name=team_name)
-    team.add_member(user=agent, role=TeamMemberRole.owner)
-    return team
+    checks = [
+        CheckUserIsAuthenticated(),
+        CheckUserIsNotServiceAccount(),
+        CheckTeamNameFree(team_name=team_name),
+        CheckNamespaceNameFree(namespace_name=team_name),
+    ]
+    if (result := run_checks(agent, checks)).success:
+        team = Team.create(name=team_name)
+        team.add_member(user=agent, role=TeamMemberRole.owner)
+        return team
+    else:
+        raise result.as_exception()
 
 
 @transaction.atomic
 def update_team(agent: UserType, team: Team, donation_link: str) -> Team:
-    team.ensure_user_can_access(agent)
-    team.ensure_user_can_edit_info(agent)
-
-    team.donation_link = donation_link
-    team.save()
-
-    return team
+    checks = [
+        CheckTeamAccessPermission(team=team),
+        CheckTeamEditPermission(team=team),
+    ]
+    if (result := run_checks(agent, checks)).success:
+        team.donation_link = donation_link
+        team.save()
+        return team
+    else:
+        raise result.as_exception()
 
 
 @transaction.atomic
 def remove_team_member(agent: UserType, member: TeamMember) -> None:
-    if member.user != agent:
-        member.team.ensure_user_can_manage_members(agent)
-    member.team.ensure_member_can_be_removed(member)
-    member.delete()
+    checks = [
+        CheckTeamMemberManagePermission(team=member.team)
+        if member.user != agent
+        else None,
+        CheckTeamMemberCanBeRemoved(member=member),
+    ]
+    if (result := run_checks(agent, checks)).success:
+        member.delete()
+    else:
+        raise result.as_exception()
 
 
 @transaction.atomic