Produits Entr'ouvert » Chrono

    url(

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

        views.events_fillslots,

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

    ),

class EventsSlotsSerializer(SlotsSerializer):

    serializer_class = EventsSlotsSerializer

class EventsFillslots(APIView):

    permission_classes = (permissions.IsAuthenticated,)

    serializer_class = EventsSlotsSerializer

    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(),

            )

        )

        waiting_list_events = [event for event in events if event.in_waiting_list]

        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),

            'waiting_list_events': [get_event_detail(request, x, agenda=agenda) for x in waiting_list_events],

        }

        return Response(response)

events_fillslots = EventsFillslots.as_view()

def test_api_events_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/events/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

    assert len(resp.json['waiting_list_events']) == 0

    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

    assert len(resp.json['waiting_list_events']) == 0

    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 resp.json['waiting_list_events'][0]['slug'] == event.slug

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