0001-api-add-lock_code-parameter-to-fillslot-and-datetime.patch

Benjamin Dauvergne, 08 juin 2021 15:17

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

Subject: [PATCH 1/9] api: add lock_code parameter to fillslot and datetimes
 (#17685)

 chrono/agendas/migrations/0088_lease.py |  64 +++++++
 chrono/agendas/models.py                |  16 ++
 chrono/api/views.py                     | 142 +++++++++++++--
 chrono/settings.py                      |   3 +
 tests/api/test_locks.py                 | 231 ++++++++++++++++++++++++
 tests/api/test_meetings_datetimes.py    |  10 +-
 tests/test_locks.py                     | 102 +++++++++++
 7 files changed, 547 insertions(+), 21 deletions(-)
 create mode 100644 chrono/agendas/migrations/0088_lease.py
 create mode 100644 tests/api/test_locks.py
 create mode 100644 tests/test_locks.py

     # Generated by Django 2.2.19 on 2021-03-16 13:44
     import django.db.models.deletion
     from django.db import migrations, models
     sql_forwards = """
     ALTER TABLE agendas_lease
     ADD CONSTRAINT lease_desk_constraint
     EXCLUDE USING GIST(desk_id WITH =, tstzrange(start_datetime, end_datetime) WITH &&)
         WHERE (desk_id IS NOT NULL);
     ALTER TABLE agendas_lease
     ADD CONSTRAINT lease_resource_constraint
     EXCLUDE USING GIST(resource_id WITH =, tstzrange(start_datetime, end_datetime) WITH &&)
         WHERE (resource_id IS NOT NULL);
     """
     sql_backwards = """
     ALTER TABLE agendas_lease DROP CONSTRAINT lease_desk_constraint;
     ALTER TABLE agendas_lease DROP CONSTRAINT lease_resource_constraint;
     """
     class Migration(migrations.Migration):
         dependencies = [
             ('agendas', '0087_booking_user_name'),
+        ]
         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')),
                     ('lock_expiration_datetime', models.DateTimeField(verbose_name='Lock expiration time')),
                     ('start_datetime', models.DateTimeField(verbose_name='Start')),
                     ('end_datetime', models.DateTimeField(verbose_name='End')),
+                    (
                         'agenda',
                         models.ForeignKey(
                             null=True, on_delete=django.db.models.deletion.CASCADE, to='agendas.Agenda'
                         ),
                     ),
+                    (
                         'desk',
                         models.ForeignKey(
                             null=True, on_delete=django.db.models.deletion.CASCADE, to='agendas.Desk'
                         ),
                     ),
+                    (
                         'resource',
                         models.ForeignKey(
                             null=True, on_delete=django.db.models.deletion.CASCADE, to='agendas.Resource'
                         ),
                     ),
                 ],
                 options={
                     'index_together': {('start_datetime', 'end_datetime')},
                 },
             ),
             migrations.RunSQL(sql=sql_forwards, reverse_sql=sql_backwards),
+        ]

         @property
         def base_slug(self):
             return slugify(self.label)
     class Lease(models.Model):
         desk = models.ForeignKey(Desk, null=True, on_delete=models.CASCADE)
         resource = models.ForeignKey(Resource, null=True, on_delete=models.CASCADE)
         agenda = models.ForeignKey(Agenda, null=True, on_delete=models.CASCADE)
         lock_code = models.CharField(_('Lock code'), max_length=64, blank=False)
         lock_expiration_datetime = models.DateTimeField(_('Lock expiration time'))
         start_datetime = models.DateTimeField(_('Start'))
         end_datetime = models.DateTimeField(_('End'))
         class Meta:
             index_together = (('start_datetime', 'end_datetime'),)
         def as_interval(self):
             return Interval(self.start_datetime, self.end_datetime)

     import itertools
     import uuid
     from django.db import transaction
     from django.conf import settings
     from django.db import IntegrityError, transaction
     from django.db.models import Count, Prefetch, Q
     from django.db.models.functions import TruncDay
     from django.http import Http404, HttpResponse
-...
         Category,
         Desk,
         Event,
         Lease,
         MeetingType,
         TimePeriodException,
+    )
-...
         start_datetime=None,
         end_datetime=None,
         excluded_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.
-...
           min/max_datetime; for each time slot check its status in the exclusion
           and bookings sets.
           If it is excluded, ignore it completely.
           It if is booked, report the slot as full.
           If it 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 :
-...
                 for event_start_datetime, event_duration in booked_events
+            )
         # delete old locks
         Lease.objects.filter(lock_expiration_datetime__lt=now()).delete()
         # aggregate non-expired locked time slots
         desk_locked_intervals = collections.defaultdict(lambda: IntervalSet())
         resource_locked_intervals = IntervalSet()
         q = Q(agenda__in=agendas)
         if resources:
             q |= Q(resource__in=resources)
         for lock in (
             Lease.objects
             # only lock related to on of the agenda or the resource
             .filter(q)
             .exclude(lock_code=lock_code)
             .order_by('start_datetime', 'end_datetime')
         ):
             if lock.desk:
                 desk_locked_intervals[lock.desk_id].add(lock.start_datetime, lock.end_datetime)
             if resources and lock.resource:
                 resource_locked_intervals.add(lock.start_datetime, lock.end_datetime)
         unique_booked = {}
         for time_period in base_agenda.get_effective_time_periods():
             duration = (
-...
                         # slot is full if an already booked event overlaps it
                         # check resources first
                         booked = resources_bookings.overlaps(start_datetime, end_datetime)
                         booked = False
                         if resources:
                             if not booked:
                                 booked = resources_bookings.overlaps(start_datetime, end_datetime)
                             if not booked:
                                 booked = resource_locked_intervals.overlaps(start_datetime, end_datetime)
                         # then check user boookings
                         if not booked:
                             booked = user_bookings.overlaps(start_datetime, end_datetime)
-...
                             booked = desk.id in bookings and bookings[desk.id].overlaps(
                                 start_datetime, end_datetime
+                            )
                         # then locks
                         if not booked and desk.id in desk_locked_intervals:
                             booked = desk_locked_intervals[desk.id].overlaps(start_datetime, end_datetime)
                         if unique and unique_booked.get(timestamp) is booked:
                             continue
                         unique_booked[timestamp] = booked
-...
             start_datetime, end_datetime = get_start_and_end_datetime_from_request(request)
             user_external_id = request.GET.get('exclude_user_external_id') or None
             lock_code = request.GET.get('lock_code', None)
             if lock_code == '':
                 raise APIError(
                     _('lock_code must not be empty'),
                     err_class='lock_code must not be empty',
                     http_status=status.HTTP_400_BAD_REQUEST,
+                )
             # 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
-...
                         start_datetime=start_datetime,
                         end_datetime=end_datetime,
                         excluded_user_external_id=user_external_id,
                         lock_code=lock_code,
+                    )
+                )
                 for slot in sorted(all_slots, key=lambda slot: slot[:3]):
-...
         force_waiting_list = serializers.BooleanField(default=False)
         use_color_for = serializers.CharField(max_length=250, allow_blank=True)
         lock_code = serializers.CharField(max_length=64, required=False, allow_blank=True)
         lock_duration = serializers.IntegerField(
             min_value=0, default=lambda: settings.CHRONO_LOCK_DURATION
         )  # in seconds
         confirm_after_lock = serializers.BooleanField(default=False)
     class StringOrListField(serializers.ListField):
         def to_internal_value(self, data):
-...
+                )
             payload = serializer.validated_data
             lock_code = payload.get('lock_code')
             if lock_code == '':  # lock_code should be absent or a non-empty string
                 raise APIError(
                     _('lock_code cannot be empty'),
                     err_class='invalid payload',
                     errors=serializer.errors,
                     http_status=status.HTTP_400_BAD_REQUEST,
+                )
             if 'slots' in payload:
                 slots = payload['slots']
             if not slots:
-...
                         meeting_type,
                         resources=resources,
                         excluded_user_external_id=user_external_id if exclude_user else None,
                         lock_code=lock_code,
                     ),
                     key=lambda slot: slot.start_datetime,
+                )
-...
                 # booking requires real Event objects (not lazy Timeslots);
                 # create them now, with data from the slots and the desk we found.
                 events = []
                 for start_datetime in datetimes:
                     event = Event.objects.create(
                         agenda=available_desk.agenda,
                         slug=str(uuid.uuid4()),  # set slug to avoid queries during slug generation
                         meeting_type=meeting_type,
                         start_datetime=start_datetime,
                         full=False,
                         places=1,
                         desk=available_desk,
+                    )
                     if resources:
                         event.resources.add(*resources)
                     events.append(event)
                 if not lock_code or payload.get('confirm_after_lock'):
                     for start_datetime in datetimes:
                         event = Event.objects.create(
                             agenda=available_desk.agenda,
                             slug=str(uuid.uuid4()),  # set slug to avoid queries during slug generation
                             meeting_type=meeting_type,
                             start_datetime=start_datetime,
                             full=False,
                             places=1,
                             desk=available_desk,
+                        )
                         if resources:
                             event.resources.add(*resources)
                         events.append(event)
                 else:
                     # remove existing locks
                     Lease.objects.filter(lock_code=lock_code).delete()
                     # create new locks
                     lock_duration = payload.get('lock_duration')
                     if not lock_duration or lock_duration < settings.CHRONO_LOCK_DURATION:
                         lock_duration = settings.CHRONO_LOCK_DURATION
                     lock_expiration_datetime = now() + datetime.timedelta(seconds=lock_duration)
                     meeting_duration = datetime.timedelta(minutes=meeting_type.duration)
                     locks = []
                     for start_datetime in datetimes:
                         locks.append(
                             Lease(
                                 desk=available_desk,
                                 agenda=available_desk.agenda,
                                 lock_code=lock_code,
                                 lock_expiration_datetime=lock_expiration_datetime,
                                 start_datetime=start_datetime,
                                 end_datetime=start_datetime + meeting_duration,
+                            )
+                        )
                         for resource in resources:
                             locks.append(
                                 Lease(
                                     resource=resource,
                                     lock_code=lock_code,
                                     lock_expiration_datetime=lock_expiration_datetime,
                                     start_datetime=start_datetime,
                                     end_datetime=start_datetime + meeting_duration,
+                                )
+                            )
                     try:
                         with transaction.atomic():
                             Lease.objects.bulk_create(locks)
                     except IntegrityError:
                         raise APIError(
                             _('no more desk available'),
                             err_class='no more desk available',
+                        )
                     else:
                         return Response({'err': 0})
             else:
                 if lock_code:
                     raise APIError(
                         _('lock_code does not work with events'),
                         err_class='lock_code does not work with events',
                         http_status=status.HTTP_400_BAD_REQUEST,
+                    )
                 # convert event recurrence identifiers to real event slugs
                 for i, slot in enumerate(slots.copy()):
                     if ':' not in slot:
-...
                     cancelled_booking_id = to_cancel_booking.pk
                     to_cancel_booking.cancel()
                 if lock_code:
                     Lease.objects.filter(lock_code=lock_code).delete()
                 # now we have a list of events, book them.
                 primary_booking = None
                 for event in events:

     # (see http://docs.python-requests.org/en/master/user/advanced/#proxies)
     REQUESTS_PROXIES = None
     # default lock duration, in seconds
     CHRONO_LOCK_DURATION = 10 * 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

     import urllib.parse as urlparse
     import pytest
     from chrono.agendas.models import Booking, Lease, MeetingType, Resource
     pytestmark = pytest.mark.django_db
     def test_booking_api_meeting_lock_and_confirm(app, meetings_agenda, user):
         agenda_id = meetings_agenda.slug
         meeting_type = MeetingType.objects.get(agenda=meetings_agenda)
         # list free slots, with or without a lock
         resp = app.get('/api/agenda/meetings/%s/datetimes/' % meeting_type.id)
         free_slots = len(resp.json['data'])
         resp = app.get('/api/agenda/meetings/%s/datetimes/?lock_code=OTHERLOCK' % meeting_type.id)
         assert free_slots == len(resp.json['data'])
         resp = app.get('/api/agenda/meetings/%s/datetimes/?lock_code=MYLOCK' % meeting_type.id)
         assert free_slots == len(resp.json['data'])
         # lock a slot
         event_id = resp.json['data'][2]['id']
         assert urlparse.urlparse(
             resp.json['data'][2]['api']['fillslot_url']
         ).path == '/api/agenda/%s/fillslot/%s/' % (meetings_agenda.slug, event_id)
         app.authorization = ('Basic', ('john.doe', 'password'))
         app.post('/api/agenda/%s/fillslot/%s/' % (agenda_id, event_id), params={'lock_code': 'MYLOCK'})
         assert Booking.objects.count() == 0
         assert Lease.objects.count() == 1
         assert (
             Lease.objects.filter(
                 agenda=meetings_agenda,
                 desk=meetings_agenda.desk_set.get(),
                 lock_code='MYLOCK',
                 lock_expiration_datetime__isnull=False,
             ).count()
             == 1
+        )
         # list free slots: one is locked ...
         resp2 = app.get('/api/agenda/meetings/%s/datetimes/' % meeting_type.id)
         assert free_slots == len([x for x in resp2.json['data']])
         assert len([x for x in resp2.json['data'] if x.get('disabled')]) == 1
         resp2 = app.get('/api/agenda/meetings/%s/datetimes/?lock_code=OTHERLOCK' % meeting_type.id)
         assert free_slots == len([x for x in resp2.json['data']])
         assert len([x for x in resp2.json['data'] if x.get('disabled')]) == 1
         # ... unless it's MYLOCK
         resp2 = app.get('/api/agenda/meetings/%s/datetimes/?lock_code=MYLOCK' % meeting_type.id)
         assert free_slots == len([x for x in resp2.json['data']])
         assert len([x for x in resp2.json['data'] if x.get('disabled')]) == 0
         # can't lock the same timeslot ...
         resp_booking = app.post(
             '/api/agenda/%s/fillslot/%s/' % (agenda_id, event_id), 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(
             '/api/agenda/%s/fillslot/%s/' % (agenda_id, event_id), params={'lock_code': 'MYLOCK'}
+        )
         assert resp_booking.json['err'] == 0
         assert Booking.objects.count() == 0
         assert Lease.objects.count() == 1
         assert (
             Lease.objects.filter(
                 agenda=meetings_agenda,
                 desk=meetings_agenda.desk_set.get(),
                 lock_code='MYLOCK',
                 lock_expiration_datetime__isnull=False,
             ).count()
             == 1
+        )
         # can't book the slot ...
         resp_booking = app.post('/api/agenda/%s/fillslot/%s/' % (agenda_id, event_id))
         assert resp_booking.json['err'] == 1
         assert resp_booking.json['reason'] == 'no more desk available'
         resp_booking = app.post(
             '/api/agenda/%s/fillslot/%s/' % (agenda_id, event_id), params={'confirm_after_lock': True}
+        )
         assert resp_booking.json['err'] == 1
         assert resp_booking.json['reason'] == 'no more desk available'
         resp_booking = app.post(
             '/api/agenda/%s/fillslot/%s/' % (agenda_id, event_id),
             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(
             '/api/agenda/%s/fillslot/%s/' % (agenda_id, event_id),
             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_booking_api_meeting_lock_and_confirm_with_resource(app, meetings_agenda, user):
         resource1 = Resource.objects.create(label='Resource 1', slug='re1')
         resource2 = Resource.objects.create(label='Resource 2', slug='re2')
         meetings_agenda.resources.add(resource1, resource2)
         agenda_id = meetings_agenda.slug
         meeting_type = MeetingType.objects.get(agenda=meetings_agenda)
         # list free slots, with or without a lock
         resp = app.get('/api/agenda/meetings/%s/datetimes/?resources=re1' % meeting_type.id)
         free_slots = len(resp.json['data'])
         resp = app.get('/api/agenda/meetings/%s/datetimes/?lock_code=OTHERLOCK&resources=re1' % meeting_type.id)
         assert free_slots == len(resp.json['data'])
         resp = app.get('/api/agenda/meetings/%s/datetimes/?lock_code=MYLOCK&resources=re1' % meeting_type.id)
         assert free_slots == len(resp.json['data'])
         # lock a slot
         event_id = resp.json['data'][2]['id']
         assert urlparse.urlparse(
             resp.json['data'][2]['api']['fillslot_url']
         ).path == '/api/agenda/%s/fillslot/%s/' % (meetings_agenda.slug, event_id)
         app.authorization = ('Basic', ('john.doe', 'password'))
         app.post(
             '/api/agenda/%s/fillslot/%s/?resources=re1' % (agenda_id, event_id), params={'lock_code': 'MYLOCK'}
+        )
         assert Booking.objects.count() == 0
         assert Lease.objects.count() == 2
         assert (
             Lease.objects.filter(
                 agenda=meetings_agenda,
                 desk=meetings_agenda.desk_set.get(),
                 resource__isnull=True,
                 lock_code='MYLOCK',
                 lock_expiration_datetime__isnull=False,
             ).count()
             == 1
+        )
         assert (
             Lease.objects.filter(
                 agenda__isnull=True,
                 desk__isnull=True,
                 resource=resource1,
                 lock_code='MYLOCK',
                 lock_expiration_datetime__isnull=False,
             ).count()
             == 1
+        )
         old_lock_ids = set(Lease.objects.values_list('id', flat=True))
         # list free slots: one is locked ...
         resp2 = app.get('/api/agenda/meetings/%s/datetimes/?resources=re1' % meeting_type.id)
         assert free_slots == len([x for x in resp2.json['data']])
         assert len([x for x in resp2.json['data'] if x.get('disabled')]) == 1
         resp2 = app.get('/api/agenda/meetings/%s/datetimes/?lock_code=OTHERLOCK&resources=re1' % meeting_type.id)
         assert free_slots == len([x for x in resp2.json['data']])
         assert len([x for x in resp2.json['data'] if x.get('disabled')]) == 1
         # ... unless it's MYLOCK
         resp2 = app.get('/api/agenda/meetings/%s/datetimes/?lock_code=MYLOCK&resources=re1' % meeting_type.id)
         assert free_slots == len([x for x in resp2.json['data']])
         assert len([x for x in resp2.json['data'] if x.get('disabled')]) == 0
         # can't lock the same timeslot ...
         resp_booking = app.post(
             '/api/agenda/%s/fillslot/%s/?resources=re1' % (agenda_id, event_id), 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(
             '/api/agenda/%s/fillslot/%s/?resources=re1' % (agenda_id, event_id), params={'lock_code': 'MYLOCK'}
+        )
         assert resp_booking.json['err'] == 0
         assert Booking.objects.count() == 0
         assert Lease.objects.count() == 2
         assert (
             Lease.objects.filter(
                 agenda=meetings_agenda,
                 desk=meetings_agenda.desk_set.get(),
                 lock_code='MYLOCK',
                 lock_expiration_datetime__isnull=False,
             ).count()
             == 1
+        )
         assert (
             Lease.objects.filter(
                 agenda__isnull=True,
                 desk__isnull=True,
                 resource=resource1,
                 lock_code='MYLOCK',
                 lock_expiration_datetime__isnull=False,
             ).count()
             == 1
+        )
         new_lock_ids = set(Lease.objects.values_list('id', flat=True))
         assert not (old_lock_ids & new_lock_ids)
         # can't book the slot ...
         resp_booking = app.post('/api/agenda/%s/fillslot/%s/?resources=re1' % (agenda_id, event_id))
         assert resp_booking.json['err'] == 1
         assert resp_booking.json['reason'] == 'no more desk available'
         resp_booking = app.post(
             '/api/agenda/%s/fillslot/%s/?resources=re1' % (agenda_id, event_id),
             params={'confirm_after_lock': True},
+        )
         assert resp_booking.json['err'] == 1
         assert resp_booking.json['reason'] == 'no more desk available'
         resp_booking = app.post(
             '/api/agenda/%s/fillslot/%s/?resources=re1' % (agenda_id, event_id),
             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(
             '/api/agenda/%s/fillslot/%s/?resources=re1' % (agenda_id, event_id),
             params={'lock_code': 'MYLOCK', 'confirm_after_lock': True},
+        )
         assert resp_booking.json['err'] == 0
         assert Booking.objects.count() == 1
         assert Lease.objects.count() == 0

+        )
         with CaptureQueriesContext(connection) as ctx:
             resp = app.get(api_url)
             assert len(ctx.captured_queries) == 10
             assert len(ctx.captured_queries) == 12
         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,
-...
                 '/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) == 11
         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'
-...
         with CaptureQueriesContext(connection) as ctx:
             resp = app.get(api_url)
             assert len(resp.json['data']) == 12
             assert len(ctx.captured_queries) == 10
             assert len(ctx.captured_queries) == 12
         # simulate booking
         dt = datetime.datetime.strptime(resp.json['data'][2]['id'].split(':')[1], '%Y-%m-%d-%H%M')
-...
                 '/api/agenda/%s/meetings/%s/datetimes/' % (virt_agenda.slug, meeting_type.slug),
                 params={'exclude_user_external_id': '42'},
+            )
             assert len(ctx.captured_queries) == 11
             assert len(ctx.captured_queries) == 13
         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'
-...
         # 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) == 12
         assert len(resp.json['data']) == len(resp2.json['data']) + 2
         # add a standard desk exception

     import datetime
     from argparse import Namespace
     import pytest
     from django.db import IntegrityError, transaction
     from django.utils.timezone import now
     from chrono.agendas.models import Agenda, Desk, Lease, MeetingType, Resource
     @pytest.fixture
     def lock(db):
         agenda = Agenda.objects.create(
             label=u'Foo bar Meeting', kind='meetings', minimal_booking_delay=1, maximal_booking_delay=56
+        )
         meeting_type = MeetingType(agenda=agenda, label='Blah', duration=30)
         meeting_type.save()
         desk1 = Desk.objects.create(agenda=agenda, label='Desk 1')
         desk2 = Desk.objects.create(agenda=agenda, label='Desk 2')
         resource = Resource.objects.create(label='re', description='re')
         return Namespace(**locals())
     def test_lock_constraint_desk(lock):
         Lease.objects.create(
             agenda=lock.agenda,
             desk=lock.desk1,
             lock_code='1',
             lock_expiration_datetime=now() + datetime.timedelta(minutes=5),
             start_datetime=now(),
             end_datetime=now() + datetime.timedelta(minutes=5),
+        )
         Lease.objects.create(
             agenda=lock.agenda,
             desk=lock.desk2,
             lock_code='2',
             lock_expiration_datetime=now() + datetime.timedelta(minutes=5),
             start_datetime=now(),
             end_datetime=now() + datetime.timedelta(minutes=5),
+        )
         Lease.objects.create(
             resource=lock.resource,
             lock_code='3',
             lock_expiration_datetime=now() + datetime.timedelta(minutes=5),
             start_datetime=now(),
             end_datetime=now() + datetime.timedelta(minutes=5),
+        )
         with pytest.raises(IntegrityError):
             # prevent IntegrityError to break the current transaction
             with transaction.atomic():
                 Lease.objects.create(
                     agenda=lock.agenda,
                     desk=lock.desk1,
                     lock_code='4',
                     lock_expiration_datetime=now() + datetime.timedelta(minutes=5),
                     # interval overlaps interval of first lock
                     start_datetime=now() + datetime.timedelta(minutes=4),
                     end_datetime=now() + datetime.timedelta(minutes=6),
+                )
     def test_lock_constraint_resource(lock):
         Lease.objects.create(
             agenda=lock.agenda,
             desk=lock.desk1,
             lock_code='1',
             lock_expiration_datetime=now() + datetime.timedelta(minutes=5),
             start_datetime=now(),
             end_datetime=now() + datetime.timedelta(minutes=5),
+        )
         Lease.objects.create(
             agenda=lock.agenda,
             desk=lock.desk2,
             lock_code='2',
             lock_expiration_datetime=now() + datetime.timedelta(minutes=5),
             start_datetime=now(),
             end_datetime=now() + datetime.timedelta(minutes=5),
+        )
         Lease.objects.create(
             resource=lock.resource,
             lock_code='3',
             lock_expiration_datetime=now() + datetime.timedelta(minutes=5),
             start_datetime=now(),
             end_datetime=now() + datetime.timedelta(minutes=5),
+        )
         with pytest.raises(IntegrityError):
             # prevent IntegrityError to break the current transaction
             with transaction.atomic():
                 Lease.objects.create(
                     resource=lock.resource,
                     lock_code='4',
                     lock_expiration_datetime=now() + datetime.timedelta(minutes=5),
                     # interval overlaps interval of first lock
                     start_datetime=now() + datetime.timedelta(minutes=4),
                     end_datetime=now() + datetime.timedelta(minutes=6),
+                )
+    -

Projet

Général

Profil

Produits Entr'ouvert » Chrono

0001-api-add-lock_code-parameter-to-fillslot-and-datetime.patch