Merge pull request #249 from PROCOLLAB-github/dev

Фильтры и типизация

Author: yakser

core/pagination.py

@@ -0,0 +1,7 @@
+from rest_framework import pagination
+
+
+class Pagination(pagination.LimitOffsetPagination):
+    default_limit = 10
+    limit_query_param = "limit"
+    offset_query_param = "offset"

invites/tests.py

@@ -6,7 +6,7 @@
 
 from users.views import UserList
 from users.models import CustomUser
-from invites.views import InviteList, InviteDetail
+from invites.views import InviteList, InviteDetail, InviteAccept, InviteDecline
 from invites.models import Invite
 from industries.models import Industry
 from projects.views import ProjectList, ProjectDetail
@@ -114,6 +114,69 @@ def test_invites_creation_with_empty_data(self):
 
         self.assertEqual(response.status_code, 400)
 
+    def test_accept_invite_by_intended_user(self):
+        sender = self._user_create("sender@example.com")
+        recipient = self._user_create("recipient@example.com")
+        project = self._project_create(sender)
+
+        invite = self._create_invite(sender, recipient, project)
+
+        request = self.factory.post(f"/invites/{invite.id}/accept/")
+        force_authenticate(request, user=recipient)
+        accept_response = InviteAccept.as_view()(request, pk=invite.id)
+
+        self.assertEqual(accept_response.status_code, 200)
+        invite.refresh_from_db()
+        self.assertTrue(invite.is_accepted)
+
+    def test_decline_invite_by_intended_user(self):
+        sender = self._user_create("sender@example.com")
+        recipient = self._user_create("recipient@example.com")
+        project = self._project_create(sender)
+
+        invite = self._create_invite(sender, recipient, project)
+
+        request = self.factory.post(f"/invites/{invite.id}/decline/")
+        force_authenticate(request, user=recipient)
+        decline_response = InviteDecline.as_view()(request, pk=invite.id)
+
+        self.assertEqual(decline_response.status_code, 200)
+        invite.refresh_from_db()
+        self.assertFalse(invite.is_accepted)
+
+    def test_accept_decline_invite_by_unintended_user(self):
+        sender = self._user_create("sender@example.com")
+        recipient = self._user_create("recipient@example.com")
+        unintended_user = self._user_create("unintended@example.com")
+        project = self._project_create(sender)
+
+        invite = self._create_invite(sender, recipient, project)
+
+        accept_request = self.factory.post(f"/invites/{invite.id}/accept/")
+        force_authenticate(accept_request, user=unintended_user)
+        accept_response = InviteAccept.as_view()(accept_request, pk=invite.id)
+
+        decline_request = self.factory.post(f"/invites/{invite.id}/decline/")
+        force_authenticate(decline_request, user=unintended_user)
+        decline_response = InviteDecline.as_view()(decline_request, pk=invite.id)
+
+        self.assertNotEqual(accept_response.status_code, 200)
+        self.assertNotEqual(decline_response.status_code, 200)
+
+    def test_delete_invite_by_sender(self):
+        sender = self._user_create("sender@example.com")
+        recipient = self._user_create("recipient@example.com")
+        project = self._project_create(sender)
+
+        invite = self._create_invite(sender, recipient, project)
+
+        request = self.factory.delete(f"/invites/{invite.id}/")
+        force_authenticate(request, user=sender)
+        delete_response = self.invite_detail_view(request, pk=invite.id)
+
+        self.assertEqual(delete_response.status_code, 204)
+        self.assertFalse(Invite.objects.filter(pk=invite.id).exists())
+
     def _project_create(self, user):
         request = self.factory.post("projects/", self.project_create_data)
         force_authenticate(request, user=user)
@@ -132,3 +195,13 @@ def _user_create(self, email):
         user.is_active = True
         user.save()
         return user
+
+    def _create_invite(self, sender, recipient, project):
+        invite_data = self.invite_create_data.copy()
+        invite_data.update({"project": project.id, "user": recipient.id})
+
+        request = self.factory.post("/invites/", invite_data, format="json")
+        force_authenticate(request, user=sender)
+        response = self.invite_list_view(request)
+
+        return Invite.objects.get(pk=response.data["id"])

mailing/constants.py

@@ -7,3 +7,6 @@ def get_default_mailing_schema() -> dict[str, dict[str, str]]:
         "text": {"title": "Основной текст письма"},
         "button_text": {"title": "Текст кнопки", "default": "Кнопка"},
     }
+
+
+MAILING_USERS_BATCH_SIZE = 100

mailing/utils.py

@@ -1,4 +1,7 @@
 from typing import Dict, List, Union
+from .constants import MAILING_USERS_BATCH_SIZE
+from .models import MailingSchema
+from users.models import CustomUser
 
 import django.db.models
 from django.contrib.auth import get_user_model
@@ -9,29 +12,64 @@
 User = get_user_model()
 
 
+def prepare_mail_data(post_data) -> dict:
+    users = post_data.getlist("users[]")
+    schema_id = post_data["schemas"]
+    subject = post_data["subject"]
+    mail_schema = MailingSchema.objects.get(pk=schema_id)
+    context = {}
+    for variable_name in mail_schema.schema:
+        key_in_post = "field-" + variable_name
+        if key_in_post in post_data:
+            context[variable_name] = post_data[key_in_post]
+    users_to_send = CustomUser.objects.filter(pk__in=users)
+    data_dict = {
+        "users_to_send": users_to_send,
+        "subject": subject,
+        "mail_schema_template": mail_schema.template,
+        "context": context,
+    }
+    return data_dict
+
+
+def create_message_groups(messages: list) -> list[list]:
+    grouped_messages: list[list] = [
+        messages[message: message + MAILING_USERS_BATCH_SIZE]
+        for message in range(0, len(messages), MAILING_USERS_BATCH_SIZE)
+    ]
+    return grouped_messages
+
+
 def send_mail(
-    user: User,
-    subject: str,
-    template_string: str,
-    template_context: Union[
-        Dict,
-        List,
-    ] = None,
-    connection=None,
+        user: User,
+        subject: str,
+        template_string: str,
+        template_context: Union[
+            Dict,
+            List,
+        ] = None,
+        connection=None,
 ):
     return send_mass_mail([user], subject, template_string, template_context, connection)
 
 
+def send_group_messages(messages: list) -> int:
+    connection = mail.get_connection()
+    num_sent = connection.send_messages(messages)
+    connection.close()
+    return num_sent
+
+
 def send_mass_mail(
-    users: django.db.models.QuerySet | List[User],
-    subject: str,
-    template_string: str,
-    template_context: Union[
-        Dict,
-        List,
-    ] = None,
-    connection=None,
-) -> None:
+        users: django.db.models.QuerySet | List[User],
+        subject: str,
+        template_string: str,
+        template_context: Union[
+            Dict,
+            List,
+        ] = None,
+        connection=None,
+) -> int:
     """
     Begin mailing to specified users, sending rendered template with template_text arg.
     Throws an error if template render is unsuccessful.
@@ -45,16 +83,18 @@ def send_mass_mail(
     if template_context is None:
         template_context = {}
 
-    connection = connection or mail.get_connection()
     template = Template(template_string)
     messages = []
     for user in users:
         template_context["user"] = user
         html_msg = template.render(Context(template_context))
         plain_msg = template.render(Context(template_context))
-        msg = EmailMultiAlternatives(
-            subject, plain_msg, None, [user.email], connection=connection
-        )
+        msg = EmailMultiAlternatives(subject, plain_msg, None, [user.email])
         msg.attach_alternative(html_msg, "text/html")
         messages.append(msg)
-    return connection.send_messages(messages)
+
+    grouped_messages = create_message_groups(messages)
+    num_sent: int = 0
+    for group in grouped_messages:
+        num_sent += send_group_messages(group)
+    return num_sent

mailing/views.py

@@ -6,21 +6,18 @@
 from users.models import CustomUser
 from .utils import send_mass_mail
 from .models import MailingSchema
+from .utils import prepare_mail_data
 
 
 class SendMailView(APIView):
-    def post(self, request):
-        users = request.POST.getlist("users[]")
-        schema_id = request.POST["schemas"]
-        subject = request.POST["subject"]
-        mail_schema = MailingSchema.objects.get(pk=schema_id)
-        context = {}
-        for variable_name in mail_schema.schema:
-            key_in_post = "field-" + variable_name
-            if key_in_post in request.POST:
-                context[variable_name] = request.POST[key_in_post]
-        users_to_send = CustomUser.objects.filter(pk__in=users)
-        send_mass_mail(users_to_send, subject, mail_schema.template, context)
+    def post(self, request) -> JsonResponse:
+        mail_data: dict = prepare_mail_data(request.POST)
+        send_mass_mail(
+            mail_data["users_to_send"],
+            mail_data["subject"],
+            mail_data["mail_schema_template"],
+            mail_data["context"],
+        )
         return JsonResponse({"detail": "ok"})
 
 

mypy.ini

@@ -0,0 +1,7 @@
+[mypy]
+plugins =
+    mypy_django_plugin.main
+    mypy_drf_plugin.main
+
+[mypy.plugins.django-stubs]
+django_settings_module = "procollab.settings"