entrouvert
/
chrono
14
1
Fork 0
Code Issues Pull Requests 4 Releases Wiki Activity

implement locking for event's agendas (#80489) 

* add code to clean event's agendas lease/bookings
* add annotation helper method annotate_queryset_for_lock_code() to
  compute corrects places statistics given a lock_code (excluding
  bookings linked to this lock_code)
* use annotate_queryset_for_lock_code() in Datetimes and
  MultipleAgendasDatetimes
* make event's fillslot method completely atomic and add mechanic for
  handling the lock code
* removed handling of IntegrityError which cannot happen for events
* lock_code is for now not supported with RecurringFillslots
This commit is contained in:
Benjamin Dauvergne 2023-08-31 11:41:04 +02:00
parent d6a5861876
commit eafa816253
7 changed files with 453 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
chrono/agendas/models.py
View File

@ -2113,6 +2113,12 @@ class Event(models.Model):
full_notification_timestamp = models.DateTimeField(null=True, blank=True)
cancelled_notification_timestamp = models.DateTimeField(null=True, blank=True)
# Store alternate version of booked_places and booked_waiting_list_places,
# when an Event queryset is annotated with
# Event.annotate_queryset_for_lock_code()
unlocked_booked_places = None
unlocked_booked_waiting_list_places = None
class Meta:
ordering = ['agenda', 'start_datetime', 'duration', 'label']
unique_together = ('agenda', 'slug')
@ -2342,6 +2348,26 @@ class Event(models.Model):
)
return qs
@staticmethod
def annotate_queryset_for_lock_code(qs, lock_code):
qs = qs.annotate(
unlocked_booked_places=Count(
'booking',
filter=Q(
booking__cancellation_datetime__isnull=True,
)
& ~Q(booking__lease__lock_code=lock_code),
),
unlocked_booked_waiting_list_places=Count(
'booking',
filter=Q(
booking__cancellation_datetime__isnull=False,
)
& ~Q(booking__lease__lock_code=lock_code),
),
)
return qs
@staticmethod
def annotate_queryset_with_overlaps(qs, other_events=None):
if not other_events:
@ -2462,13 +2488,36 @@ class Event(models.Model):
notchecked_count=Coalesce(Subquery(notchecked_count, output_field=IntegerField()), Value(0)),
)
def get_booked_places(self):
if self.unlocked_booked_places is None:
return self.booked_places
else:
return self.unlocked_booked_places
def get_booked_waiting_list_places(self):
if self.unlocked_booked_waiting_list_places is None:
return self.booked_waiting_list_places
else:
return self.unlocked_booked_waiting_list_places
def get_full(self):
if self.agenda.partial_bookings:
return False
elif self.unlocked_booked_places is None:
return self.full
else:
if self.waiting_list_places == 0:
return self.get_booked_places() >= self.places
else:
return self.get_booked_waiting_list_places() >= self.waiting_list_places
@property
def remaining_places(self):
return max(0, self.places - self.booked_places)
return max(0, self.places - self.get_booked_places())
@property
def remaining_waiting_list_places(self):
return max(0, self.waiting_list_places - self.booked_waiting_list_places)
return max(0, self.waiting_list_places - self.get_booked_waiting_list_places())
@property
def end_datetime(self):
@ -4664,3 +4713,6 @@ class Lease(models.Model):
# Delete expired meeting's events, bookings and leases.'''
Event.objects.filter(agenda__kind='meetings', booking__lease__expiration_datetime__lt=now()).delete()
# Delete expired event's bookings and leases'''
Booking.objects.filter(event__agenda__kind='events', lease__expiration_datetime__lt=now()).delete()

1
chrono/api/serializers.py
View File

@ -451,6 +451,7 @@ class DatetimesSerializer(DateRangeSerializer):
events = serializers.CharField(required=False, max_length=32, allow_blank=True)
hide_disabled = serializers.BooleanField(default=False)
bypass_delays = serializers.BooleanField(default=False)
lock_code = serializers.CharField(max_length=64, required=False, allow_blank=False, trim_whitespace=True)
def validate(self, attrs):
super().validate(attrs)

194
chrono/api/views.py
View File

@ -129,17 +129,18 @@ def get_event_places(event):
available = event.remaining_places
places = {
'total': event.places,
'reserved': event.booked_places,
'reserved': event.get_booked_places(),
'available': available,
'full': event.full,
'full': event.get_full(),
'has_waiting_list': False,
}
if event.waiting_list_places:
places['has_waiting_list'] = True
places['waiting_list_total'] = event.waiting_list_places
places['waiting_list_reserved'] = event.booked_waiting_list_places
# use the alternate computation of booked waiting place if a lock_code is currently used
places['waiting_list_reserved'] = event.get_booked_waiting_list_places()
places['waiting_list_available'] = event.remaining_waiting_list_places
places['waiting_list_activated'] = event.booked_waiting_list_places > 0 or available <= 0
places['waiting_list_activated'] = event.get_booked_waiting_list_places() > 0 or available <= 0
# 'waiting_list_activated' means next booking will go into the waiting list
return places
@ -635,6 +636,7 @@ class Datetimes(APIView):
bookable_events = bookable_events_raw or 'future'
book_past = bookable_events in ['all', 'past']
book_future = bookable_events in ['all', 'future']
lock_code = payload.get('lock_code')
entries = Event.objects.none()
if book_past:
@ -649,6 +651,8 @@ class Datetimes(APIView):
bypass_delays=payload.get('bypass_delays'),
)
entries = Event.annotate_queryset_for_user(entries, user_external_id)
if lock_code:
entries = Event.annotate_queryset_for_lock_code(entries, lock_code=lock_code)
entries = entries.order_by('start_datetime', 'duration', 'label')
if payload['hide_disabled']:
@ -710,6 +714,7 @@ class MultipleAgendasDatetimes(APIView):
show_only_subscribed = bool('subscribed' in payload)
with_status = bool(payload.get('with_status'))
check_overlaps = bool(payload.get('check_overlaps'))
lock_code = payload.get('lock_code')
entries = Event.objects.none()
if agendas:
@ -726,6 +731,9 @@ class MultipleAgendasDatetimes(APIView):
show_out_of_minimal_delay=show_past_events,
)
entries = Event.annotate_queryset_for_user(entries, user_external_id, with_status=with_status)
if lock_code:
Event.annotate_queryset_for_lock_code(entries, lock_code)
if check_overlaps:
entries = Event.annotate_queryset_with_overlaps(entries)
if show_only_subscribed:
@ -1194,7 +1202,8 @@ class EventsAgendaFillslot(APIView):
def post(self, request, agenda, slot):
return self.fillslot(request=request, agenda=agenda, slot=slot)
def fillslot(self, request, agenda, slot, retry=False):
@transaction.atomic()
def fillslot(self, request, agenda, slot):
known_body_params = set(request.query_params).intersection(
{'label', 'user_name', 'backoffice_url', 'user_display_label'}
)
@ -1210,6 +1219,14 @@ class EventsAgendaFillslot(APIView):
raise APIErrorBadRequest(N_('invalid payload'), errors=serializer.errors)
payload = serializer.validated_data
lock_code = payload.get('lock_code')
if lock_code:
# cannot apply a default value on the serializer bcause it is used with partial=True
confirm_after_lock = payload.get('confirm_after_lock') or False
# delete events/bookings with the same lock code in the
# same agenda(s) to allow rebooking or confirming
Booking.objects.filter(event__agenda=agenda, lease__lock_code=lock_code).delete()
if 'count' in payload:
places_count = payload['count']
elif 'count' in request.query_params:
@ -1257,6 +1274,7 @@ class EventsAgendaFillslot(APIView):
# search free places. Switch to waiting list if necessary.
in_waiting_list = False
if event.start_datetime > now():
if payload.get('force_waiting_list') and not event.waiting_list_places:
raise APIError(N_('no waiting list'))
@ -1264,52 +1282,41 @@ class EventsAgendaFillslot(APIView):
if event.waiting_list_places:
if (
payload.get('force_waiting_list')
or (event.booked_places + places_count) > event.places
or event.booked_waiting_list_places
or (event.get_booked_places() + places_count) > event.places
or event.get_booked_waiting_list_places()
):
# if this is full or there are people waiting, put new bookings
# in the waiting list.
in_waiting_list = True
if (event.booked_waiting_list_places + places_count) > event.waiting_list_places:
if (event.get_booked_waiting_list_places() + places_count) > event.waiting_list_places:
raise APIError(N_('sold out'))
else:
if (event.booked_places + places_count) > event.places:
if (event.get_booked_places() + places_count) > event.places:
raise APIError(N_('sold out'))
try:
with transaction.atomic():
if to_cancel_booking:
cancelled_booking_id = to_cancel_booking.pk
to_cancel_booking.cancel()
if to_cancel_booking:
cancelled_booking_id = to_cancel_booking.pk
to_cancel_booking.cancel()
# now we have a list of events, book them.
primary_booking = None
for dummy in range(places_count):
new_booking = make_booking(
event=event,
payload=payload,
extra_data=extra_data,
primary_booking=primary_booking,
in_waiting_list=in_waiting_list,
)
new_booking.save()
if primary_booking is None:
primary_booking = new_booking
except IntegrityError as e:
if 'tstzrange_constraint' in str(e):
# "optimistic concurrency control", between our availability
# check with get_all_slots() and now, new event can have been
# created and conflict with the events we want to create, and
# so we get an IntegrityError exception. In this case we
# restart the fillslot() from the begginning to redo the
# availability check and return a proper error to the client.
#
# To prevent looping, we raise an APIError during the second run
# of fillslot().
if retry:
raise APIError(N_('no more desk available'))
return self.fillslot(request=request, agenda=agenda, slot=slot, retry=True)
raise
# now we have a list of events, book them.
primary_booking = None
for dummy in range(places_count):
new_booking = make_booking(
event=event,
payload=payload,
extra_data=extra_data,
primary_booking=primary_booking,
in_waiting_list=in_waiting_list,
)
new_booking.save()
if lock_code and not confirm_after_lock:
Lease.objects.create(
booking=new_booking,
lock_code=lock_code,
)
if primary_booking is None:
primary_booking = new_booking
response = {
'err': 0,
@ -1625,6 +1632,9 @@ class RecurringFillslots(APIView):
if not serializer.is_valid():
raise APIErrorBadRequest(N_('invalid payload'), errors=serializer.errors)
data = serializer.validated_data
lock_code = data.get('lock_code')
if lock_code:
raise APIErrorBadRequest(N_('lock_code is unsupported'), errors=serializer.errors)
guardian_external_id = data.get('guardian_external_id')
start_datetime, end_datetime = data.get('date_start'), data.get('date_end')
@ -1898,6 +1908,7 @@ class EventsFillslots(APIView):
)
return self.fillslots(request)
@transaction.atomic()
def fillslots(self, request):
request_uuid = uuid.uuid4() if self.multiple_agendas else None
start_datetime, end_datetime = get_start_and_end_datetime_from_request(request)
@ -1911,6 +1922,11 @@ class EventsFillslots(APIView):
user_external_id = payload['user_external_id']
bypass_delays = payload.get('bypass_delays')
check_overlaps = payload.get('check_overlaps')
lock_code = payload.get('lock_code')
if lock_code:
confirm_after_lock = payload.get('confirm_after_lock') or False
Booking.objects.filter(event__agenda=self.agenda, lease__lock_code=lock_code).delete()
events = self.get_events(request, payload, start_datetime, end_datetime)
@ -2002,52 +2018,52 @@ class EventsFillslots(APIView):
events_by_id = {
x.id: x for x in (list(events) + events_to_unbook + events_to_unbook_out_of_min_delay)
}
with transaction.atomic():
# cancel existing bookings
cancellation_datetime = now()
Booking.objects.filter(primary_booking__in=bookings_to_cancel_out_of_min_delay).update(
cancellation_datetime=cancellation_datetime,
out_of_min_delay=True,
request_uuid=request_uuid,
previous_state='booked' if self.multiple_agendas else None,
cancellation_datetime = now()
Booking.objects.filter(primary_booking__in=bookings_to_cancel_out_of_min_delay).update(
cancellation_datetime=cancellation_datetime,
out_of_min_delay=True,
request_uuid=request_uuid,
previous_state='booked' if self.multiple_agendas else None,
)
cancelled_events = [
get_short_event_detail(
request,
events_by_id[x.event_id],
multiple_agendas=self.multiple_agendas,
)
cancelled_events = [
get_short_event_detail(
request,
events_by_id[x.event_id],
multiple_agendas=self.multiple_agendas,
)
for x in bookings_to_cancel_out_of_min_delay
]
cancelled_count = bookings_to_cancel_out_of_min_delay.update(
cancellation_datetime=cancellation_datetime, out_of_min_delay=True
for x in bookings_to_cancel_out_of_min_delay
]
cancelled_count = bookings_to_cancel_out_of_min_delay.update(
cancellation_datetime=cancellation_datetime, out_of_min_delay=True
)
Booking.objects.filter(primary_booking__in=bookings_to_cancel).update(
cancellation_datetime=cancellation_datetime, out_of_min_delay=False
)
cancelled_events += [
get_short_event_detail(
request,
events_by_id[x.event_id],
multiple_agendas=self.multiple_agendas,
)
Booking.objects.filter(primary_booking__in=bookings_to_cancel).update(
cancellation_datetime=cancellation_datetime,
out_of_min_delay=False,
request_uuid=request_uuid,
previous_state='booked' if self.multiple_agendas else None,
for x in bookings_to_cancel
]
cancelled_count += bookings_to_cancel.update(
cancellation_datetime=cancellation_datetime,
out_of_min_delay=False,
request_uuid=request_uuid,
previous_state='booked' if self.multiple_agendas else None,
)
# and delete outdated cancelled bookings
Booking.objects.filter(
user_external_id=user_external_id, event__in=events_cancelled_to_delete
).delete()
# create missing bookings
created_bookings = Booking.objects.bulk_create(bookings)
if lock_code and not confirm_after_lock:
Lease.objects.bulk_create(
Lease(booking=created_booking, lock_code=lock_code) for created_booking in created_bookings
)
cancelled_events += [
get_short_event_detail(
request,
events_by_id[x.event_id],
multiple_agendas=self.multiple_agendas,
)
for x in bookings_to_cancel
]
cancelled_count += bookings_to_cancel.update(
cancellation_datetime=cancellation_datetime,
out_of_min_delay=False,
request_uuid=request_uuid,
previous_state='booked' if self.multiple_agendas else None,
)
# and delete outdated cancelled bookings
Booking.objects.filter(
user_external_id=user_external_id, event__in=events_cancelled_to_delete
).delete()
# create missing bookings
created_bookings = Booking.objects.bulk_create(bookings)
# don't reload agendas and events types
for event in events:
@ -2931,7 +2947,7 @@ class AcceptBooking(APIView):
response = {
'err': 0,
'booking_id': booking.pk,
'overbooked_places': max(0, event.booked_places - event.places),
'overbooked_places': max(0, event.get_booked_places() - event.places),
}
return Response(response)
@ -3019,7 +3035,9 @@ class ResizeBooking(APIView):
# total places for the event (in waiting or main list, depending on the primary booking location)
places = event.waiting_list_places if booking.in_waiting_list else event.places
# total booked places for the event (in waiting or main list, depending on the primary booking location)
booked_places = event.booked_waiting_list_places if booking.in_waiting_list else event.booked_places
booked_places = (
event.get_booked_waiting_list_places() if booking.in_waiting_list else event.get_booked_places()
)
# places to book for this primary booking
primary_wanted_places = payload['count']
@ -3039,7 +3057,7 @@ class ResizeBooking(APIView):
if booking.in_waiting_list:
# booking in waiting list: can not be overbooked
raise APIError(N_('sold out'), err=3)
if event.booked_places <= event.places:
if event.get_booked_places() <= event.places:
# in main list and no overbooking for the moment: can not be overbooked
raise APIError(N_('sold out'), err=3)
return self.increase(booking, secondary_bookings, primary_booked_places, primary_wanted_places)

258
tests/api/fillslot/test_events.py
View File

@ -1,11 +1,13 @@
import datetime
import pytest
from django.core.management import call_command
from django.db import connection
from django.test.utils import CaptureQueriesContext
from chrono.agendas.models import Agenda, Booking, Event, EventsType
from chrono.agendas.models import Agenda, Booking, Event, EventsType, Lease
from chrono.utils.timezone import localtime, now
from tests.utils import build_event_agenda
pytestmark = pytest.mark.django_db
@ -589,3 +591,257 @@ def test_api_events_fillslots_partial_bookings(app, user):
resp.json['errors']['non_field_errors'][0]
== 'must include start_time and end_time for partial bookings agenda'
)
def test_lock_code(app, user, freezer):
agenda = build_event_agenda(
events={
'Event 1': {
'start_datetime': now() + datetime.timedelta(days=1),
'places': 1,
}
}
)
# list events
resp = app.get(agenda.get_datetimes_url())
slot = resp.json['data'][0]
assert slot['places']['available'] == 1
assert slot['places']['full'] is False
# book first one
fillslot_url = slot['api']['fillslot_url']
app.authorization = ('Basic', ('john.doe', 'password'))
resp = app.post_json(fillslot_url, params={'lock_code': 'MYLOCK'})
assert resp.json['err'] == 0
# list events without lock code
resp = app.get(agenda.get_datetimes_url())
slot = resp.json['data'][0]
assert slot['places']['available'] == 0
assert slot['places']['full'] is True
# list events with lock code
resp = app.get(agenda.get_datetimes_url(), params={'lock_code': 'MYLOCK', 'hide_disabled': 'true'})
slot = resp.json['data'][0]
assert slot['places']['available'] == 1
assert slot['places']['full'] is False
# re-book first one without lock code
app.authorization = ('Basic', ('john.doe', 'password'))
resp = app.post_json(fillslot_url)
assert resp.json['err'] == 1
# rebook first one with lock code
app.authorization = ('Basic', ('john.doe', 'password'))
resp = app.post_json(fillslot_url, params={'lock_code': 'MYLOCK'})
assert resp.json['err'] == 0
# confirm booking
app.authorization = ('Basic', ('john.doe', 'password'))
resp = app.post_json(fillslot_url, params={'lock_code': 'MYLOCK', 'confirm_after_lock': True})
assert resp.json['err'] == 0
# list events without lock code, after 30 minutes slot is still booked
freezer.move_to(datetime.timedelta(minutes=30))
resp = app.get(agenda.get_datetimes_url())
slot = resp.json['data'][0]
assert slot['places']['available'] == 0
assert slot['places']['full'] is True
def test_lock_code_expiration(app, user, freezer):
agenda = build_event_agenda(
events={
'Event 1': {
'start_datetime': now() + datetime.timedelta(days=1),
'places': 1,
}
}
)
# list events
resp = app.get(agenda.get_datetimes_url())
slot = resp.json['data'][0]
# book first one
fillslot_url = slot['api']['fillslot_url']
app.authorization = ('Basic', ('john.doe', 'password'))
resp = app.post_json(fillslot_url, params={'lock_code': 'MYLOCK'})
assert resp.json['err'] == 0
# list events without lock code
resp = app.get(agenda.get_datetimes_url())
slot = resp.json['data'][0]
assert slot['places']['available'] == 0
assert slot['places']['full'] is True
# list events without lock code, after 30 minutes slot is available
freezer.move_to(datetime.timedelta(minutes=30))
call_command('clean_leases')
resp = app.get(agenda.get_datetimes_url())
slot = resp.json['data'][0]
assert slot['places']['available'] == 1
assert slot['places']['full'] is False
def test_api_events_fillslots_with_lock_code(app, user, freezer):
freezer.move_to('2021-09-06 12:00')
app.authorization = ('Basic', ('john.doe', 'password'))
agenda = build_event_agenda(
events={
'Event': {
'start_datetime': now() + datetime.timedelta(days=1),
'places': 2,
'waiting_list_places': 1,
},
'Event 2': {
'start_datetime': now() + datetime.timedelta(days=1),
'places': 2,
'waiting_list_places': 1,
},
}
)
fillslots_url = '/api/agenda/%s/events/fillslots/' % agenda.slug
params = {
'user_external_id': 'user_id',
'check_overlaps': True,
'slots': 'event,event-2',
'lock_code': 'MYLOCK',
}
with CaptureQueriesContext(connection) as ctx:
resp = app.post_json(fillslots_url, params=params)
assert len(ctx.captured_queries) == 14
assert resp.json['booking_count'] == 2
assert len(resp.json['booked_events']) == 2
assert resp.json['booked_events'][0]['id'] == 'event'
assert (
resp.json['booked_events'][0]['booking']['id']
== Booking.objects.filter(event=agenda._event).latest('pk').pk
)
assert resp.json['booked_events'][1]['id'] == 'event-2'
assert (
resp.json['booked_events'][1]['booking']['id']
== Booking.objects.filter(event=agenda._event_2).latest('pk').pk
)
assert len(resp.json['waiting_list_events']) == 0
assert resp.json['cancelled_booking_count'] == 0
assert len(resp.json['cancelled_events']) == 0
events = Event.objects.all()
assert events.filter(booked_places=1).count() == 2
assert Booking.objects.count() == 2
assert Lease.objects.count() == 2
response = app.get(agenda.get_datetimes_url())
assert response.json['data'][0]['places']['available'] == 1
assert response.json['data'][0]['places']['reserved'] == 1
assert response.json['data'][1]['places']['available'] == 1
assert response.json['data'][1]['places']['reserved'] == 1
response = app.get(agenda.get_datetimes_url() + '?lock_code=MYLOCK')
assert response.json['data'][0]['places']['available'] == 2
assert response.json['data'][0]['places']['reserved'] == 0
assert response.json['data'][1]['places']['available'] == 2
assert response.json['data'][1]['places']['reserved'] == 0
# rebooking, nothing change
resp = app.post_json(fillslots_url, params=params)
assert Booking.objects.count() == 2
assert Lease.objects.count() == 2
response = app.get(agenda.get_datetimes_url())
assert response.json['data'][0]['places']['available'] == 1
assert response.json['data'][0]['places']['reserved'] == 1
assert response.json['data'][1]['places']['available'] == 1
assert response.json['data'][1]['places']['reserved'] == 1
response = app.get(agenda.get_datetimes_url() + '?lock_code=MYLOCK')
assert response.json['data'][0]['places']['available'] == 2
assert response.json['data'][0]['places']['reserved'] == 0
assert response.json['data'][1]['places']['available'] == 2
assert response.json['data'][1]['places']['reserved'] == 0
params['confirm_after_lock'] = 'true'
resp = app.post_json(fillslots_url, params=params)
assert Booking.objects.count() == 2
assert Lease.objects.count() == 0
response = app.get(agenda.get_datetimes_url())
assert response.json['data'][0]['places']['available'] == 1
assert response.json['data'][0]['places']['reserved'] == 1
assert response.json['data'][1]['places']['available'] == 1
assert response.json['data'][1]['places']['reserved'] == 1
response = app.get(agenda.get_datetimes_url() + '?lock_code=MYLOCK')
assert response.json['data'][0]['places']['available'] == 1
assert response.json['data'][0]['places']['reserved'] == 1
assert response.json['data'][1]['places']['available'] == 1
assert response.json['data'][1]['places']['reserved'] == 1
def test_api_events_fillslots_with_lock_code_expiration(app, user, freezer):
freezer.move_to('2021-09-06 12:00')
app.authorization = ('Basic', ('john.doe', 'password'))
agenda = build_event_agenda(
events={
'Event': {
'start_datetime': now() + datetime.timedelta(days=1),
'places': 2,
'waiting_list_places': 1,
},
'Event 2': {
'start_datetime': now() + datetime.timedelta(days=1),
'places': 2,
'waiting_list_places': 1,
},
}
)
fillslots_url = '/api/agenda/%s/events/fillslots/' % agenda.slug
params = {
'user_external_id': 'user_id',
'check_overlaps': True,
'slots': 'event,event-2',
'lock_code': 'MYLOCK',
}
with CaptureQueriesContext(connection) as ctx:
resp = app.post_json(fillslots_url, params=params)
assert len(ctx.captured_queries) == 14
assert resp.json['booking_count'] == 2
assert len(resp.json['booked_events']) == 2
assert resp.json['booked_events'][0]['id'] == 'event'
assert (
resp.json['booked_events'][0]['booking']['id']
== Booking.objects.filter(event=agenda._event).latest('pk').pk
)
assert resp.json['booked_events'][1]['id'] == 'event-2'
assert (
resp.json['booked_events'][1]['booking']['id']
== Booking.objects.filter(event=agenda._event_2).latest('pk').pk
)
assert len(resp.json['waiting_list_events']) == 0
assert resp.json['cancelled_booking_count'] == 0
assert len(resp.json['cancelled_events']) == 0
events = Event.objects.all()
assert events.filter(booked_places=1).count() == 2
assert Booking.objects.count() == 2
assert Lease.objects.count() == 2
response = app.get(agenda.get_datetimes_url())
assert response.json['data'][0]['places']['available'] == 1
assert response.json['data'][0]['places']['reserved'] == 1
assert response.json['data'][1]['places']['available'] == 1
assert response.json['data'][1]['places']['reserved'] == 1
freezer.move_to('2021-09-06 13:00')
call_command('clean_leases')
response = app.get(agenda.get_datetimes_url())
assert response.json['data'][0]['places']['available'] == 2
assert response.json['data'][0]['places']['reserved'] == 0
assert response.json['data'][1]['places']['available'] == 2
assert response.json['data'][1]['places']['reserved'] == 0

2
tests/api/fillslot/test_events_multiple_agendas.py
View File

@ -1161,4 +1161,4 @@ def test_api_events_fillslots_multiple_agendas_revert(app, user):
)
with CaptureQueriesContext(connection) as ctx:
resp = app.post(revert_url)
assert len(ctx.captured_queries) == 13
assert len(ctx.captured_queries) == 14

26
tests/test_utils.py
View File

@ -9,8 +9,9 @@ from requests.models import Response
from chrono.agendas.models import Agenda, TimePeriod
from chrono.utils.date import get_weekday_index
from chrono.utils.lingo import CheckType, get_agenda_check_types
from chrono.utils.timezone import localtime, now
from .utils import build_agendas, build_meetings_agenda, build_virtual_agenda, paris, utc
from .utils import build_agendas, build_event_agenda, build_meetings_agenda, build_virtual_agenda, paris, utc
def test_get_weekday_index():
@ -266,3 +267,26 @@ def test_paris():
def test_utc():
assert utc('2023-04-19T11:00:00').isoformat() == '2023-04-19T11:00:00+00:00'
def test_build_event_agenda(db):
start = now()
events = {
f'Event {i}': {
'start_datetime': start + datetime.timedelta(days=i),
'places': 10,
}
for i in range(10)
}
agenda = build_event_agenda(label='Agenda 3', events=events)
assert agenda.label == 'Agenda 3'
assert agenda.slug == 'agenda-3'
assert agenda.event_set.count() == 10
# ten spaced events
assert len(set(agenda.event_set.values_list('start_datetime', flat=True))) == 10
# at the same hour
assert (
len({localtime(start).time() for start in agenda.event_set.values_list('start_datetime', flat=True)})
== 1
)
assert agenda._event_1

9
tests/utils.py
View File

@ -111,6 +111,15 @@ def build_agenda(kind, label='Agenda', slug=None, **kwargs):
return Agenda.objects.create(**agenda_kwargs)
def build_event_agenda(label='Agenda', slug=None, events=None, **kwargs):
agenda = build_agenda('events', label=label, slug=slug, **(kwargs or {}))
for label, event_defn in (events or {}).items():
event = Event.objects.create(agenda=agenda, label=label, **event_defn)
setattr(agenda, f'_{event.slug.replace("-", "_")}', event)
return agenda
def build_meetings_agenda(
label='Agenda',
slug=None,