0002-api-endpoint-to-get-pricing-data-for-a-list-of-event.patch

Lauréline Guérin, 25 juillet 2022 09:35

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

Subject: [PATCH 2/2] api: endpoint to get pricing data for a list of events
 (#66354)

 lingo/agendas/chrono.py      |  23 ++-
 lingo/api/serializers.py     | 146 ++++++++++++++++
 lingo/api/urls.py            |   5 +
 lingo/api/views.py           |  20 ++-
 lingo/settings.py            |   1 +
 tests/agendas/test_chrono.py |  61 ++++++-
 tests/api/conftest.py        |  14 ++
 tests/api/test_pricing.py    | 314 +++++++++++++++++++++++++++++++++++
 8 files changed, 576 insertions(+), 8 deletions(-)
 create mode 100644 lingo/api/serializers.py
 create mode 100644 tests/api/conftest.py
 create mode 100644 tests/api/test_pricing.py

         return list(settings.KNOWN_SERVICES.get('chrono').values())[0]
     def get_chrono_json(path, log_errors=True):
     def get_chrono_json(path, params=None, log_errors=True):
         chrono_site = get_chrono_service()
         if chrono_site is None:
             return
         try:
             response = requests.get(
                 path,
                 params=params or {},
                 remote_service=chrono_site,
                 without_user=True,
                 headers={'accept': 'application/json'},
-...
     def get_event(event_slug):
         result = get_chrono_json('api/agendas/events/?slots=%s' % event_slug)
         return get_events(
             [event_slug],
             error_message=_('Unable to get event details'),
             error_message_with_details=_('Unable to get event details (%s)'),
         )[0]
     def get_events(event_slugs, error_message=None, error_message_with_details=None):
         error_message = error_message or _('Unable to get events details')
         error_message_with_details = error_message_with_details or _('Unable to get events details (%s)')
         result = get_chrono_json('api/agendas/events/', params={'slots': event_slugs})
         if not result:
             raise ChronoError(_('Unable to get event details'))
             raise ChronoError(error_message)
         if result.get('err'):
             raise ChronoError(_('Unable to get event details (%s)') % result['err_desc'])
             raise ChronoError(error_message_with_details % result['err_desc'])
         if not result.get('data'):
             raise ChronoError(_('Unable to get event details'))
         return result['data'][0]
             raise ChronoError(error_message)
         return result['data']
     def get_subscriptions(agenda_slug, user_external_id):

     # lingo - payment and billing system
     # Copyright (C) 2022  Entr'ouvert
+    #
     # This program is free software: you can redistribute it and/or modify it
     # under the terms of the GNU Affero General Public License as published
     # by the Free Software Foundation, either version 3 of the License, or
     # (at your option) any later version.
+    #
     # 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/>.
     import datetime
     from django.utils.translation import ugettext_lazy as _
     from rest_framework import serializers
     from rest_framework.exceptions import ValidationError
     from lingo.agendas.chrono import ChronoError, get_events, get_subscriptions
     from lingo.agendas.models import Agenda
     from lingo.pricing.models import AgendaPricing, PricingError
     class CommaSeparatedStringField(serializers.ListField):
         def get_value(self, dictionary):
             return super(serializers.ListField, self).get_value(dictionary)
         def to_internal_value(self, data):
             data = [s.strip() for s in data.split(',') if s.strip()]
             return super().to_internal_value(data)
     class PricingComputeSerializer(serializers.Serializer):
         slots = CommaSeparatedStringField(
             required=True, child=serializers.CharField(max_length=160, allow_blank=False)
+        )
         user_external_id = serializers.CharField(required=True, max_length=250)
         adult_external_id = serializers.CharField(required=True, max_length=250)
         agenda_slugs = []
         agendas = {}
         serialized_events = {}
         event_subscriptions = {}
         def validate_slots(self, value):
             self.agendas = {a.slug: a for a in Agenda.objects.all()}
             allowed_agenda_slugs = self.agendas.keys()
             agenda_slugs = set()
             for slot in value:
                 try:
                     agenda_slug, event_slug = slot.split('@')
                 except ValueError:
                     raise ValidationError(_('Invalid format for slot %s') % slot)
                 if not agenda_slug:
                     raise ValidationError(_('Missing agenda slug in slot %s') % slot)
                 if not event_slug:
                     raise ValidationError(_('Missing event slug in slot %s') % slot)
                 agenda_slugs.add(agenda_slug)
             extra_agendas = agenda_slugs - set(allowed_agenda_slugs)
             if extra_agendas:
                 extra_agendas = ', '.join(sorted(extra_agendas))
                 raise ValidationError(_('Unknown agendas: %s') % extra_agendas)
             self.agenda_slugs = sorted(agenda_slugs)
             try:
                 serialized_events = get_events(value)
             except ChronoError as e:
                 raise ValidationError(e)
             else:
                 for serialized_event in serialized_events:
                     event_slug = '%s@%s' % (serialized_event['agenda'], serialized_event['slug'])
                     self.serialized_events[event_slug] = serialized_event
             return value
         def get_subscriptions(self, user_external_id):
             agenda_subscriptions = {}
             for agenda_slug in self.agenda_slugs:
                 try:
                     agenda_subscriptions[agenda_slug] = get_subscriptions(agenda_slug, user_external_id)
                 except ChronoError as e:
                     raise ValidationError({'user_external_id': e})
             self.event_subscriptions = {}
             for event_slug, serialized_event in self.serialized_events.items():
                 start_date = datetime.datetime.fromisoformat(serialized_event['start_datetime']).date()
                 end_date = start_date + datetime.timedelta(days=1)
                 agenda_slug = serialized_event['agenda']
                 event_subscription = None
                 for subscription in agenda_subscriptions[agenda_slug]:
                     sub_start_date = datetime.date.fromisoformat(subscription['date_start'])
                     sub_end_date = datetime.date.fromisoformat(subscription['date_end'])
                     if sub_start_date >= end_date:
                         continue
                     if sub_end_date <= start_date:
                         continue
                     event_subscription = subscription
                     break
                 if event_subscription is None:
                     raise ValidationError(
                         {'user_external_id': _('No subscription found for event %s') % event_slug}
+                    )
                 self.event_subscriptions[event_slug] = event_subscription
         def validate(self, attrs):
             super().validate(attrs)
             if attrs.get('user_external_id') and self.serialized_events:
                 self.get_subscriptions(attrs['user_external_id'])
             return attrs
         def compute(self, request):
             if not self.serialized_events or not self.event_subscriptions:
                 return
             result = []
             event_slugs = sorted(self.serialized_events.keys())
             for event_slug in event_slugs:
                 serialized_event = self.serialized_events[event_slug]
                 try:
                     pricing_data = AgendaPricing.get_pricing_data(
                         request=request,
                         agenda=self.agendas[serialized_event['agenda']],
                         event=serialized_event,
                         subscription=self.event_subscriptions[event_slug],
                         check_status={
                             'status': 'presence',
                             'check_type': None,
                         },
                         booking={},
                         user_external_id=self.validated_data['user_external_id'],
                         adult_external_id=self.validated_data['adult_external_id'],
+                    )
                     result.append(
+                        {
                             'event': event_slug,
                             'pricing_data': pricing_data,
+                        }
+                    )
                 except PricingError as e:
                     result.append({'event': event_slug, 'error': e.details})
             result = sorted(result, key=lambda d: d['event'])
             return result

             views.agenda_check_type_list,
             name='api-agenda-check-types',
         ),
         url(
             r'^pricing/compute/$',
             views.pricing_compute,
             name='api-pricing-compute',
         ),
+    ]

     # along with this program.  If not, see <http://www.gnu.org/licenses/>.
     from django.shortcuts import get_object_or_404
     from django.utils.translation import gettext_noop as N_
     from rest_framework import permissions
     from rest_framework.views import APIView
     from lingo.agendas.models import Agenda
     from lingo.api.utils import Response
     from lingo.api import serializers
     from lingo.api.utils import APIErrorBadRequest, Response
     class AgendaCheckTypeList(APIView):
-...
     agenda_check_type_list = AgendaCheckTypeList.as_view()
     class PricingCompute(APIView):
         permission_classes = (permissions.IsAuthenticated,)
         serializer_class = serializers.PricingComputeSerializer
         def get(self, request, format=None):
             serializer = self.serializer_class(data=request.query_params)
             if not serializer.is_valid():
                 raise APIErrorBadRequest(N_('invalid payload'), errors=serializer.errors)
             return Response({'data': serializer.compute(self.request)})
     pricing_compute = PricingCompute.as_view()

lingo/settings.py
54	54	'django.contrib.humanize',
55	55	'eopayment',
56	56	'gadjo',
	57	'rest_framework',
57	58	'lingo.agendas',
58	59	'lingo.api',
59	60	'lingo.manager',

         ChronoError,
         collect_agenda_data,
         get_event,
         get_events,
         get_subscriptions,
         refresh_agendas,
+    )
-...
             with pytest.raises(ChronoError) as e:
                 get_event('foo')
             assert str(e.value) == 'Unable to get event details'
             assert requests_get.call_args_list[0][0] == ('api/agendas/events/?slots=foo',)
             assert requests_get.call_args_list[0][0] == ('api/agendas/events/',)
             assert requests_get.call_args_list[0][1]['params']['slots'] == ['foo']
             assert requests_get.call_args_list[0][1]['remote_service']['url'] == 'http://chrono.example.org'
         data = {'data': ['foo']}
-...
             assert get_event('foo') == 'foo'
     def test_get_events_no_service(settings):
         settings.KNOWN_SERVICES = {}
         with pytest.raises(ChronoError) as e:
             get_events(['foo', 'bar'])
         assert str(e.value) == 'Unable to get events details'
         settings.KNOWN_SERVICES = {'other': []}
         with pytest.raises(ChronoError) as e:
             get_events(['foo', 'bar'])
         assert str(e.value) == 'Unable to get events details'
     def test_get_events():
         with mock.patch('requests.Session.get') as requests_get:
             requests_get.side_effect = ConnectionError()
             with pytest.raises(ChronoError) as e:
                 get_events(['foo', 'bar'])
             assert str(e.value) == 'Unable to get events details'
         with mock.patch('requests.Session.get') as requests_get:
             mock_resp = Response()
             mock_resp.status_code = 500
             requests_get.return_value = mock_resp
             with pytest.raises(ChronoError) as e:
                 get_events(['foo', 'bar'])
             assert str(e.value) == 'Unable to get events details'
         with mock.patch('requests.Session.get') as requests_get:
             mock_resp = Response()
             mock_resp.status_code = 404
             requests_get.return_value = mock_resp
             with pytest.raises(ChronoError) as e:
                 get_events(['foo', 'bar'])
             assert str(e.value) == 'Unable to get events details'
         with mock.patch('requests.Session.get') as requests_get:
             requests_get.return_value = MockedRequestResponse(content=json.dumps({'foo': 'bar'}))
             with pytest.raises(ChronoError) as e:
                 get_events(['foo', 'bar'])
             assert str(e.value) == 'Unable to get events details'
         data = {'data': []}
         with mock.patch('requests.Session.get') as requests_get:
             requests_get.return_value = MockedRequestResponse(content=json.dumps(data))
             with pytest.raises(ChronoError) as e:
                 get_events(['foo', 'bar'])
             assert str(e.value) == 'Unable to get events details'
             assert requests_get.call_args_list[0][0] == ('api/agendas/events/',)
             assert requests_get.call_args_list[0][1]['params']['slots'] == ['foo', 'bar']
             assert requests_get.call_args_list[0][1]['remote_service']['url'] == 'http://chrono.example.org'
         data = {'data': ['foo', 'bar']}
         with mock.patch('requests.Session.get') as requests_get:
             requests_get.return_value = MockedRequestResponse(content=json.dumps(data))
             assert get_events(['foo', 'bar']) == ['foo', 'bar']
     def test_get_subscriptions_no_service(settings):
         settings.KNOWN_SERVICES = {}
         with pytest.raises(ChronoError) as e:

     import pytest
     from django.contrib.auth import get_user_model
     User = get_user_model()
     @pytest.fixture
     def user():
         user = User.objects.create(
             username='john.doe', first_name='John', last_name='Doe', email='john.doe@example.net'
+        )
         user.set_password('password')
         user.save()
         return user

     from unittest import mock
     import pytest
     from lingo.agendas.chrono import ChronoError
     from lingo.agendas.models import Agenda
     from lingo.pricing.models import PricingError
     pytestmark = pytest.mark.django_db
     def test_pricing_compute_params(app, user):
         app.authorization = ('Basic', ('john.doe', 'password'))
         # missing slots
         resp = app.get(
             '/api/pricing/compute/',
             params={'user_external_id': 'user:1', 'adult_external_id': 'adult:1'},
             status=400,
+        )
         assert resp.json['err'] == 1
         assert resp.json['err_desc'] == 'invalid payload'
         assert resp.json['errors']['slots'] == ['This field is required.']
         # missing user_external_id
         resp = app.get(
             '/api/pricing/compute/',
             params={'slots': 'foo@foo', 'adult_external_id': 'adult:1'},
             status=400,
+        )
         assert resp.json['err'] == 1
         assert resp.json['err_desc'] == 'invalid payload'
         assert resp.json['errors']['user_external_id'] == ['This field is required.']
         # missing adult_external_id
         resp = app.get(
             '/api/pricing/compute/',
             params={'slots': 'foo@foo', 'user_external_id': 'user:1'},
             status=400,
+        )
         assert resp.json['err'] == 1
         assert resp.json['err_desc'] == 'invalid payload'
         assert resp.json['errors']['adult_external_id'] == ['This field is required.']
     def test_pricing_compute_slots(app, user):
         app.authorization = ('Basic', ('john.doe', 'password'))
         # bad slot format
         resp = app.get(
             '/api/pricing/compute/',
             params={
                 'slots': 'event-bar-slug',
                 'user_external_id': 'user:1',
                 'adult_external_id': 'adult:1',
             },
             status=400,
+        )
         assert resp.json['err'] == 1
         assert resp.json['err_desc'] == 'invalid payload'
         assert resp.json['errors']['slots'] == ['Invalid format for slot event-bar-slug']
         resp = app.get(
             '/api/pricing/compute/',
             params={
                 'slots': '@event-bar-slug',
                 'user_external_id': 'user:1',
                 'adult_external_id': 'adult:1',
             },
             status=400,
+        )
         assert resp.json['err'] == 1
         assert resp.json['err_desc'] == 'invalid payload'
         assert resp.json['errors']['slots'] == ['Missing agenda slug in slot @event-bar-slug']
         resp = app.get(
             '/api/pricing/compute/',
             params={
                 'slots': 'agenda@',
                 'user_external_id': 'user:1',
                 'adult_external_id': 'adult:1',
             },
             status=400,
+        )
         assert resp.json['err'] == 1
         assert resp.json['err_desc'] == 'invalid payload'
         assert resp.json['errors']['slots'] == ['Missing event slug in slot agenda@']
         # unknown agenda
         resp = app.get(
             '/api/pricing/compute/',
             params={
                 'slots': 'agenda@event-bar-slug, agenda2@event-bar-slug',
                 'user_external_id': 'user:1',
                 'adult_external_id': 'adult:1',
             },
             status=400,
+        )
         assert resp.json['err'] == 1
         assert resp.json['err_desc'] == 'invalid payload'
         assert resp.json['errors']['slots'] == ['Unknown agendas: agenda, agenda2']
     @mock.patch('lingo.api.serializers.get_events')
     def test_pricing_compute_events_error(mock_events, app, user):
         Agenda.objects.create(label='Agenda')
         app.authorization = ('Basic', ('john.doe', 'password'))
         mock_events.side_effect = ChronoError('foo bar')
         resp = app.get(
             '/api/pricing/compute/',
             params={
                 'slots': 'agenda@event-bar-slug',
                 'user_external_id': 'user:1',
                 'adult_external_id': 'adult:1',
             },
             status=400,
+        )
         assert resp.json['err'] == 1
         assert resp.json['err_desc'] == 'invalid payload'
         assert resp.json['errors']['slots'] == ['foo bar']
     @mock.patch('lingo.api.serializers.get_events')
     @mock.patch('lingo.api.serializers.get_subscriptions')
     def test_pricing_compute_subscriptions_error(mock_subscriptions, mock_events, app, user):
         Agenda.objects.create(label='Agenda')
         app.authorization = ('Basic', ('john.doe', 'password'))
         mock_events.return_value = [
             {'start_datetime': '2021-09-01T12:00:00+02:00', 'agenda': 'agenda', 'slug': 'event-bar-slug'}
+        ]
         mock_subscriptions.side_effect = ChronoError('foo baz')
         resp = app.get(
             '/api/pricing/compute/',
             params={
                 'slots': 'agenda@event-bar-slug',
                 'user_external_id': 'user:1',
                 'adult_external_id': 'adult:1',
             },
             status=400,
+        )
         assert resp.json['err'] == 1
         assert resp.json['err_desc'] == 'invalid payload'
         assert resp.json['errors']['user_external_id'] == ['foo baz']
     @mock.patch('lingo.api.serializers.get_events')
     @mock.patch('lingo.api.serializers.get_subscriptions')
     @mock.patch('lingo.pricing.models.AgendaPricing.get_pricing_data')
     def test_pricing_compute(mock_pricing_data, mock_subscriptions, mock_events, app, user):
         agenda = Agenda.objects.create(label='Agenda')
         agenda2 = Agenda.objects.create(label='Agenda2')
         app.authorization = ('Basic', ('john.doe', 'password'))
         mock_events.return_value = [
             {'start_datetime': '2021-09-01T12:00:00+02:00', 'agenda': 'agenda', 'slug': 'event-bar-slug'}
+        ]
         mock_subscriptions.return_value = []
         resp = app.get(
             '/api/pricing/compute/',
             params={
                 'slots': 'agenda@event-bar-slug',
                 'user_external_id': 'user:1',
                 'adult_external_id': 'adult:1',
             },
             status=400,
+        )
         assert resp.json['err'] == 1
         assert resp.json['err_desc'] == 'invalid payload'
         assert resp.json['errors']['user_external_id'] == [
             'No subscription found for event agenda@event-bar-slug'
+        ]
         assert mock_pricing_data.call_args_list == []
         assert mock_subscriptions.call_args_list == [mock.call('agenda', 'user:1')]
         mock_subscriptions.reset_mock()
         mock_subscriptions.return_value = [
+            {
                 'date_start': '2021-08-01',
                 'date_end': '2021-09-01',
             },
+            {
                 'date_start': '2021-09-02',
                 'date_end': '2021-09-03',
             },
+        ]
         resp = app.get(
             '/api/pricing/compute/',
             params={
                 'slots': 'agenda@event-bar-slug',
                 'user_external_id': 'user:1',
                 'adult_external_id': 'adult:1',
             },
             status=400,
+        )
         assert resp.json['err'] == 1
         assert resp.json['err_desc'] == 'invalid payload'
         assert resp.json['errors']['user_external_id'] == [
             'No subscription found for event agenda@event-bar-slug'
+        ]
         assert mock_pricing_data.call_args_list == []
         assert mock_subscriptions.call_args_list == [mock.call('agenda', 'user:1')]
         mock_events.return_value = [
             {'start_datetime': '2021-09-02T12:00:00+02:00', 'agenda': 'agenda', 'slug': 'event-bar-slug'},
             {'start_datetime': '2021-09-01T12:00:00+02:00', 'agenda': 'agenda2', 'slug': 'event-baz-slug'},
+        ]
         mock_subscriptions.reset_mock()
         mock_subscriptions.return_value = [
+            {
                 'date_start': '2021-08-01',
                 'date_end': '2021-09-01',
             },
+            {
                 'date_start': '2021-09-02',
                 'date_end': '2021-09-03',
             },
+        ]
         resp = app.get(
             '/api/pricing/compute/',
             params={
                 'slots': 'agenda@event-bar-slug, agenda2@event-baz-slug',
                 'user_external_id': 'user:1',
                 'adult_external_id': 'adult:1',
             },
             status=400,
+        )
         assert resp.json['err'] == 1
         assert resp.json['err_desc'] == 'invalid payload'
         assert resp.json['errors']['user_external_id'] == [
             'No subscription found for event agenda2@event-baz-slug'
+        ]
         assert mock_pricing_data.call_args_list == []
         assert mock_subscriptions.call_args_list == [
             mock.call('agenda', 'user:1'),
             mock.call('agenda2', 'user:1'),
+        ]
         mock_subscriptions.return_value = [
+            {
                 'date_start': '2021-08-01',
                 'date_end': '2021-09-01',
             },
+            {
                 'date_start': '2021-09-01',
                 'date_end': '2021-09-02',
             },
+            {
                 'date_start': '2021-09-02',
                 'date_end': '2021-09-03',
             },
+        ]
         mock_pricing_data.side_effect = [
             {'foo': 'baz'},
             {'foo': 'bar'},
+        ]
         resp = app.get(
             '/api/pricing/compute/',
             params={
                 'slots': 'agenda@event-bar-slug, agenda2@event-baz-slug',
                 'user_external_id': 'user:1',
                 'adult_external_id': 'adult:1',
             },
+        )
         assert resp.json['data'] == [
             {'event': 'agenda2@event-baz-slug', 'pricing_data': {'foo': 'baz'}},
             {'event': 'agenda@event-bar-slug', 'pricing_data': {'foo': 'bar'}},
+        ]
         assert mock_pricing_data.call_args_list == [
             mock.call(
                 request=mock.ANY,
                 agenda=agenda2,
                 event={
                     'start_datetime': '2021-09-01T12:00:00+02:00',
                     'agenda': 'agenda2',
                     'slug': 'event-baz-slug',
                 },
                 subscription={'date_start': '2021-09-01', 'date_end': '2021-09-02'},
                 check_status={'status': 'presence', 'check_type': None},
                 booking={},
                 user_external_id='user:1',
                 adult_external_id='adult:1',
             ),
             mock.call(
                 request=mock.ANY,
                 agenda=agenda,
                 event={
                     'start_datetime': '2021-09-02T12:00:00+02:00',
                     'agenda': 'agenda',
                     'slug': 'event-bar-slug',
                 },
                 subscription={'date_start': '2021-09-02', 'date_end': '2021-09-03'},
                 check_status={'status': 'presence', 'check_type': None},
                 booking={},
                 user_external_id='user:1',
                 adult_external_id='adult:1',
             ),
+        ]
         mock_pricing_data.side_effect = [
             PricingError(details={'foo': 'error'}),
             {'foo': 'bar'},
+        ]
         resp = app.get(
             '/api/pricing/compute/',
             params={
                 'slots': 'agenda@event-bar-slug, agenda2@event-baz-slug',
                 'user_external_id': 'user:1',
                 'adult_external_id': 'adult:1',
             },
+        )
         assert resp.json['data'] == [
             {'event': 'agenda2@event-baz-slug', 'error': {'foo': 'error'}},
             {'event': 'agenda@event-bar-slug', 'pricing_data': {'foo': 'bar'}},
+        ]
+    -

Projet

Général

Profil

Produits Entr'ouvert » Lingo

0002-api-endpoint-to-get-pricing-data-for-a-list-of-event.patch