0002-notification-send-notification-to-several-users-6064.patch

Nicolas Roche, 18 janvier 2022 22:46

Voir les différences: en ligne côte à côte

Subject: [PATCH 2/2] notification: send notification to several users (#60643)

 combo/apps/notifications/api_views.py | 52 ++++++++++++++++------
 tests/test_notification.py            | 64 +++++++++++++++++++++++++++
 2 files changed, 103 insertions(+), 13 deletions(-)

     # This program is distributed in the hope that it will be useful,
     # but WITHOUT ANY WARRANTY; without even the implied warranty of
     # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
     # GNU Affero General Public License for more details.
+    #
     # You should have received a copy of the GNU Affero General Public License
     # along with this program.  If not, see <http://www.gnu.org/licenses/>.
     from django.contrib.auth.models import User
     from django.db import transaction
     from django.utils.encoding import force_text
     from django.utils.translation import ugettext_lazy as _
     from rest_framework import authentication, permissions, serializers, status
     from rest_framework.generics import GenericAPIView
     from rest_framework.response import Response
     from combo.profile.utils import get_user_from_name_id
     from .models import Notification
     class NotificationSerializer(serializers.Serializer):
         summary = serializers.CharField(required=True, allow_blank=False, max_length=140)
         id = serializers.CharField(required=False, allow_null=True)
         body = serializers.CharField(allow_blank=False, default='')
         url = serializers.URLField(allow_blank=True, default='')
         origin = serializers.CharField(allow_blank=True, default='')
         start_timestamp = serializers.DateTimeField(required=False, allow_null=True)
         end_timestamp = serializers.DateTimeField(required=False, allow_null=True)
         duration = serializers.IntegerField(required=False, allow_null=True, min_value=0)
         name_ids = serializers.ListField(
             required=False, child=serializers.CharField(required=True, max_length=150)
+        )
         def validate(self, attrs):
             super().validate(attrs)
             users = []
             for uuid in attrs.get('name_ids') or []:
                 try:
                     users.append(get_user_from_name_id(uuid, raise_on_missing=True))
                 except User.DoesNotExist:
                     raise serializers.ValidationError({'users': _('"%s" user does not exist.') % uuid})
             attrs['users'] = users
             return attrs
     class Add(GenericAPIView):
         permission_classes = (permissions.IsAuthenticated,)
         serializer_class = NotificationSerializer
         def post(self, request, *args, **kwargs):
             serializer = self.get_serializer(data=request.data)
             if not serializer.is_valid():
                 response = {'err': 1, 'err_desc': serializer.errors}
                 return Response(response, status.HTTP_400_BAD_REQUEST)
             data = serializer.validated_data
             notification_ids = []
             try:
                 notification = Notification.notify(
                     user=request.user,
                     summary=data['summary'],
                     id=data.get('id'),
                     body=data.get('body'),
                     url=data.get('url'),
                     origin=data.get('origin'),
                     start_timestamp=data.get('start_timestamp'),
                     end_timestamp=data.get('end_timestamp'),
                     duration=data.get('duration'),
+                )
                 with transaction.atomic():
                     for user in data.get('users') or [request.user]:
                         notification = Notification.notify(
                             user=user,
                             summary=data['summary'],
                             id=data.get('id'),
                             body=data.get('body'),
                             url=data.get('url'),
                             origin=data.get('origin'),
                             start_timestamp=data.get('start_timestamp'),
                             end_timestamp=data.get('end_timestamp'),
                             duration=data.get('duration'),
+                        )
                         notification_ids.append(notification.public_id)
             except ValueError as e:
                 response = {'err': 1, 'err_desc': {'id': [force_text(e)]}}
                 return Response(response, status.HTTP_400_BAD_REQUEST)
             uniq_ids = set(notification_ids)
             if len(uniq_ids) == 1:
                 response = {'err': 0, 'data': {'nb_notify': len(notification_ids), 'id': uniq_ids.pop()}}
             else:
                 response = {'err': 0, 'data': {'id': notification.public_id}}
                 return Response(response)
                 response = {'err': 0, 'data': {'nb_notify': len(notification_ids), 'ids': notification_ids}}
             return Response(response)
     add = Add.as_view()
     class Ack(GenericAPIView):
         authentication_classes = (authentication.SessionAuthentication,)
         permission_classes = (permissions.IsAuthenticated,)

     import pytest
     from django.contrib.auth.models import User
     from django.test import Client
     from django.test.client import RequestFactory
     from django.urls import reverse
     from django.utils.encoding import force_text
     from django.utils.timezone import now
     from mellon.models import UserSAMLIdentifier
     from combo.apps.lingo.models import ActiveItems, PaymentBackend, Regie
     from combo.apps.notifications.models import Notification, NotificationsCell
     from combo.data.models import Page
     from .test_manager import login as login_app
     pytestmark = pytest.mark.django_db
-...
     def test_notification_ws(john_doe):
         def notify(data, count, check_id=None):
             resp = client.post(reverse('api-notification-add'), json.dumps(data), content_type='application/json')
             assert resp.status_code == 200
             result = json.loads(force_text(resp.content))
             assert result['err'] == 0
             assert result['data']['nb_notify'] == 1
             assert 'id' in result['data']
             if 'id' in data:
                 assert check_id is not None and result['data']['id'] == check_id
             assert Notification.objects.filter(user=john_doe).count() == count
             return Notification.objects.filter(user=john_doe).order_by('id').last()
         login(john_doe)
         notify({'summary': 'foo'}, 1)
         notify({'summary': 'bar'}, 2)
-...
         Notification.objects.visible(john_doe).ack()
         # acking a notification without and end_timestamp, still visible
         freezer.move_to(start + timedelta(days=365, seconds=1))
         content = cell.render(context)
         assert Notification.objects.visible(john_doe).count() == 1
         assert 'notibar' in content
         assert 'notifoo' not in content
     def test_notification_ws_create_many(john_doe, jane_doe):
         UserSAMLIdentifier.objects.create(user=john_doe, name_id='00a')
         UserSAMLIdentifier.objects.create(user=jane_doe, name_id='00b')
         login(john_doe)
         data = {'summary': 'foo', 'name_ids': ['00a', '00b']}
         resp = client.post(reverse('api-notification-add'), json.dumps(data), content_type='application/json')
         assert resp.status_code == 200
         result = resp.json()
         assert result['err'] == 0
         assert result['data']['nb_notify'] == 2
         assert len(result['data']['ids']) == 2
         assert Notification.objects.count() == 2
         notif = Notification.objects.filter(user=john_doe).last()
         assert notif.public_id == result['data']['ids'][0]
         assert notif.summary == 'foo'
         notif = Notification.objects.filter(user=jane_doe).last()
         assert notif.public_id == result['data']['ids'][1]
         assert notif.summary == 'foo'
         # wrong username
         data = {'summary': 'foo', 'name_ids': ['00a', 'unknown']}
         resp = client.post(reverse('api-notification-add'), json.dumps(data), content_type='application/json')
         assert resp.status_code == 400
         result = resp.json()
         assert result['err_desc'] == {'users': ['"unknown" user does not exist.']}
         assert Notification.objects.count() == 2
     def test_notification_ws_update_many(john_doe, jane_doe):
         UserSAMLIdentifier.objects.create(user=john_doe, name_id='00a')
         UserSAMLIdentifier.objects.create(user=jane_doe, name_id='00b')
         login(john_doe)
         data = {
             'summary': 'foo',
             'name_ids': ['00a', '00b'],
             'id': 'test:42',
+        }
         resp = client.post(reverse('api-notification-add'), json.dumps(data), content_type='application/json')
         assert resp.status_code == 200
         result = resp.json()
         assert result['err'] == 0
         assert result['data']['nb_notify'] == 2
         assert result['data']['id'] == 'test:42'
         assert Notification.objects.count() == 2
         assert Notification.objects.get(user=john_doe, external_id='test:42').summary == 'foo'
         assert Notification.objects.get(user=jane_doe, external_id='test:42').summary == 'foo'
         data['summary'] = 'bar'
         resp = client.post(reverse('api-notification-add'), json.dumps(data), content_type='application/json')
         assert resp.status_code == 200
         result = resp.json()
         assert result['err'] == 0
         assert result['data']['nb_notify'] == 2
         assert result['data']['id'] == 'test:42'
         assert Notification.objects.count() == 2
         assert Notification.objects.get(user=john_doe, external_id='test:42').summary == 'bar'
         assert Notification.objects.get(user=jane_doe, external_id='test:42').summary == 'bar'
+    -

Projet

Général

Profil

Produits Entr'ouvert » Combo

0002-notification-send-notification-to-several-users-6064.patch