Cyberstorm api create/delete service account (TS-2320)

django/thunderstore/api/cyberstorm/serializers/service_account.py

@@ -0,0 +1,7 @@
+from rest_framework import serializers
+
+
+class CreateServiceAccountSerializer(serializers.Serializer):
+    nickname = serializers.CharField(max_length=32)
+    team_name = serializers.CharField(read_only=True)
+    api_token = serializers.CharField(read_only=True)

django/thunderstore/api/cyberstorm/tests/test_create_service_account.py

@@ -0,0 +1,129 @@
+import json
+
+import pytest
+from django.contrib.auth import get_user_model
+from rest_framework.test import APIClient
+
+from thunderstore.account.models.service_account import ServiceAccount
+from thunderstore.repository.models.team import Team, TeamMember
+
+User = get_user_model()
+
+
+def get_create_service_account_url(team_name: str) -> str:
+    return f"/api/cyberstorm/team/{team_name}/service-account/create/"
+
+
+@pytest.mark.django_db
+def test_create_service_account_success(api_client: APIClient, team_owner: TeamMember):
+    api_client.force_authenticate(team_owner.user)
+
+    url = get_create_service_account_url(team_owner.team.name)
+    data = json.dumps({"nickname": "CoolestTeamServiceAccountName"})
+
+    response = api_client.post(url, data, content_type="application/json")
+
+    expected_response = {
+        "nickname": "CoolestTeamServiceAccountName",
+        "team_name": team_owner.team.name,
+        "api_token": "tss_",
+    }
+
+    service_account_count = ServiceAccount.objects.filter(
+        owner__name=team_owner.team.name,
+        user__first_name="CoolestTeamServiceAccountName",
+    ).count()
+
+    assert response.status_code == 201
+    assert response.json()["nickname"] == expected_response["nickname"]
+    assert response.json()["team_name"] == expected_response["team_name"]
+    assert response.json()["api_token"][:4] == expected_response["api_token"]
+    assert service_account_count == 1
+
+
+@pytest.mark.django_db
+def test_create_service_account_not_authenticated(
+    api_client: APIClient, team_owner: TeamMember
+):
+    url = get_create_service_account_url(team_owner.team.name)
+    data = json.dumps({"nickname": "CoolestTeamServiceAccountName"})
+
+    response = api_client.post(url, data, content_type="application/json")
+    expected_response = {"detail": "Authentication credentials were not provided."}
+
+    assert response.status_code == 401
+    assert response.json() == expected_response
+
+
+@pytest.mark.django_db
+def test_create_service_account_fails_because_nickname_too_long(
+    api_client: APIClient,
+    team_owner: TeamMember,
+):
+    api_client.force_authenticate(team_owner.user)
+    url = get_create_service_account_url(team_owner.team.name)
+    data = json.dumps({"nickname": "LongestCoolestTeamServiceAccountNameEver"})
+
+    response = api_client.post(url, data, content_type="application/json")
+
+    expected_response = {
+        "nickname": ["Ensure this field has no more than 32 characters."]
+    }
+
+    service_account_count = ServiceAccount.objects.filter(
+        owner__name=team_owner.team.name,
+        user__first_name="LongestCoolestTeamServiceAccountNameEver",
+    ).count()
+
+    assert response.status_code == 400
+    assert response.json() == expected_response
+    assert service_account_count == 0
+
+
+@pytest.mark.django_db
+def test_create_service_account_fail_because_user_is_not_team_member(
+    api_client: APIClient,
+    team: Team,
+):
+    non_team_user = User.objects.create()
+    api_client.force_authenticate(non_team_user)
+
+    url = get_create_service_account_url(team.name)
+    data = json.dumps({"nickname": "CoolestTeamServiceAccountName"})
+
+    response = api_client.post(url, data, content_type="application/json")
+    account_count = ServiceAccount.objects.filter(
+        owner__name=team.name, user__first_name="CoolestTeamServiceAccountName"
+    ).count()
+
+    expected_response = {"detail": "User does not have permission to access this team."}
+
+    assert response.status_code == 403
+    assert account_count == 0
+    assert response.json() == expected_response
+
+
+@pytest.mark.django_db
+def test_create_service_account_fail_because_user_is_not_team_owner(
+    api_client: APIClient,
+    team: Team,
+    team_member: TeamMember,
+):
+    api_client.force_authenticate(team_member.user)
+    url = get_create_service_account_url(team.name)
+    data = json.dumps({"nickname": "CoolestTeamServiceAccountName"})
+
+    response = api_client.post(url, data, content_type="application/json")
+    account_count = ServiceAccount.objects.filter(
+        owner__name=team.name, user__first_name="CoolestTeamServiceAccountName"
+    ).count()
+
+    expected_response = {
+        "detail": (
+            "User does not have permission to create service accounts for this team."
+        )
+    }
+
+    assert response.status_code == 403
+    assert account_count == 0
+    assert response.json() == expected_response

django/thunderstore/api/cyberstorm/tests/test_delete_service_account.py

@@ -0,0 +1,91 @@
+import pytest
+from django.contrib.auth import get_user_model
+from rest_framework.test import APIClient
+
+from thunderstore.account.models.service_account import ServiceAccount
+from thunderstore.repository.models.team import Team, TeamMember
+
+User = get_user_model()
+
+
+def get_delete_service_account_url(team_name: str, uuid: str) -> str:
+    return f"/api/cyberstorm/team/{team_name}/service-account/delete/{uuid}/"
+
+
+def make_request(api_client: APIClient, team_name: str, account: ServiceAccount):
+    return api_client.delete(
+        path=get_delete_service_account_url(team_name, account.uuid),
+        content_type="application/json",
+    )
+
+
+@pytest.mark.django_db
+def test_delete_service_account_success(
+    api_client: APIClient,
+    team_owner: TeamMember,
+    service_account: ServiceAccount,
+):
+    assert ServiceAccount.objects.filter(uuid=service_account.uuid).count() == 1
+
+    api_client.force_authenticate(team_owner.user)
+    response = make_request(api_client, team_owner.team.name, service_account)
+
+    assert response.status_code == 204
+    assert ServiceAccount.objects.filter(uuid=service_account.uuid).count() == 0
+
+
+@pytest.mark.django_db
+def test_delete_service_account_fail_user_is_not_authenticated(
+    api_client: APIClient,
+    team: Team,
+    service_account: ServiceAccount,
+):
+    assert ServiceAccount.objects.filter(uuid=service_account.uuid).count() == 1
+
+    response = make_request(api_client, team.name, service_account)
+    expected_response = {"detail": "Authentication credentials were not provided."}
+
+    assert response.status_code == 401
+    assert response.json() == expected_response
+    assert ServiceAccount.objects.filter(uuid=service_account.uuid).count() == 1
+
+
+@pytest.mark.django_db
+def test_delete_service_account_fails_because_user_is_not_team_member(
+    api_client: APIClient,
+    team: Team,
+    service_account: ServiceAccount,
+):
+    assert ServiceAccount.objects.filter(uuid=service_account.uuid).count() == 1
+
+    non_team_user = User.objects.create()
+    api_client.force_authenticate(non_team_user)
+
+    response = make_request(api_client, team.name, service_account)
+    expected_response = {"detail": "User does not have permission to access this team."}
+
+    assert response.status_code == 403
+    assert response.json() == expected_response
+    assert ServiceAccount.objects.filter(uuid=service_account.uuid).count() == 1
+
+
+@pytest.mark.django_db
+def test_delete_service_account_fail_because_user_is_not_team_owner(
+    api_client: APIClient,
+    team_member: TeamMember,
+    team: Team,
+    service_account: ServiceAccount,
+):
+    assert ServiceAccount.objects.filter(uuid=service_account.uuid).count() == 1
+
+    api_client.force_authenticate(team_member.user)
+    response = make_request(api_client, team.name, service_account)
+
+    error_message = (
+        "User does not have permission to delete service accounts for this team."
+    )
+    expected_response = {"detail": error_message}
+
+    assert response.status_code == 403
+    assert response.json() == expected_response
+    assert ServiceAccount.objects.filter(uuid=service_account.uuid).count() == 1

django/thunderstore/api/cyberstorm/views/__init__.py

@@ -11,6 +11,7 @@
 )
 from .package_rating import RatePackageAPIView
 from .package_version_list import PackageVersionListAPIView
+from .service_account import CreateServiceAccountAPIView, DeleteServiceAccountAPIView
 from .team import (
     TeamAPIView,
     TeamMemberAddAPIView,
@@ -23,6 +24,8 @@
     "CommunityFiltersAPIView",
     "CommunityListAPIView",
     "DeprecatePackageAPIView",
+    "CreateServiceAccountAPIView",
+    "DeleteServiceAccountAPIView",
     "PackageListingAPIView",
     "PackageListingByCommunityListAPIView",
     "PackageListingByDependencyListAPIView",

django/thunderstore/api/cyberstorm/views/service_account.py

@@ -0,0 +1,94 @@
+from django.http import HttpRequest
+from rest_framework import status
+from rest_framework.exceptions import PermissionDenied
+from rest_framework.generics import CreateAPIView, DestroyAPIView, get_object_or_404
+from rest_framework.permissions import IsAuthenticated
+from rest_framework.response import Response
+
+from thunderstore.account.models import ServiceAccount
+from thunderstore.api.cyberstorm.serializers.service_account import (
+    CreateServiceAccountSerializer,
+)
+from thunderstore.api.utils import conditional_swagger_auto_schema
+from thunderstore.repository.models import Team
+
+
+class TeamPermissionMixin:
+    permission_classes = [IsAuthenticated]
+
+    def dispatch(self, request, *args, **kwargs):
+        self.team = self.get_team()
+        return super().dispatch(request, *args, **kwargs)
+
+    def get_team(self) -> Team:
+        team_name = self.kwargs.get("team_name")
+        return get_object_or_404(Team, name__iexact=team_name)
+
+    def check_permissions(self, request: HttpRequest) -> None:
+        super().check_permissions(request)
+        self.check_user_team_permissions()
+
+    def check_user_team_permissions(self) -> None:
+        if not self.team.can_user_access(self.request.user):
+            raise PermissionDenied("User does not have permission to access this team.")
+
+
+class CreateServiceAccountAPIView(TeamPermissionMixin, CreateAPIView):
+    queryset = ServiceAccount.objects.all()
+    serializer_class = CreateServiceAccountSerializer
+
+    def check_permissions(self, request: HttpRequest) -> None:
+        super().check_permissions(request)
+        if not self.team.can_user_create_service_accounts(request.user):
+            raise PermissionDenied(
+                "User does not have permission to create service accounts "
+                "for this team."
+            )
+
+    def perform_create(self, serializer: CreateServiceAccountSerializer):
+        nickname = serializer.validated_data["nickname"]
+        service_account_data = self.create_service_account(nickname)
+        serializer.instance = service_account_data
+
+    def create_service_account(self, nickname: str):
+        service_account, token = ServiceAccount.create(
+            owner=self.team,
+            nickname=nickname,
+            creator=self.request.user,
+        )
+
+        return {
+            "nickname": service_account.nickname,
+            "team_name": service_account.owner.name,
+            "api_token": token,
+        }
+
+    @conditional_swagger_auto_schema(
+        request_body=serializer_class,
+        responses={status.HTTP_201_CREATED: serializer_class},
+        operation_id="cyberstorm.team.service-account.create",
+        tags=["cyberstorm"],
+    )
+    def post(self, request, *args, **kwargs) -> Response:
+        return super().post(request, *args, **kwargs)
+
+
+class DeleteServiceAccountAPIView(TeamPermissionMixin, DestroyAPIView):
+    queryset = ServiceAccount.objects.all()
+    lookup_field = "uuid"
+
+    def check_permissions(self, request: HttpRequest) -> None:
+        super().check_permissions(request)
+        if not self.team.can_user_delete_service_accounts(request.user):
+            raise PermissionDenied(
+                "User does not have permission to delete service accounts "
+                "for this team."
+            )
+
+    @conditional_swagger_auto_schema(
+        responses={status.HTTP_204_NO_CONTENT: ""},
+        operation_id="cyberstorm.team.service-account.delete",
+        tags=["cyberstorm"],
+    )
+    def delete(self, request, *args, **kwargs) -> Response:
+        return super().delete(request, *args, **kwargs)

django/thunderstore/api/urls.py

@@ -4,6 +4,8 @@
     CommunityAPIView,
     CommunityFiltersAPIView,
     CommunityListAPIView,
+    CreateServiceAccountAPIView,
+    DeleteServiceAccountAPIView,
     DeprecatePackageAPIView,
     PackageListingAPIView,
     PackageListingByCommunityListAPIView,
@@ -110,4 +112,14 @@
         TeamServiceAccountListAPIView.as_view(),
         name="cyberstorm.team.service-account",
     ),
+    path(
+        "team/<str:team_name>/service-account/create/",
+        CreateServiceAccountAPIView.as_view(),
+        name="cyberstorm.team.service-account.create",
+    ),
+    path(
+        "team/<str:team_name>/service-account/delete/<uuid:uuid>/",
+        DeleteServiceAccountAPIView.as_view(),
+        name="cyberstorm.team.service-account.delete",
+    ),
 ]