Produits Entr'ouvert » Chrono

    url(

        r'^agenda/(?P<agenda_identifier>[\w-]+)/event_fillslots/$',

        views.event_fillslots,

        name='api-agenda-event-fillslots',

    ),

class EventFillslots(APIView):

    permission_classes = (permissions.IsAuthenticated,)

    serializer_class = EventSlotsSerializer

    def post(self, request, agenda_identifier):

        agenda = get_object_or_404(Agenda, slug=agenda_identifier, kind='events')

        serializer = self.serializer_class(data=request.data, partial=True)

        if not serializer.is_valid():

            raise APIError(

                _('invalid payload'),

                err_class='invalid payload',

                errors=serializer.errors,

                http_status=status.HTTP_400_BAD_REQUEST,

            )

        payload = serializer.validated_data

        events = get_events_from_slots(payload['slots'], request, agenda, payload)

        events = Event.annotate_queryset(events)

        full_events = [str(event) for event in events.filter(full=True)]

        if full_events:

            raise APIError(

                _('some events are full: %s') % ', '.join(full_events), err_class='some events are full'

            )

        events = events.annotate(

            in_waiting_list=ExpressionWrapper(

                Q(booked_places_count__gte=F('places')) | Q(waiting_list_count__gt=0),

                output_field=BooleanField(),

            )

        )

        extra_data = {k: v for k, v in request.data.items() if k not in payload}

        bookings = [make_booking(event, payload, extra_data) for event in events]

        with transaction.atomic():

            Booking.objects.bulk_create(bookings)

            events.update(

                full=Q(booked_places_count__gte=F('places'), waiting_list_places=0)

                | Q(waiting_list_places__gt=0, waiting_list_count__gte=F('waiting_list_places')),

                almost_full=Q(booked_places_count__gte=0.9 * F('places')),

            )

        response = {

            'err': 0,

            'booking_count': len(bookings),

        }

        return Response(response)

event_fillslots = EventFillslots.as_view()

def test_api_event_fillslots(app, user, freezer):

    freezer.move_to('2021-09-06 12:00')

    agenda = Agenda.objects.create(label='Foo bar', kind='events')

    event = Event.objects.create(

        label='Event',

        start_datetime=now() + datetime.timedelta(days=1),

        places=2,

        waiting_list_places=1,

        agenda=agenda,

    )

    second_event = Event.objects.create(

        label='Event 2', start_datetime=now() + datetime.timedelta(days=2), places=2, agenda=agenda

    )

    app.authorization = ('Basic', ('john.doe', 'password'))

    fillslots_url = '/api/agenda/%s/event_fillslots/' % agenda.slug

    params = {'user_external_id': 'user_id', 'slots': 'event,event-2'}

    resp = app.post_json(fillslots_url, params=params)

    assert resp.json['booking_count'] == 2

    events = Event.annotate_queryset(Event.objects.all())

    assert events.filter(booked_places_count=1).count() == 2

    resp = app.post_json(fillslots_url, params=params)

    assert resp.json['booking_count'] == 2

    resp = app.post_json(fillslots_url, params=params)

    assert resp.json['err'] == 1

    assert resp.json['err_desc'] == 'some events are full: Event 2'

    params['slots'] = 'event'

    resp = app.post_json(fillslots_url, params=params)

    assert resp.json['booking_count'] == 1

    assert Booking.objects.filter(in_waiting_list=True, event=event).count() == 1