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

"préblocage" d'une réservation (#17685) #58

Closed
bdauvergne wants to merge 2 commits from wip/17685-preblocage-d-une-reservation into main
pull from: wip/17685-preblocage-d-une-reservation
merge into: entrouvert:main
Conversation 2 Commits 2 Files Changed 17 +1296 -30
17 changed files with 1296 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
chrono/agendas/migrations/0159_lease.py Normal file
View File

@ -0,0 +1,36 @@
# Generated by Django 3.2.18 on 2023-08-22 08:19
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('agendas', '0158_partial_booking_check_fields'),
]
operations = [
migrations.CreateModel(
name='Lease',
fields=[
(
'id',
models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
),
('lock_code', models.CharField(max_length=64, verbose_name='Lock code')),
('expiration_datetime', models.DateTimeField(verbose_name='Lease expiration time')),
(
'booking',
models.OneToOneField(
on_delete=django.db.models.deletion.CASCADE,
to='agendas.booking',
verbose_name='Booking',
),
),
],
options={
'verbose_name': 'Lease',
'verbose_name_plural': 'Leases',
},
),
]

77
chrono/agendas/models.py
View File

@ -1170,6 +1170,7 @@ class Agenda(models.Model):
start_datetime=None,
end_datetime=None,
user_external_id=None,
lock_code=None,
):
"""Get all occupation state of all possible slots for the given agenda (of
its real agendas for a virtual agenda) and the given meeting_type.
@ -1188,6 +1189,7 @@ class Agenda(models.Model):
and bookings sets.
If it is excluded, ignore it completely.
It if is booked, report the slot as full.
If it is booked but match the lock code, report the slot as open.
"""
resources = resources or []
# virtual agendas have one constraint :
@ -1291,6 +1293,8 @@ class Agenda(models.Model):
.order_by('desk_id', 'start_datetime', 'meeting_type__duration')
.values_list('desk_id', 'start_datetime', 'meeting_type__duration')
)
if lock_code:
booked_events = booked_events.exclude(booking__lease__lock_code=lock_code)
# compute exclusion set by desk from all bookings, using
# itertools.groupby() to group them by desk_id
bookings.update(
@ -1324,6 +1328,8 @@ class Agenda(models.Model):
.order_by('start_datetime', 'meeting_type__duration')
.values_list('start_datetime', 'meeting_type__duration')
)
if lock_code:
booked_events = booked_events.exclude(booking__lease__lock_code=lock_code)
# compute exclusion set
resources_bookings = IntervalSet.from_ordered(
(event_start_datetime, event_start_datetime + datetime.timedelta(minutes=event_duration))
@ -1349,6 +1355,8 @@ class Agenda(models.Model):
.order_by('start_datetime', 'meeting_type__duration')
.values_list('start_datetime', 'meeting_type__duration')
)
if lock_code:
booked_events = booked_events.exclude(booking__lease__lock_code=lock_code)
# compute exclusion set by desk from all bookings, using
# itertools.groupby() to group them by desk_id
user_bookings = IntervalSet.from_ordered(
@ -1565,6 +1573,11 @@ class Agenda(models.Model):
free_time += desk_free_time
return free_time
@property
def api_datetimes_url(self):
assert self.kind == 'events'
return reverse('api-agenda-datetimes', kwargs={'agenda_identifier': self.slug})
class VirtualMember(models.Model):
"""Trough model to link virtual agendas to their real agendas.
@ -1956,6 +1969,17 @@ class MeetingType(models.Model):
return new_meeting_type
@property
def api_datetimes_url(self):
assert self.agenda.kind in ('meetings', 'virtual')
return reverse(
'api-agenda-meeting-datetimes',
kwargs={
'agenda_identifier': self.agenda.slug,
'meeting_identifier': self.slug,
},
)
class Event(models.Model):
id = models.BigAutoField(primary_key=True)
@ -2017,6 +2041,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()
locked_booked_places = None
locked_booked_waiting_list_places = None
class Meta:
ordering = ['agenda', 'start_datetime', 'duration', 'label']
unique_together = ('agenda', 'slug')
@ -2193,6 +2223,27 @@ class Event(models.Model):
)
return qs
@staticmethod
def annotate_queryset_for_lock_code(qs, lock_code):
assert lock_code
qs = qs.annotate(
locked_booked_places=Count(
'booking',
filter=Q(
booking__cancellation_datetime__isnull=True,
)
& ~Q(booking__lease__lock_code=lock_code),
),
locked_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:
@ -2311,11 +2362,23 @@ class Event(models.Model):
@property
def remaining_places(self):
return max(0, self.places - self.booked_places)
return max(
0,
self.places
- (self.locked_booked_places if self.locked_booked_places is not None else self.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.locked_booked_waiting_list_places
if self.locked_booked_waiting_list_places is not None
else self.booked_waiting_list_places
),
)
@property
def end_datetime(self):
@ -4266,3 +4329,13 @@ class SharedCustodySettings(models.Model):
return cls.objects.get()
except cls.DoesNotExist:
return cls()
class Lease(models.Model):
booking = models.OneToOneField(Booking, on_delete=models.CASCADE, verbose_name=_('Booking'))
lock_code = models.CharField(verbose_name=_('Lock code'), max_length=64, blank=False)
expiration_datetime = models.DateTimeField(verbose_name=_('Lease expiration time'))
class Meta:
verbose_name = _('Lease')
verbose_name_plural = _('Leases')

8
chrono/api/serializers.py
View File

@ -1,6 +1,7 @@
import collections
import datetime
from django.conf import settings
from django.contrib.auth.models import Group
from django.db import models, transaction
from django.db.models import ExpressionWrapper, F, Q
@ -95,6 +96,12 @@ class FillSlotSerializer(serializers.Serializer):
required=False, child=serializers.CharField(max_length=16, allow_blank=False)
)
check_overlaps = serializers.BooleanField(default=False)
lock_code = serializers.CharField(max_length=64, required=False, allow_blank=False, trim_whitespace=True)
lock_duration = serializers.IntegerField(
min_value=settings.CHRONO_LOCK_MIN_DURATION,
max_value=settings.CHRONO_LOCK_MAX_DURATION,
) # in seconds
confirm_after_lock = serializers.BooleanField(default=False)
class SlotsSerializer(serializers.Serializer):
@ -357,6 +364,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)

118
chrono/api/views.py
View File

@ -20,6 +20,7 @@ import datetime
import json
import uuid
from django.conf import settings
from django.db import IntegrityError, transaction
from django.db.models import BooleanField, Count, ExpressionWrapper, F, Func, Prefetch, Q
from django.db.models.expressions import RawSQL
@ -47,6 +48,7 @@ from chrono.agendas.models import (
BookingColor,
Desk,
Event,
Lease,
MeetingType,
SharedCustodyAgenda,
Subscription,
@ -123,17 +125,26 @@ def get_event_places(event):
available = event.remaining_places
places = {
'total': event.places,
'reserved': event.booked_places,
# use the alternate computation of booked places if a lock_code is currently used
'reserved': event.booked_places if event.locked_booked_places is None else event.locked_booked_places,
'available': available,
'full': event.full,
# use the alternate computation of full status if a lock_code is currently used
'full': event.full
if event.locked_booked_places is None
else (available - event.locked_booked_places) < 1,
'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
if event.locked_booked_waiting_list_places is None:
booked_waiting_list_places = event.booked_waiting_list_places
else:
booked_waiting_list_places = event.locked_booked_waiting_list_places
places['waiting_list_reserved'] = 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'] = booked_waiting_list_places > 0 or available <= 0
# 'waiting_list_activated' means next booking will go into the waiting list
return places
@ -379,7 +390,7 @@ def get_events_meta_detail(
}
def get_events_from_slots(slots, request, agenda, payload):
def get_events_from_slots(slots, request, agenda, payload, lock_code=None):
user_external_id = payload.get('user_external_id') or None
exclude_user = payload.get('exclude_user')
book_events = payload.get('events') or request.query_params.get('events') or 'future'
@ -392,6 +403,9 @@ def get_events_from_slots(slots, request, agenda, payload):
except ValueError:
events = get_objects_from_slugs(slots, qs=agenda.event_set).order_by('start_datetime')
if lock_code:
events = Event.annotate_queryset_for_lock_code(events, lock_code)
for event in events:
if event.start_datetime >= now():
if not book_future or not event.in_bookable_period(bypass_delays=bypass_delays):
@ -494,6 +508,16 @@ def get_extra_data(request, payload):
return extra_data
def clean_meeting_events_with_expired_lease():
'''Delete expired meeting events, bookings and leases.'''
Event.objects.filter(agenda__kind='meetings', booking__lease__expiration_datetime__lt=now()).delete()
def clean_bookings_with_expired_lease():
'''Delete expired meeting event's bookings and leases'''
Booking.objects.filter(event__agenda__kind='events', lease__expiration_datetime__lt=now()).delete()
class Agendas(APIView):
serializer_class = serializers.AgendaSerializer
@ -595,6 +619,9 @@ class Datetimes(APIView):
serializer_class = serializers.DatetimesSerializer
def get(self, request, agenda_identifier=None, format=None):
# delete bookings with expired leases as we go
clean_bookings_with_expired_lease()
agenda_qs = Agenda.objects.select_related('events_type')
try:
agenda = agenda_qs.get(slug=agenda_identifier)
@ -632,6 +659,8 @@ class Datetimes(APIView):
bypass_delays=payload.get('bypass_delays'),
)
entries = Event.annotate_queryset_for_user(entries, user_external_id)
if payload.get('lock_code'):
entries = Event.annotate_queryset_for_lock_code(entries, lock_code=payload['lock_code'])
entries = entries.order_by('start_datetime', 'duration', 'label')
if payload['hide_disabled']:
@ -680,6 +709,9 @@ class MultipleAgendasDatetimes(APIView):
serializer_class = serializers.MultipleAgendasDatetimesSerializer
def get(self, request):
# delete bookings with expired leases as we go
clean_bookings_with_expired_lease()
serializer = self.serializer_class(data=request.query_params)
if not serializer.is_valid():
raise APIErrorBadRequest(N_('invalid payload'), errors=serializer.errors)
@ -807,6 +839,12 @@ class MeetingDatetimes(APIView):
N_('user_external_id and exclude_user_external_id have different values')
)
lock_code = request.GET.get('lock_code', None)
if lock_code is not None:
lock_code = lock_code.strip()
if lock_code == '':
raise APIErrorBadRequest(_('lock_code must not be empty'))
# Generate an unique slot for each possible meeting [start_datetime,
# end_datetime] range.
# First use get_all_slots() to get each possible meeting by desk and
@ -830,6 +868,7 @@ class MeetingDatetimes(APIView):
start_datetime=start_datetime,
end_datetime=end_datetime,
user_external_id=booked_user_external_id or excluded_user_external_id,
lock_code=lock_code,
)
)
for slot in sorted(all_slots, key=lambda slot: slot[:3]):
@ -851,6 +890,7 @@ class MeetingDatetimes(APIView):
start_datetime=start_datetime,
end_datetime=end_datetime,
user_external_id=booked_user_external_id or excluded_user_external_id,
lock_code=lock_code,
)
)
last_slot, slot_agendas = None, set()
@ -868,6 +908,9 @@ class MeetingDatetimes(APIView):
if last_slot:
yield last_slot, slot_agendas
# delete meetings with expired leases as we go
clean_meeting_events_with_expired_lease()
generator_of_unique_slots = unique_slots()
# create fillslot API URL as a template, to avoid expensive calls
@ -1197,6 +1240,12 @@ 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
lock_duration = payload.get('lock_duration') or settings.CHRONO_LOCK_DEFAULT_DURATION
confirm_after_lock = payload.get('confirm_after_lock') or False
if 'count' in payload:
places_count = payload['count']
elif 'count' in request.query_params:
@ -1240,10 +1289,18 @@ class EventsAgendaFillslot(APIView):
extra_data = get_extra_data(request, serializer.validated_data)
event = get_events_from_slots([slot], request, agenda, payload)[0]
# delete bookings with expired leases as we go
clean_bookings_with_expired_lease()
event = get_events_from_slots([slot], request, agenda, payload, lock_code=lock_code)[0]
# search free places. Switch to waiting list if necessary.
in_waiting_list = False
if payload.get('lock_code'):
# prevent overwriting of booked_places and booked_waiting_list_places
event.save = NotImplemented
event.booked_places = event.locked_booked_places
event.booked_waiting_list_places = event.locked_booked_waiting_list_places
if event.start_datetime > now():
if payload.get('force_waiting_list') and not event.waiting_list_places:
raise APIError(N_('no waiting list'))
@ -1265,6 +1322,11 @@ class EventsAgendaFillslot(APIView):
try:
with transaction.atomic():
if lock_code:
# 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 to_cancel_booking:
cancelled_booking_id = to_cancel_booking.pk
to_cancel_booking.cancel()
@ -1280,6 +1342,13 @@ class EventsAgendaFillslot(APIView):
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,
expiration_datetime=now() + datetime.timedelta(seconds=lock_duration),
)
if primary_booking is None:
primary_booking = new_booking
except IntegrityError as e:
@ -1364,6 +1433,12 @@ class MeetingsAgendaFillslot(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
lock_duration = payload.get('lock_duration') or settings.CHRONO_LOCK_DEFAULT_DURATION
confirm_after_lock = payload.get('confirm_after_lock') or False
to_cancel_booking = None
cancel_booking_id = None
if payload.get('cancel_booking_id'):
@ -1412,6 +1487,10 @@ class MeetingsAgendaFillslot(APIView):
meeting_type = agenda.get_meetingtype(id_=meeting_type_id)
except (MeetingType.DoesNotExist, ValueError):
raise APIErrorBadRequest(N_('invalid meeting type id: %s'), meeting_type_id)
# delete meeting events with expired leases as we go
clean_meeting_events_with_expired_lease()
all_slots = sorted(
agenda.get_all_slots(
meeting_type,
@ -1419,6 +1498,7 @@ class MeetingsAgendaFillslot(APIView):
user_external_id=user_external_id if exclude_user else None,
start_datetime=slot_datetime,
end_datetime=slot_datetime + datetime.timedelta(minutes=meeting_type.duration),
lock_code=lock_code,
),
key=lambda slot: slot.start_datetime,
)
@ -1495,6 +1575,12 @@ class MeetingsAgendaFillslot(APIView):
try:
with transaction.atomic():
if lock_code:
# delete events/bookings with the same lock code in the
# same agenda(s) to allow rebooking or confirming
Event.objects.filter(
agenda__in=agenda.get_real_agendas(), booking__lease__lock_code=lock_code
).delete()
if to_cancel_booking:
cancelled_booking_id = to_cancel_booking.pk
to_cancel_booking.cancel()
@ -1510,6 +1596,13 @@ class MeetingsAgendaFillslot(APIView):
color=color,
)
booking.save()
if lock_code and not confirm_after_lock:
Lease.objects.create(
booking=booking,
lock_code=lock_code,
expiration_datetime=now() + datetime.timedelta(seconds=lock_duration),
)
except IntegrityError as e:
if 'tstzrange_constraint' in str(e):
# "optimistic concurrency control", between our availability
@ -1861,6 +1954,9 @@ class EventsFillslots(APIView):
bypass_delays = payload.get('bypass_delays')
check_overlaps = payload.get('check_overlaps')
# delete bookings with expired leases as we go
clean_bookings_with_expired_lease()
events = self.get_events(request, payload, start_datetime, end_datetime)
if check_overlaps:
@ -2042,6 +2138,9 @@ class MultipleAgendasEventsFillslots(EventsFillslots):
agenda, event = slot.split('@')
events_by_agenda[agenda].append(event)
# delete bookings with expired leases as we go
clean_bookings_with_expired_lease()
agendas_by_slug = get_objects_from_slugs(
events_by_agenda.keys(), qs=Agenda.objects.filter(kind='events')
).in_bulk(field_name='slug')
@ -2149,6 +2248,9 @@ class MultipleAgendasEventsCheckStatus(APIView):
serializer_class = serializers.MultipleAgendasEventsCheckStatusSerializer
def get(self, request):
# delete bookings with expired leases as we go
clean_bookings_with_expired_lease()
serializer = self.serializer_class(data=request.query_params)
if not serializer.is_valid():
@ -2172,6 +2274,7 @@ class MultipleAgendasEventsCheckStatus(APIView):
booking_queryset = Booking.objects.filter(
event__in=events,
user_external_id=user_external_id,
lease__isnull=True,
)
bookings_by_event_id = collections.defaultdict(list)
for booking in booking_queryset:
@ -2515,6 +2618,9 @@ class BookingsAPI(ListAPIView):
filterset_class = BookingFilter
def get(self, request, *args, **kwargs):
# delete bookings with expired leases as we go
clean_bookings_with_expired_lease()
if not request.GET.get('user_external_id'):
raise APIError(N_('missing param user_external_id'))

5
chrono/settings.py
View File

@ -177,6 +177,11 @@ MELLON_IDENTITY_PROVIDERS = []
# (see http://docs.python-requests.org/en/master/user/advanced/#proxies)
REQUESTS_PROXIES = None
# default and max lock duration, in seconds
CHRONO_LOCK_MIN_DURATION = 5 * 60
CHRONO_LOCK_DEFAULT_DURATION = 10 * 60
CHRONO_LOCK_MAX_DURATION = 20 * 60
# timeout used in python-requests call, in seconds
# we use 28s by default: timeout just before web server, which is usually 30s
REQUESTS_TIMEOUT = 28

4
tests/api/datetimes/test_all.py
View File

@ -241,7 +241,7 @@ def test_datetimes_api_(app):
)
with CaptureQueriesContext(connection) as ctx:
resp = app.get('/api/agenda/%s/datetimes/' % agenda.slug)
assert len(ctx.captured_queries) == 2
assert len(ctx.captured_queries) == 3
assert resp.json['data'][0]['custom_field_text'] == 'foo'
assert resp.json['data'][0]['custom_field_textarea'] == 'foo bar'
assert resp.json['data'][0]['custom_field_bool'] is True
@ -713,7 +713,7 @@ def test_recurring_events_api(app, user, freezer):
# check querysets
with CaptureQueriesContext(connection) as ctx:
app.get('/api/agenda/%s/datetimes/' % agenda.slug)
assert len(ctx.captured_queries) == 3
assert len(ctx.captured_queries) == 4
# events follow agenda display template
agenda.event_display_template = '{{ event.label }} - {{ event.start_datetime }}'

8
tests/api/datetimes/test_events_multiple_agendas.py
View File

@ -387,7 +387,7 @@ def test_datetimes_multiple_agendas_queries(app):
},
)
assert len(resp.json['data']) == 30
assert len(ctx.captured_queries) == 2
assert len(ctx.captured_queries) == 3
with CaptureQueriesContext(connection) as ctx:
resp = app.get(
@ -400,7 +400,7 @@ def test_datetimes_multiple_agendas_queries(app):
},
)
assert len(resp.json['data']) == 30
assert len(ctx.captured_queries) == 2
assert len(ctx.captured_queries) == 3
with CaptureQueriesContext(connection) as ctx:
resp = app.get(
@ -413,7 +413,7 @@ def test_datetimes_multiple_agendas_queries(app):
},
)
assert len(resp.json['data']) == 30
assert len(ctx.captured_queries) == 2
assert len(ctx.captured_queries) == 3
father = Person.objects.create(user_external_id='father_id', first_name='John', last_name='Doe')
mother = Person.objects.create(user_external_id='mother_id', first_name='Jane', last_name='Doe')
@ -442,7 +442,7 @@ def test_datetimes_multiple_agendas_queries(app):
},
)
assert len(resp.json['data']) == 30
assert len(ctx.captured_queries) == 3
assert len(ctx.captured_queries) == 4
@pytest.mark.freeze_time('2021-05-06 14:00')

650
tests/api/datetimes/test_meetings.py
View File

@ -14,6 +14,7 @@ from chrono.agendas.models import (
TimePeriod,
TimePeriodException,
UnavailabilityCalendar,
VirtualMember,
)
from chrono.utils.timezone import localtime, make_aware, now
@ -298,7 +299,7 @@ def test_datetimes_api_meetings_agenda_with_resources(app):
)
with CaptureQueriesContext(connection) as ctx:
resp = app.get(api_url)
assert len(ctx.captured_queries) == 10
assert len(ctx.captured_queries) == 11
assert len(resp.json['data']) == 32
assert [s['datetime'] for s in resp.json['data'] if s['disabled'] is True] == [
'%s 09:00:00' % tomorrow_str,
@ -502,7 +503,7 @@ def test_datetimes_api_meetings_agenda_exclude_slots(app):
'/api/agenda/%s/meetings/%s/datetimes/' % (agenda.slug, meeting_type.slug),
params={'exclude_user_external_id': '42'},
)
assert len(ctx.captured_queries) == 9
assert len(ctx.captured_queries) == 10
assert resp.json['data'][0]['id'] == 'foo-bar:2021-02-26-0900'
assert resp.json['data'][0]['disabled'] is True
assert 'booked_for_external_user' not in resp.json['data'][0]
@ -558,7 +559,7 @@ def test_datetimes_api_meetings_agenda_user_external_id(app):
'/api/agenda/%s/meetings/%s/datetimes/' % (agenda.slug, meeting_type.slug),
params={'user_external_id': '42'},
)
assert len(ctx.captured_queries) == 9
assert len(ctx.captured_queries) == 10
assert resp.json['data'][0]['booked_for_external_user'] is True
# mix with exclude_user_external_id
@ -1045,6 +1046,647 @@ def test_datetimes_api_meetings_agenda_start_hour_change(app, meetings_agenda):
assert len([x for x in resp.json['data'] if x['disabled']]) == 2
def test_virtual_agendas_meetings_datetimes_api(app, virtual_meetings_agenda):
real_agenda = virtual_meetings_agenda.real_agendas.first()
meeting_type = real_agenda.meetingtype_set.first()
default_desk = real_agenda.desk_set.first()
# Unkown meeting
app.get('/api/agenda/%s/meetings/xxx/datetimes/' % virtual_meetings_agenda.slug, status=404)
virt_meeting_type = virtual_meetings_agenda.iter_meetingtypes()[0]
api_url = '/api/agenda/%s/meetings/%s/datetimes/' % (virtual_meetings_agenda.slug, virt_meeting_type.slug)
resp = app.get(api_url)
assert len(resp.json['data']) == 144
# cover completely to test limit condition in get_all_slots()
full_coverage = TimePeriodException.objects.create(
desk=default_desk,
start_datetime=make_aware(datetime.datetime(2017, 1, 1, 0, 0)),
end_datetime=make_aware(datetime.datetime(2018, 1, 1, 0, 0)),
)
resp = app.get(api_url)
assert len(resp.json['data']) == 0
full_coverage.delete()
virtual_meetings_agenda.minimal_booking_delay = 7
virtual_meetings_agenda.maximal_booking_delay = 28
virtual_meetings_agenda.save()
resp = app.get(api_url)
assert len(resp.json['data']) == 54
virtual_meetings_agenda.minimal_booking_delay = 1
virtual_meetings_agenda.maximal_booking_delay = 56
virtual_meetings_agenda.save()
resp = app.get(api_url)
assert len(resp.json['data']) == 144
resp = app.get(api_url)
dt = datetime.datetime.strptime(resp.json['data'][2]['id'].split(':')[1], '%Y-%m-%d-%H%M')
ev = Event(
agenda=real_agenda,
meeting_type=meeting_type,
places=1,
full=False,
start_datetime=make_aware(dt),
desk=default_desk,
)
ev.save()
booking = Booking(event=ev)
booking.save()
resp2 = app.get(api_url)
assert len(resp2.json['data']) == 144
assert resp.json['data'][0] == resp2.json['data'][0]
assert resp.json['data'][1] == resp2.json['data'][1]
assert resp.json['data'][2] != resp2.json['data'][2]
assert resp.json['data'][2]['disabled'] is False
assert resp2.json['data'][2]['disabled'] is True
assert resp.json['data'][3] == resp2.json['data'][3]
# test with a timeperiod overlapping current moment, it should get one
# datetime for the current timeperiod + two from the next week.
if localtime(now()).time().hour == 23:
# skip this part of the test as it would require support for events
# crossing midnight
return
TimePeriod.objects.filter(desk=default_desk).delete()
start_time = localtime(now()) - datetime.timedelta(minutes=10)
time_period = TimePeriod(
weekday=localtime(now()).weekday(),
start_time=start_time,
end_time=start_time + datetime.timedelta(hours=1),
desk=default_desk,
)
time_period.save()
virtual_meetings_agenda.minimal_booking_delay = 0
virtual_meetings_agenda.maximal_booking_delay = 10
virtual_meetings_agenda.save()
resp = app.get(api_url)
assert len(resp.json['data']) == 3
def test_virtual_agendas_meetings_datetimes_api_with_similar_desk(app):
agenda_foo = Agenda.objects.create(
label='Agenda Foo', kind='meetings', minimal_booking_delay=1, maximal_booking_delay=4
)
MeetingType.objects.create(agenda=agenda_foo, label='Meeting Type', duration=30)
test_1st_weekday = (localtime(now()).weekday() + 1) % 7
test_2nd_weekday = (localtime(now()).weekday() + 2) % 7
test_3rd_weekday = (localtime(now()).weekday() + 3) % 7
desk_foo = Desk.objects.create(agenda=agenda_foo, label='Desk 1')
TimePeriod.objects.create(
weekday=test_1st_weekday,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=desk_foo,
)
TimePeriod.objects.create(
weekday=test_2nd_weekday,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=desk_foo,
)
TimePeriod.objects.create(
weekday=test_3rd_weekday,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=desk_foo,
)
agenda_bar = Agenda.objects.create(
label='Agenda Bar', kind='meetings', minimal_booking_delay=1, maximal_booking_delay=4
)
meeting_type_bar = MeetingType.objects.create(agenda=agenda_bar, label='Meeting Type', duration=30)
desk_bar = Desk.objects.create(agenda=agenda_bar, label='Desk 1')
TimePeriod.objects.create(
weekday=test_1st_weekday,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=desk_bar,
)
TimePeriod.objects.create(
weekday=test_2nd_weekday,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=desk_bar,
)
TimePeriod.objects.create(
weekday=test_3rd_weekday,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=desk_bar,
)
virtual_agenda = Agenda.objects.create(
label='Agenda Virtual', kind='virtual', minimal_booking_delay=1, maximal_booking_delay=4
)
VirtualMember.objects.create(virtual_agenda=virtual_agenda, real_agenda=agenda_foo)
VirtualMember.objects.create(virtual_agenda=virtual_agenda, real_agenda=agenda_bar)
# 4 slots each day * 3 days
foo_api_url = '/api/agenda/%s/meetings/%s/datetimes/' % (agenda_foo.slug, meeting_type_bar.slug)
resp = app.get(foo_api_url)
assert len(resp.json['data']) == 12
# same thing bar agenda
bar_api_url = '/api/agenda/%s/meetings/%s/datetimes/' % (agenda_foo.slug, meeting_type_bar.slug)
resp = app.get(bar_api_url)
assert len(resp.json['data']) == 12
# same thing on the virtual agenda
virtual_api_url = '/api/agenda/%s/meetings/%s/datetimes/' % (virtual_agenda.slug, meeting_type_bar.slug)
resp = app.get(virtual_api_url)
assert len(resp.json['data']) == 12
# exclude first day
start = (localtime(now()) + datetime.timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
end = (localtime(now()) + datetime.timedelta(days=1)).replace(
hour=23, minute=59, second=59, microsecond=0
)
TimePeriodException.objects.create(start_datetime=start, end_datetime=end, desk=desk_foo)
TimePeriodException.objects.create(start_datetime=start, end_datetime=end, desk=desk_bar)
# exclude second day
start = (localtime(now()) + datetime.timedelta(days=2)).replace(hour=0, minute=0, second=0, microsecond=0)
end = (localtime(now()) + datetime.timedelta(days=2)).replace(
hour=23, minute=59, second=59, microsecond=0
)
TimePeriodException.objects.create(start_datetime=start, end_datetime=end, desk=desk_foo)
TimePeriodException.objects.create(start_datetime=start, end_datetime=end, desk=desk_bar)
# 4 slots each day * 1 day
resp = app.get(foo_api_url)
assert len(resp.json['data']) == 4
# same thing bar agenda
resp = app.get(bar_api_url)
assert len(resp.json['data']) == 4
# same thing on the virtual agenda
resp = app.get(virtual_api_url)
assert len(resp.json['data']) == 4
def test_virtual_agendas_meetings_datetimes_delays_api(app, mock_now):
foo_agenda = Agenda.objects.create(label='Foo Meeting', kind='meetings', maximal_booking_delay=7)
MeetingType.objects.create(agenda=foo_agenda, label='Meeting Type', duration=30)
foo_desk_1 = Desk.objects.create(agenda=foo_agenda, label='Foo desk 1')
TimePeriod.objects.create(
weekday=0,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=foo_desk_1,
)
TimePeriod.objects.create(
weekday=1,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=foo_desk_1,
)
bar_agenda = Agenda.objects.create(label='Bar Meeting', kind='meetings', maximal_booking_delay=7)
MeetingType.objects.create(agenda=bar_agenda, label='Meeting Type', duration=30)
bar_desk_1 = Desk.objects.create(agenda=bar_agenda, label='Bar desk 1')
TimePeriod.objects.create(
weekday=2,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=bar_desk_1,
)
TimePeriod.objects.create(
weekday=3,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=bar_desk_1,
)
virt_agenda = Agenda.objects.create(label='Virtual Agenda', kind='virtual')
VirtualMember.objects.create(virtual_agenda=virt_agenda, real_agenda=foo_agenda)
VirtualMember.objects.create(virtual_agenda=virt_agenda, real_agenda=bar_agenda)
virt_meeting_type = virt_agenda.iter_meetingtypes()[0]
api_url = '/api/agenda/%s/meetings/%s/datetimes/' % (virt_agenda.slug, virt_meeting_type.slug)
resp = app.get(api_url)
# 8 slots for m each agenda
assert len(resp.json['data']) == 16
# restrict foo's minimal_booking_delay : only bar's slots are left
foo_agenda.minimal_booking_delay = 6
foo_agenda.save()
resp = app.get(api_url)
assert len(resp.json['data']) == 8
# restrict bar's maximal_booking_delay : only half of bar's slots are left
bar_agenda.maximal_booking_delay = 5
bar_agenda.save()
resp = app.get(api_url)
assert len(resp.json['data']) == 4
# put back very slots from foo
foo_agenda.minimal_booking_delay = 1
foo_agenda.maximal_booking_delay = 7
foo_agenda.save()
resp = app.get(api_url)
assert len(resp.json['data']) == 12
def test_virtual_agendas_meetings_datetimes_exluded_periods(app, mock_now):
foo_agenda = Agenda.objects.create(label='Foo Meeting', kind='meetings', maximal_booking_delay=7)
MeetingType.objects.create(agenda=foo_agenda, label='Meeting Type', duration=30)
foo_desk_1 = Desk.objects.create(agenda=foo_agenda, label='Foo desk 1')
TimePeriod.objects.create(
weekday=0,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=foo_desk_1,
)
TimePeriod.objects.create(
weekday=1,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=foo_desk_1,
)
virt_agenda = Agenda.objects.create(label='Virtual Agenda', kind='virtual')
VirtualMember.objects.create(virtual_agenda=virt_agenda, real_agenda=foo_agenda)
api_url = '/api/agenda/%s/meetings/meeting-type/datetimes/' % (virt_agenda.slug)
resp = app.get(api_url)
# 8 slots
data = resp.json['data']
assert len(data) == 8
assert data[0]['datetime'] == '2017-05-22 10:00:00'
assert data[1]['datetime'] == '2017-05-22 10:30:00'
assert data[2]['datetime'] == '2017-05-22 11:00:00'
# exclude one hour the first day
tp1 = TimePeriod.objects.create(
weekday=0, start_time=datetime.time(11, 0), end_time=datetime.time(12, 0), agenda=virt_agenda
)
resp = app.get(api_url)
data = resp.json['data']
assert len(data) == 6
assert data[0]['datetime'] == '2017-05-22 10:00:00'
assert data[1]['datetime'] == '2017-05-22 10:30:00'
# no more slots the 22 thanks to the exclusion period
assert data[2]['datetime'] == '2017-05-23 10:00:00'
# exclude the second day
tp2 = TimePeriod.objects.create(
weekday=1, start_time=datetime.time(9, 0), end_time=datetime.time(18, 0), agenda=virt_agenda
)
resp = app.get(api_url)
data = resp.json['data']
assert len(data) == 2
assert data[0]['datetime'] == '2017-05-22 10:00:00'
assert data[1]['datetime'] == '2017-05-22 10:30:00'
# go back to no restriction
tp1.delete()
tp2.delete()
resp = app.get(api_url)
data = resp.json['data']
assert len(data) == 8
# excluded period applies to every desk
foo_desk_2 = Desk.objects.create(agenda=foo_agenda, label='Foo desk 2')
TimePeriod.objects.create(
weekday=3,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=foo_desk_2,
)
TimePeriod.objects.create(
weekday=4,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=foo_desk_2,
)
resp = app.get(api_url)
data = resp.json['data']
assert len(data) == 16
# exclude one hour the first day
tp1 = TimePeriod.objects.create(
weekday=0, start_time=datetime.time(11, 0), end_time=datetime.time(12, 0), agenda=virt_agenda
)
resp = app.get(api_url)
data = resp.json['data']
assert len(data) == 14
# exclude one hour the last day
tp2 = TimePeriod.objects.create(
weekday=4, start_time=datetime.time(11, 0), end_time=datetime.time(12, 0), agenda=virt_agenda
)
resp = app.get(api_url)
data = resp.json['data']
assert len(data) == 12
# go back to no restriction
tp1.delete()
tp2.delete()
resp = app.get(api_url)
data = resp.json['data']
assert len(data) == 16
# add a second real agenda
bar_agenda = Agenda.objects.create(label='Bar Meeting', kind='meetings', maximal_booking_delay=7)
VirtualMember.objects.create(virtual_agenda=virt_agenda, real_agenda=bar_agenda)
MeetingType.objects.create(agenda=bar_agenda, label='Meeting Type', duration=30)
bar_desk_1 = Desk.objects.create(agenda=bar_agenda, label='Bar desk 1')
bar_desk_2 = Desk.objects.create(agenda=bar_agenda, label='Bar desk 2')
TimePeriod.objects.create(
weekday=0,
start_time=datetime.time(14, 0),
end_time=datetime.time(16, 0),
desk=bar_desk_1,
)
TimePeriod.objects.create(
weekday=1,
start_time=datetime.time(14, 0),
end_time=datetime.time(16, 0),
desk=bar_desk_1,
)
TimePeriod.objects.create(
weekday=2,
start_time=datetime.time(14, 0),
end_time=datetime.time(16, 0),
desk=bar_desk_2,
)
TimePeriod.objects.create(
weekday=3,
start_time=datetime.time(14, 0),
end_time=datetime.time(16, 0),
desk=bar_desk_2,
)
resp = app.get(api_url)
data = resp.json['data']
assert len(data) == 32
# exclude the first day, 11 to 15 : 4 slots
tp1 = TimePeriod.objects.create(
weekday=0, start_time=datetime.time(11, 0), end_time=datetime.time(15, 0), agenda=virt_agenda
)
resp = app.get(api_url)
data = resp.json['data']
assert len(data) == 28
def test_virtual_agendas_meetings_exception(app, user, virtual_meetings_agenda):
app.authorization = ('Basic', ('john.doe', 'password'))
real_agenda = virtual_meetings_agenda.real_agendas.first()
desk = real_agenda.desk_set.first()
virt_meeting_type = virtual_meetings_agenda.iter_meetingtypes()[0]
datetimes_url = '/api/agenda/%s/meetings/%s/datetimes/' % (
virtual_meetings_agenda.slug,
virt_meeting_type.slug,
)
resp = app.get(datetimes_url)
# test exception at the lowest limit
excp1 = TimePeriodException.objects.create(
desk=desk,
start_datetime=make_aware(datetime.datetime(2017, 5, 22, 10, 0)),
end_datetime=make_aware(datetime.datetime(2017, 5, 22, 12, 0)),
)
resp2 = app.get(datetimes_url)
assert len(resp.json['data']) == len(resp2.json['data']) + 4
# test exception at the highest limit
excp1.end_datetime = make_aware(datetime.datetime(2017, 5, 22, 11, 0))
excp1.save()
resp2 = app.get(datetimes_url)
assert len(resp.json['data']) == len(resp2.json['data']) + 2
# add an exception with an end datetime less than excp1 end datetime
# and make sure that excp1 end datetime preveil
excp1.end_datetime = make_aware(datetime.datetime(2017, 5, 23, 11, 0))
excp1.save()
TimePeriodException.objects.create(
desk=excp1.desk,
start_datetime=make_aware(datetime.datetime(2017, 5, 22, 15, 0)),
end_datetime=make_aware(datetime.datetime(2017, 5, 23, 9, 0)),
)
resp2 = app.get(datetimes_url)
assert len(resp.json['data']) == len(resp2.json['data']) + 6
# with a second desk
desk2 = Desk.objects.create(label='Desk 2', agenda=real_agenda)
time_period = desk.timeperiod_set.first()
TimePeriod.objects.create(
desk=desk2,
start_time=time_period.start_time,
end_time=time_period.end_time,
weekday=time_period.weekday,
)
resp3 = app.get(datetimes_url)
assert len(resp.json['data']) == len(resp3.json['data']) + 2 # +2 because excp1 changed
def test_virtual_agendas_meetings_datetimes_multiple_agendas(app, mock_now):
foo_agenda = Agenda.objects.create(
label='Foo Meeting', kind='meetings', minimal_booking_delay=1, maximal_booking_delay=5
)
foo_meeting_type = MeetingType.objects.create(agenda=foo_agenda, label='Meeting Type', duration=30)
foo_desk_1 = Desk.objects.create(agenda=foo_agenda, label='Foo desk 1')
test_1st_weekday = (localtime(now()).weekday() + 2) % 7
test_2nd_weekday = (localtime(now()).weekday() + 3) % 7
test_3rd_weekday = (localtime(now()).weekday() + 4) % 7
test_4th_weekday = (localtime(now()).weekday() + 5) % 7
def create_time_perdiods(desk, end=12):
TimePeriod.objects.create(
weekday=test_1st_weekday,
start_time=datetime.time(10, 0),
end_time=datetime.time(end, 0),
desk=desk,
)
TimePeriod.objects.create(
weekday=test_2nd_weekday,
start_time=datetime.time(10, 0),
end_time=datetime.time(end, 0),
desk=desk,
)
create_time_perdiods(foo_desk_1)
virt_agenda = Agenda.objects.create(
label='Virtual Agenda', kind='virtual', minimal_booking_delay=1, maximal_booking_delay=6
)
VirtualMember.objects.create(virtual_agenda=virt_agenda, real_agenda=foo_agenda)
virt_meeting_type = virt_agenda.iter_meetingtypes()[0]
# We are saturday and we can book for next monday and tuesday, 4 slots available each day
api_url = '/api/agenda/%s/meetings/%s/datetimes/' % (virt_agenda.slug, virt_meeting_type.slug)
resp = app.get(api_url)
assert len(resp.json['data']) == 8
assert resp.json['data'][0]['id'] == 'meeting-type:2017-05-22-1000'
virt_agenda.maximal_booking_delay = 10 # another monday comes in
virt_agenda.save()
resp = app.get(api_url)
assert len(resp.json['data']) == 12
# Back to next monday and tuesday restriction
virt_agenda.maximal_booking_delay = 6
virt_agenda.save()
# Add another agenda
bar_agenda = Agenda.objects.create(
label='Bar Meeting', kind='meetings', minimal_booking_delay=1, maximal_booking_delay=5
)
bar_meeting_type = MeetingType.objects.create(agenda=bar_agenda, label='Meeting Type', duration=30)
bar_desk_1 = Desk.objects.create(agenda=bar_agenda, label='Bar desk 1')
create_time_perdiods(bar_desk_1, end=13) # bar_agenda has two more slots each day
VirtualMember.objects.create(virtual_agenda=virt_agenda, real_agenda=bar_agenda)
with CaptureQueriesContext(connection) as ctx:
resp = app.get(api_url)
assert len(resp.json['data']) == 12
assert len(ctx.captured_queries) == 13
# simulate booking
dt = datetime.datetime.strptime(resp.json['data'][2]['id'].split(':')[1], '%Y-%m-%d-%H%M')
ev = Event.objects.create(
agenda=foo_agenda,
meeting_type=foo_meeting_type,
places=1,
full=False,
start_datetime=make_aware(dt),
desk=foo_desk_1,
)
booking1 = Booking.objects.create(event=ev)
resp = app.get(api_url)
assert len(resp.json['data']) == 12
# No disabled slot, because the booked slot is still available in second agenda
for slot in resp.json['data']:
assert slot['disabled'] is False
ev = Event.objects.create(
agenda=bar_agenda,
meeting_type=bar_meeting_type,
places=1,
full=False,
start_datetime=make_aware(dt),
desk=bar_desk_1,
)
booking2 = Booking.objects.create(event=ev)
resp = app.get(api_url)
assert len(resp.json['data']) == 12
# now one slot is disabled
for i, slot in enumerate(resp.json['data']):
if i == 2:
assert slot['disabled']
else:
assert slot['disabled'] is False
# Cancel booking, every slot available
booking1.cancel()
booking2.cancel()
resp = app.get(api_url)
assert len(resp.json['data']) == 12
for slot in resp.json['data']:
assert slot['disabled'] is False
# Add new desk on foo_agenda, open on wednesday
foo_desk_2 = Desk.objects.create(agenda=foo_agenda, label='Foo desk 2')
TimePeriod.objects.create(
weekday=test_3rd_weekday,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=foo_desk_2,
)
resp = app.get(api_url)
assert len(resp.json['data']) == 16
# Add new desk on bar_agenda, open on thursday
bar_desk_2 = Desk.objects.create(agenda=bar_agenda, label='Bar desk 2')
TimePeriod.objects.create(
weekday=test_4th_weekday,
start_time=datetime.time(10, 0),
end_time=datetime.time(12, 0),
desk=bar_desk_2,
)
resp = app.get(api_url)
assert len(resp.json['data']) == 20
@pytest.mark.freeze_time('2021-02-25')
def test_virtual_agendas_meetings_datetimes_exclude_slots(app):
tomorrow = now() + datetime.timedelta(days=1)
agenda = Agenda.objects.create(
label='Agenda', kind='meetings', minimal_booking_delay=0, maximal_booking_delay=10
)
desk = Desk.objects.create(agenda=agenda, slug='desk')
meeting_type = MeetingType.objects.create(agenda=agenda, slug='foo-bar')
TimePeriod.objects.create(
weekday=tomorrow.date().weekday(),
start_time=datetime.time(9, 0),
end_time=datetime.time(17, 00),
desk=desk,
)
agenda2 = agenda.duplicate()
virt_agenda = Agenda.objects.create(
label='Virtual Agenda', kind='virtual', minimal_booking_delay=1, maximal_booking_delay=10
)
VirtualMember.objects.create(virtual_agenda=virt_agenda, real_agenda=agenda)
VirtualMember.objects.create(virtual_agenda=virt_agenda, real_agenda=agenda2)
event = Event.objects.create(
agenda=agenda,
meeting_type=meeting_type,
places=1,
start_datetime=localtime(tomorrow).replace(hour=9, minute=0),
desk=desk,
)
Booking.objects.create(event=event, user_external_id='42')
event2 = Event.objects.create(
agenda=agenda,
meeting_type=meeting_type,
places=1,
start_datetime=localtime(tomorrow).replace(hour=10, minute=0),
desk=desk,
)
cancelled = Booking.objects.create(event=event2, user_external_id='35')
cancelled.cancel()
resp = app.get('/api/agenda/%s/meetings/%s/datetimes/' % (virt_agenda.slug, meeting_type.slug))
assert resp.json['data'][0]['id'] == 'foo-bar:2021-02-26-0900'
assert resp.json['data'][0]['disabled'] is False
assert resp.json['data'][2]['id'] == 'foo-bar:2021-02-26-1000'
assert resp.json['data'][2]['disabled'] is False
resp = app.get(
'/api/agenda/%s/meetings/%s/datetimes/' % (virt_agenda.slug, meeting_type.slug),
params={'exclude_user_external_id': '35'},
)
assert resp.json['data'][0]['id'] == 'foo-bar:2021-02-26-0900'
assert resp.json['data'][0]['disabled'] is False
assert resp.json['data'][2]['id'] == 'foo-bar:2021-02-26-1000'
assert resp.json['data'][2]['disabled'] is False
with CaptureQueriesContext(connection) as ctx:
resp = app.get(
'/api/agenda/%s/meetings/%s/datetimes/' % (virt_agenda.slug, meeting_type.slug),
params={'exclude_user_external_id': '42'},
)
assert len(ctx.captured_queries) == 14
assert resp.json['data'][0]['id'] == 'foo-bar:2021-02-26-0900'
assert resp.json['data'][0]['disabled'] is True
assert resp.json['data'][2]['id'] == 'foo-bar:2021-02-26-1000'
assert resp.json['data'][2]['disabled'] is False
virt_agenda.minimal_booking_delay = None
virt_agenda.maximal_booking_delay = None
virt_agenda.save()
resp = app.get(
'/api/agenda/%s/meetings/%s/datetimes/' % (virt_agenda.slug, meeting_type.slug),
params={'exclude_user_external_id': '42'},
)
assert resp.json['data'][0]['id'] == 'foo-bar:2021-02-26-0900'
assert resp.json['data'][0]['disabled'] is True
assert resp.json['data'][2]['id'] == 'foo-bar:2021-02-26-1000'
assert resp.json['data'][2]['disabled'] is False
@pytest.mark.freeze_time('2017-05-21')
def test_unavailabilitycalendar_meetings_datetimes(app, user):
meetings_agenda = Agenda.objects.create(label='Meeting', kind='meetings', maximal_booking_delay=7)
@ -1079,7 +1721,7 @@ def test_unavailabilitycalendar_meetings_datetimes(app, user):
# 2 slots are gone
with CaptureQueriesContext(connection) as ctx:
resp2 = app.get(datetimes_url)
assert len(ctx.captured_queries) == 10
assert len(ctx.captured_queries) == 11
assert len(resp.json['data']) == len(resp2.json['data']) + 2
# add a standard desk exception

4
tests/api/datetimes/test_virtual.py
View File

@ -515,7 +515,7 @@ def test_virtual_agendas_meetings_datetimes_multiple_agendas(app, mock_now):
with CaptureQueriesContext(connection) as ctx:
resp = app.get(api_url)
assert len(resp.json['data']) == 12
assert len(ctx.captured_queries) == 12
assert len(ctx.captured_queries) == 13
# simulate booking
dt = datetime.datetime.strptime(resp.json['data'][2]['id'].split(':')[1], '%Y-%m-%d-%H%M')
@ -644,7 +644,7 @@ def test_virtual_agendas_meetings_datetimes_exclude_slots(app):
'/api/agenda/%s/meetings/%s/datetimes/' % (virt_agenda.slug, meeting_type.slug),
params={'exclude_user_external_id': '42'},
)
assert len(ctx.captured_queries) == 13
assert len(ctx.captured_queries) == 14
assert resp.json['data'][0]['id'] == 'foo-bar:2021-02-26-0900'
assert resp.json['data'][0]['disabled'] is True
assert resp.json['data'][2]['id'] == 'foo-bar:2021-02-26-1000'

10
tests/api/fillslot/test_all.py
View File

@ -797,7 +797,7 @@ def test_booking_api_available(app, user):
with CaptureQueriesContext(connection) as ctx:
resp = app.get('/api/agenda/%s/datetimes/' % agenda.slug)
assert len(ctx.captured_queries) == 3
assert len(ctx.captured_queries) == 4
event_data = [d for d in resp.json['data'] if d['id'] == event.slug][0]
assert event_data['places']['total'] == 20
assert event_data['places']['available'] == 20
@ -1240,12 +1240,12 @@ def test_agenda_meeting_api_multiple_desk(app, user):
resp = app.post(cancel_url)
with CaptureQueriesContext(connection) as ctx:
resp = app.get('/api/agenda/meetings/%s/datetimes/' % meeting_type.id)
assert len(ctx.captured_queries) == 9
assert len(ctx.captured_queries) == 10
assert len(resp2.json['data']) == len([x for x in resp.json['data'] if not x['disabled']])
with CaptureQueriesContext(connection) as ctx:
resp = app.post('/api/agenda/%s/fillslot/%s/' % (agenda.pk, event_id))
assert len(ctx.captured_queries) == 18
assert len(ctx.captured_queries) == 19
assert resp_booking.json['datetime'] == localtime(Booking.objects.last().event.start_datetime).strftime(
'%Y-%m-%d %H:%M:%S'
@ -1280,11 +1280,11 @@ def test_agenda_meeting_api_multiple_desk(app, user):
booking_url = event_data['api']['fillslot_url']
with CaptureQueriesContext(connection) as ctx:
app.post(booking_url)
assert len(ctx.captured_queries) == 17
assert len(ctx.captured_queries) == 18
with CaptureQueriesContext(connection) as ctx:
app.get('/api/agenda/meetings/%s/datetimes/' % meeting_type.id)
assert len(ctx.captured_queries) == 9
assert len(ctx.captured_queries) == 10
def test_agenda_meeting_same_day(app, mock_now, user):

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

@ -33,7 +33,7 @@ def test_api_events_fillslots(app, user):
params = {'user_external_id': 'user_id', 'check_overlaps': True, 'slots': 'event,event-2'}
with CaptureQueriesContext(connection) as ctx:
resp = app.post_json(fillslots_url, params=params)
assert len(ctx.captured_queries) == 12
assert len(ctx.captured_queries) == 13
assert resp.json['booking_count'] == 2
assert len(resp.json['booked_events']) == 2
assert resp.json['booked_events'][0]['id'] == 'event'

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

@ -163,7 +163,7 @@ def test_api_events_fillslots_multiple_agendas(app, user):
params = {'user_external_id': 'user_id', 'check_overlaps': True, 'slots': event_slugs}
with CaptureQueriesContext(connection) as ctx:
resp = app.post_json('/api/agendas/events/fillslots/?agendas=%s' % agenda_slugs, params=params)
assert len(ctx.captured_queries) == 17
assert len(ctx.captured_queries) == 19
assert resp.json['booking_count'] == 2
assert len(resp.json['booked_events']) == 2
assert resp.json['booked_events'][0]['id'] == 'first-agenda@event'

363
tests/api/fillslot/test_lock_code.py Normal file
View File

@ -0,0 +1,363 @@
# chrono - agendas system
# Copyright (C) 2023 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
import pytest
from chrono.agendas.models import Booking, Lease
from chrono.utils.timezone import now
from tests.utils import build_event_agenda, build_meetings_agenda, build_virtual_agenda
pytestmark = pytest.mark.django_db
def test_meetings_agenda(app, user):
'''Test fillslot on meetings agenda with lock_code'''
agenda = build_meetings_agenda(
'Agenda',
resources=['Re1'],
meeting_types=(30,),
desks={'desk-1': ['monday-friday 9:00-12:00 14:00-17:00']},
)
datetimes_url = agenda._mt_30.api_datetimes_url
# list free slots, with or without a lock
resp = app.get(datetimes_url + '?lock_code=MYLOCK')
free_slots = len(resp.json['data'])
resp = app.get(datetimes_url + '?lock_code=OTHERLOCK')
assert free_slots == len(resp.json['data'])
resp = app.get(datetimes_url)
assert free_slots == len(resp.json['data'])
# lock a slot
fillslot_url = resp.json['data'][2]['api']['fillslot_url']
app.authorization = ('Basic', ('john.doe', 'password'))
app.post_json(fillslot_url, params={'lock_code': 'MYLOCK'})
assert Booking.objects.count() == 1
assert Lease.objects.get().lock_code == 'MYLOCK'
# list free slots: one is locked ...
resp = app.get(datetimes_url)
assert len([x for x in resp.json['data'] if x.get('disabled')]) == 1
resp = app.get(datetimes_url, params={'lock_code': 'OTHERLOCK'})
assert len([x for x in resp.json['data'] if x.get('disabled')]) == 1
# ... unless it's MYLOCK
resp = app.get(datetimes_url, params={'lock_code': 'MYLOCK'})
assert len([x for x in resp.json['data'] if x.get('disabled')]) == 0
# can't lock the same timeslot ...
resp_booking = app.post_json(fillslot_url, params={'lock_code': 'OTHERLOCK'})
assert resp_booking.json['err'] == 1
assert resp_booking.json['reason'] == 'no more desk available'
# ... unless with MYLOCK (aka "relock")
resp_booking = app.post_json(fillslot_url, params={'lock_code': 'MYLOCK'})
assert resp_booking.json['err'] == 0
assert Booking.objects.count() == 1
assert Lease.objects.get().lock_code == 'MYLOCK'
# can't book the slot ...
resp_booking = app.post_json(fillslot_url)
assert resp_booking.json['err'] == 1
assert resp_booking.json['reason'] == 'no more desk available'
resp_booking = app.post_json(fillslot_url, params={'confirm_after_lock': True})
assert resp_booking.json['err'] == 1
assert resp_booking.json['reason'] == 'no more desk available'
resp_booking = app.post_json(fillslot_url, params={'lock_code': 'OTHERLOCK', 'confirm_after_lock': True})
assert resp_booking.json['err'] == 1
assert resp_booking.json['reason'] == 'no more desk available'
# ... unless with MYLOCK (aka "confirm")
resp_booking = app.post_json(fillslot_url, params={'lock_code': 'MYLOCK', 'confirm_after_lock': True})
assert resp_booking.json['err'] == 0
assert Booking.objects.count() == 1
assert Lease.objects.count() == 0
def test_meetings_agenda_expiration(app, user, freezer):
'''Test fillslot on meetings agenda with lock_code'''
agenda = build_meetings_agenda(
'Agenda',
resources=['Re1'],
meeting_types=(30,),
desks={'desk-1': ['monday-friday 9:00-12:00 14:00-17:00']},
)
datetimes_url = agenda._mt_30.api_datetimes_url
# list free slots
resp = app.get(datetimes_url)
# lock a slot
fillslot_url = resp.json['data'][2]['api']['fillslot_url']
app.authorization = ('Basic', ('john.doe', 'password'))
app.post_json(fillslot_url, params={'lock_code': 'MYLOCK'})
assert Booking.objects.count() == 1
assert Lease.objects.get().lock_code == 'MYLOCK'
# list free slots: one is locked ...
resp = app.get(datetimes_url)
assert len([x for x in resp.json['data'] if x.get('disabled')]) == 1
# after 30 minutes it is not locked anymore
freezer.move_to(datetime.timedelta(minutes=30))
resp = app.get(datetimes_url)
assert len([x for x in resp.json['data'] if x.get('disabled')]) == 0
def test_meetings_agenda_with_resource_exclusion(app, user):
'''Test fillslot on meetings agenda with lock_code and ressources'''
agenda1 = build_meetings_agenda(
'Agenda 1',
resources=['Re1'],
meeting_types=(30,),
desks={'desk-1': ['monday-friday 9:00-12:00 14:00-17:00']},
)
agenda2 = build_meetings_agenda(
'Agenda 2',
resources=['Re1'],
meeting_types=(30,),
desks={'desk-1': ['monday-friday 9:00-12:00 14:00-17:00']},
)
resource = agenda2._re_re1
agenda1_datetimes_url = agenda1._mt_30.api_datetimes_url
agenda2_datetimes_url = agenda2._mt_30.api_datetimes_url
# list free slots, with or without a lock
resp = app.get(agenda1_datetimes_url, params={'lock_code': 'OTHERLOCK', 'resources': resource.slug})
free_slots = len(resp.json['data'])
resp = app.get(agenda1_datetimes_url, params={'lock_code': 'MYLOCK', 'resources': resource.slug})
assert free_slots == len(resp.json['data'])
resp = app.get(agenda1_datetimes_url, params={'resources': resource.slug})
assert free_slots == len(resp.json['data'])
resp = app.get(agenda1_datetimes_url)
assert free_slots == len(resp.json['data'])
# lock a slot
fillslot_url = resp.json['data'][2]['api']['fillslot_url']
app.authorization = ('Basic', ('john.doe', 'password'))
app.post_json(fillslot_url + '?resources=re1', params={'lock_code': 'MYLOCK'})
assert Booking.objects.count() == 1
assert Lease.objects.get().lock_code == 'MYLOCK'
# list free slots: one is locked ...
resp = app.get(agenda1_datetimes_url, params={'resources': resource.slug})
assert free_slots == len(resp.json['data'])
assert len([x for x in resp.json['data'] if x.get('disabled')]) == 1
resp = app.get(agenda1_datetimes_url, params={'lock_code': 'OTHERLOCK', 'resources': resource.slug})
assert len([x for x in resp.json['data'] if x.get('disabled')]) == 1
# ... unless it's MYLOCK
resp = app.get(agenda1_datetimes_url, params={'lock_code': 'MYLOCK', 'resources': resource.slug})
assert free_slots == len(resp.json['data'])
assert len([x for x in resp.json['data'] if x.get('disabled')]) == 0
# check slot is also disabled on another agenda with same resource
resp = app.get(agenda2_datetimes_url)
assert len([x for x in resp.json['data'] if x.get('disabled')]) == 0
resp = app.get(agenda2_datetimes_url, params={'resources': resource.slug})
assert len([x for x in resp.json['data'] if x.get('disabled')]) == 1
# can't lock the same timeslot ...
resp_booking = app.post_json(fillslot_url + '?resources=re1', params={'lock_code': 'OTHERLOCK'})
assert resp_booking.json['err'] == 1
assert resp_booking.json['reason'] == 'no more desk available'
# ... unless with MYLOCK (aka "relock")
resp_booking = app.post_json(fillslot_url + '?resources=re1', params={'lock_code': 'MYLOCK'})
assert resp_booking.json['err'] == 0
assert Booking.objects.count() == 1
assert Lease.objects.get().lock_code == 'MYLOCK'
# can't book the slot ...
resp_booking = app.post_json(fillslot_url + '?resources=re1')
assert resp_booking.json['err'] == 1
assert resp_booking.json['reason'] == 'no more desk available'
resp_booking = app.post_json(fillslot_url + '?resources=re1', params={'confirm_after_lock': True})
assert resp_booking.json['err'] == 1
assert resp_booking.json['reason'] == 'no more desk available'
resp_booking = app.post_json(
fillslot_url + '?resources=re1', params={'lock_code': 'OTHERLOCK', 'confirm_after_lock': True}
)
assert resp_booking.json['err'] == 1
assert resp_booking.json['reason'] == 'no more desk available'
# unless with MYLOCK (aka "confirm")
resp_booking = app.post_json(
fillslot_url + '?resources=re1', params={'lock_code': 'MYLOCK', 'confirm_after_lock': True}
)
assert resp_booking.json['err'] == 0
assert Booking.objects.count() == 1
assert Lease.objects.count() == 0
def test_virtual_agenda_with_external_user_id_exclusion(app, user):
'''Test lock_code use when excluding an external_user_id'''
agenda = build_virtual_agenda(
meeting_types=(30,),
agendas={
'Agenda 1': {
'desks': {
'desk': 'monday-friday 08:00-12:00 14:00-17:00',
},
},
'Agenda 2': {
'desks': {
'desk': 'monday-friday 09:00-12:00',
},
},
'Agenda 3': {
'desks': {
'desk': 'monday-friday 15:00-17:00',
},
},
},
)
datetimes_url = agenda._mt_30.api_datetimes_url
resp = app.get(datetimes_url)
slots = resp.json['data']
# get first slot between 11 and 11:30
slot = [slot for slot in slots if ' 11:00:00' in slot['datetime']][0]
fillslot_url = slot['api']['fillslot_url']
app.authorization = ('Basic', ('john.doe', 'password'))
resp = app.post_json(fillslot_url, params={'lock_code': 'MYLOCK', 'user_external_id': 'abcd'})
assert resp.json['err'] == 0
# check the lease was created
assert Booking.objects.filter(user_external_id='abcd', lease__lock_code='MYLOCK').count() == 1
# check 11:00 slot is still available
resp = app.get(datetimes_url)
slots = resp.json['data']
assert any(
s['datetime'] == slot['datetime'] for s in slots if not s['disabled']
), f"slot {slot['datetime']} should be available"
# check 11:00 slot is unavailable when tested with user_external_id
resp = app.get(datetimes_url, params={'user_external_id': 'abcd'})
slots = resp.json['data']
assert not any(
s['datetime'] == slot['datetime'] for s in slots if not s['disabled']
), f"slot {slot['datetime']} should not be available"
# check 11:00 slot is available if tested with user_external_id *AND* lock_code
resp = app.get(datetimes_url, params={'lock_code': 'MYLOCK', 'user_external_id': 'abcd'})
slots = resp.json['data']
assert any(
s['datetime'] == slot['datetime'] for s in slots if not s['disabled']
), f"slot {slot['datetime']} should be available"
def test_events_agenda_no_waiting_list(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.api_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.api_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.api_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.api_datetimes_url)
slot = resp.json['data'][0]
assert slot['places']['available'] == 0
assert slot['places']['full'] is True
def test_events_agenda_no_waiting_list_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.api_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.api_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))
resp = app.get(agenda.api_datetimes_url)
slot = resp.json['data'][0]
assert slot['places']['available'] == 1
assert slot['places']['full'] is False

2
tests/api/test_booking.py
View File

@ -143,7 +143,7 @@ def test_bookings_api(app, user):
with CaptureQueriesContext(connection) as ctx:
resp = app.get('/api/bookings/', params={'user_external_id': 'enfant-1234'})
assert len(ctx.captured_queries) == 2
assert len(ctx.captured_queries) == 3
assert resp.json['err'] == 0
assert resp.json['data'] == [

2
tests/api/test_event.py
View File

@ -1407,7 +1407,7 @@ def test_events_check_status_events(app, user):
}
with CaptureQueriesContext(connection) as ctx:
resp = app.get(url, params=params)
assert len(ctx.captured_queries) == 5
assert len(ctx.captured_queries) == 6
assert resp.json['err'] == 0
assert resp.json['data'] == [
{

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,