Produits Entr'ouvert » Chrono

# -*- coding: utf-8 -*-

# Generated by Django 1.11.18 on 2021-01-26 13:10

from __future__ import unicode_literals

from django.db import migrations, models

import django.db.models.deletion

import jsonfield.fields

class Migration(migrations.Migration):

    dependencies = [

        ('agendas', '0073_auto_20210125_1800'),

    ]

    operations = [

        migrations.AddField(

            model_name='event',

            name='primary_event',

            field=models.ForeignKey(

                null=True,

                on_delete=django.db.models.deletion.CASCADE,

                related_name='recurrences',

                to='agendas.Event',

            ),

        ),

        migrations.AddField(

            model_name='event',

            name='recurrence_rule',

            field=jsonfield.fields.JSONField(null=True, verbose_name='Recurrence rule'),

        ),

        migrations.AddField(

            model_name='event',

            name='repeat',

            field=models.CharField(

                blank=True,

                choices=[

                    ('daily', 'Daily'),

                    ('weekly', 'Weekly'),

                    ('2-weeks', 'Once every two weeks'),

                    ('weekdays', 'Every weekdays (Monday to Friday)'),

                ],

                max_length=16,

                verbose_name='Repeat',

            ),

        ),

    ]

from dateutil.rrule import rrule, rruleset, DAILY, WEEKLY

from dateutil.relativedelta import relativedelta

from django.utils.timezone import localtime, now, make_aware, make_naive, is_aware, utc

            agenda['events'] = [x.export_json() for x in self.event_set.filter(primary_event__isnull=True)]

        exclude_full=False,

            # recurring events are never opened

            entries = self.event_set.filter(recurrence_rule__isnull=True)

            # exclude canceled events except for event recurrences

            entries = entries.filter(Q(cancelled=False) | Q(primary_event__isnull=False))

            if exclude_full:

        if max_start:

            entries = self.add_event_recurrences(

                entries,

                min_start or localtime(now()),

                max_start,

                exclude_full=exclude_full,

                prefetched_queryset=prefetched_queryset,

            )

    def add_event_recurrences(

        self,

        events,

        min_start,

        max_start,

        exclude_full=False,

        exclude_cancelled=True,

        prefetched_queryset=False,

    ):

        excluded_datetimes = [make_naive(event.start_datetime) for event in events]

        events = [

            e for e in events if not (e.cancelled and exclude_cancelled) and not (exclude_full and e.full)

        ]

        if prefetched_queryset:

            recurring_events = self.prefetched_recurring_events

        else:

            recurring_events = self.event_set.filter(recurrence_rule__isnull=False)

        for event in recurring_events:

            events.extend(event.get_recurrences(min_start, max_start, excluded_datetimes))

        events.sort(key=lambda x: [getattr(x, field) for field in Event._meta.ordering])

        return events

    REPEAT_CHOICES = [

        ('daily', _('Daily')),

        ('weekly', _('Weekly')),

        ('2-weeks', _('Once every two weeks')),

        ('weekdays', _('Every weekdays (Monday to Friday)')),

    ]

    repeat = models.CharField(_('Repeat'), max_length=16, blank=True, choices=REPEAT_CHOICES)

    recurrence_rule = JSONField(_('Recurrence rule'), null=True)

    primary_event = models.ForeignKey('self', null=True, on_delete=models.CASCADE, related_name='recurrences')

        self.recurrence_rule = self.get_recurrence_rule()

            'repeat': self.repeat,

            'recurrence_rule': self.recurrence_rule,

    def get_or_create_event_recurrence(self, start_datetime):

        events = self.get_recurrences(start_datetime, start_datetime)

        if len(events) == 0:

            raise ValueError('No event recurrence found for specified datetime.')

        elif len(events) > 1:  # should not happen

            raise ValueError('Multiple events found for specified datetime.')

        event = events[0]

        event.slug = event.slug.replace(':', '-')

        with transaction.atomic():

            try:

                return Event.objects.get(agenda=self.agenda, slug=event.slug)

            except Event.DoesNotExist:

                event.save()

                return event

    def get_recurrences(self, min_datetime, max_datetime, excluded_datetimes=None):

        recurrences = []

        rrule_set = rruleset()

        # do not generate recurrences for existing events

        rrule_set._exdate = excluded_datetimes or []

        event_base = Event(

            agenda=self.agenda,

            primary_event=self,

            slug=self.slug,

            duration=self.duration,

            places=self.places,

            waiting_list_places=self.waiting_list_places,

            label=self.label,

            description=self.description,

            pricing=self.pricing,

            url=self.url,

        )

        # remove pytz info because dateutil doesn't support DST changes

        min_datetime = make_naive(min_datetime)

        max_datetime = make_naive(max_datetime)

        rrule_set.rrule(rrule(dtstart=make_naive(self.start_datetime), **self.recurrence_rule))

        for start_datetime in rrule_set.between(min_datetime, max_datetime, inc=True):

            event = copy.copy(event_base)

            # add timezone back

            aware_start_datetime = make_aware(start_datetime)

            event.slug = '%s:%s' % (event.slug, aware_start_datetime.strftime('%Y-%m-%d-%H%M'))

            event.start_datetime = aware_start_datetime.astimezone(utc)

            recurrences.append(event)

        return recurrences

    def get_recurrence_display(self):

        repeat = str(self.get_repeat_display())

        time = date_format(localtime(self.start_datetime), 'TIME_FORMAT')

        if self.repeat in ('weekly', '2-weeks'):

            day = date_format(localtime(self.start_datetime), 'l')

            return _('%(every_x_days)s on %(day)s at %(time)s') % {

                'every_x_days': repeat,

                'day': day,

                'time': time,

            }

        else:

            return _('%(every_x_days)s at %(time)s') % {'every_x_days': repeat, 'time': time}

    def get_recurrence_rule(self):

        rrule = {}

        if self.repeat == 'daily':

            rrule['freq'] = DAILY

        elif self.repeat == 'weekly':

            rrule['freq'] = WEEKLY

            rrule['byweekday'] = [self.start_datetime.weekday()]

        elif self.repeat == '2-weeks':

            rrule['freq'] = WEEKLY

            rrule['byweekday'] = [self.start_datetime.weekday()]

            rrule['interval'] = 2

        elif self.repeat == 'weekdays':

            rrule['freq'] = WEEKLY

            rrule['byweekday'] = [i for i in range(5)]

        else:

            return None

        return rrule

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

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

            agenda_detail['opened_events_available'] = bool(agenda.get_open_events(exclude_full=True))

    if event.label and event.primary_event is not None:

        event.label = '%s (%s)' % (

            event.label,

            date_format(localtime(event.start_datetime), 'DATETIME_FORMAT'),

        )

def get_event_recurrence(agenda, event_identifier):

    event_slug, datetime_str = event_identifier.split(':')

    try:

        start_datetime = make_aware(datetime.datetime.strptime(datetime_str, '%Y-%m-%d-%H%M'))

    except ValueError:

        raise APIError(

            _('bad datetime format: %s') % datetime_str,

            err_class='bad datetime format: %s' % datetime_str,

            http_status=status.HTTP_400_BAD_REQUEST,

        )

    try:

        event = agenda.event_set.get(slug=event_slug)

    except Event.DoesNotExist:

        raise APIError(

            _('unknown recurring event slug: %s') % event_slug,

            err_class='unknown recurring event slug:' % event_slug,

            http_status=status.HTTP_400_BAD_REQUEST,

        )

    try:

        return event.get_or_create_event_recurrence(start_datetime)

    except ValueError:

        raise APIError(

            _('invalid datetime for event %s') % event_identifier,

            err_class='invalid datetime for event %s' % event_identifier,

            http_status=status.HTTP_400_BAD_REQUEST,

        )

            recurring_event_queryset = Event.objects.filter(recurrence_rule__isnull=False)

                ),

                Prefetch(

                    'event_set',

                    queryset=recurring_event_queryset,

                    to_attr='prefetched_recurring_events',

                ),

            # convert event recurrence identifiers to real event slugs

            for i, slot in enumerate(slots.copy()):

                if not ':' in slot:

                    continue

                event = get_event_recurrence(agenda, slot)

                slots[i] = event.slug

        if ':' in event_identifier:

            return get_event_recurrence(agenda, event_identifier)

        if ':' in event_identifier:

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

            return get_event_recurrence(agenda, event_identifier)

            'repeat',

            'repeat',

                event.full_clean(exclude=['desk', 'meeting_type', 'primary_event'])

    ><a href="{% if settings_view %}{% url 'chrono-manager-event-edit' pk=agenda.pk event_pk=event.pk %}?next=settings{% elif event.pk %}{% url 'chrono-manager-event-view' pk=agenda.pk event_pk=event.pk %}{% else %}{% url 'chrono-manager-event-create-recurrence' pk=agenda.pk event_identifier=event.slug %}{% endif %}">

    {% if not event.repeat %}

    {% else %}

    {{ event.get_recurrence_display }}

    {% endif %}

    url(

        r'^agendas/(?P<pk>\d+)/create_event_recurrence/(?P<event_identifier>[\w:-]+)/$',

        views.event_create_recurrence,

        name='chrono-manager-event-create-recurrence',

    ),

    RedirectView,

            queryset = self.agenda.event_set.filter(recurrence_rule__isnull=True)

    def get_dated_items(self):

        date_list, object_list, extra_context = super().get_dated_items()

        if self.agenda.kind == 'events':

            min_start = make_aware(datetime.datetime.combine(extra_context['month'], datetime.time(0, 0)))

            max_start = make_aware(

                datetime.datetime.combine(extra_context['next_month'], datetime.time(0, 0))

            )

            object_list = self.agenda.add_event_recurrences(

                object_list, min_start, max_start, exclude_cancelled=False

            )

        return date_list, object_list, extra_context

        qs = self.agenda.event_set.filter(primary_event__isnull=True)

class EventCreateRecurrenceView(ManagedAgendaMixin, RedirectView):

    def get_redirect_url(self, *args, **kwargs):

        event_slug, datetime_str = kwargs['event_identifier'].split(':')

        try:

            start_datetime = make_aware(datetime.datetime.strptime(datetime_str, '%Y-%m-%d-%H%M'))

        except ValueError:

            raise Http404()

        event = self.agenda.event_set.get(slug=event_slug)

        try:

            event_recurrence = event.get_or_create_event_recurrence(start_datetime)

        except ValueError:

            raise Http404()

        return event_recurrence.get_absolute_view_url()

event_create_recurrence = EventCreateRecurrenceView.as_view()

def test_recurring_events(freezer):

    freezer.move_to('2021-01-06 12:00')  # Wednesday

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

    event = Event.objects.create(

        agenda=agenda,

        start_datetime=now(),

        repeat='weekly',

        label='Event',

        places=10,

        waiting_list_places=10,

        duration=10,

        description='Description',

        url='https://example.com',

        pricing='10€',

    )

    event.refresh_from_db()

    recurrences = event.get_recurrences(localtime(), localtime() + datetime.timedelta(days=15))

    assert len(recurrences) == 3

    first_event = recurrences[0]

    assert first_event.slug == event.slug + ':2021-01-06-1300'

    event_json = event.export_json()

    first_event_json = first_event.export_json()

    different_fields = ['slug', 'repeat', 'recurrence_rule']

    assert all(first_event_json[k] == event_json[k] for k in event_json if k not in different_fields)

    second_event = recurrences[1]

    assert second_event.start_datetime == first_event.start_datetime + datetime.timedelta(days=7)

    assert second_event.start_datetime.weekday() == first_event.start_datetime.weekday()

    assert second_event.slug == 'event:2021-01-13-1300'

    different_fields = ['slug', 'start_datetime']

    second_event_json = second_event.export_json()

    assert all(first_event_json[k] == second_event_json[k] for k in event_json if k not in different_fields)

    new_recurrences = event.get_recurrences(

        localtime() + datetime.timedelta(days=15),

        localtime() + datetime.timedelta(days=30),

    )

    assert len(recurrences) == 3

    assert new_recurrences[0].start_datetime == recurrences[-1].start_datetime + datetime.timedelta(days=7)

def test_recurring_events_dst(freezer, settings):

    freezer.move_to('2020-10-24 12:00')

    settings.TIME_ZONE = 'Europe/Brussels'

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

    event = Event.objects.create(agenda=agenda, start_datetime=now(), repeat='weekly', places=5)

    event.refresh_from_db()

    dt = localtime()

    recurrences = event.get_recurrences(dt, dt + datetime.timedelta(days=8))

    event_before_dst, event_after_dst = recurrences

    assert event_before_dst.start_datetime.hour + 1 == event_after_dst.start_datetime.hour

    assert event_before_dst.slug == 'agenda-event:2020-10-24-1400'

    assert event_after_dst.slug == 'agenda-event:2020-10-31-1400'

    freezer.move_to('2020-11-24 12:00')

    new_recurrences = event.get_recurrences(dt, dt + datetime.timedelta(days=8))

    new_event_before_dst, new_event_after_dst = new_recurrences

    assert event_before_dst.start_datetime == new_event_before_dst.start_datetime

    assert event_after_dst.start_datetime == new_event_after_dst.start_datetime

    assert event_before_dst.slug == new_event_before_dst.slug

    assert event_after_dst.slug == new_event_after_dst.slug

def test_recurring_events_repeat(freezer):

    freezer.move_to('2021-01-06 12:00')  # Wednesday

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

    event = Event.objects.create(

        agenda=agenda,

        start_datetime=now(),

        repeat='daily',

        places=5,

    )

    event.refresh_from_db()

    start_datetime = localtime(event.start_datetime)

    freezer.move_to('2021-01-06 12:01')  # recurrence on same day should not be returned

    recurrences = event.get_recurrences(

        localtime() + datetime.timedelta(days=1), localtime() + datetime.timedelta(days=7)

    )

    assert len(recurrences) == 6

    assert recurrences[0].start_datetime == start_datetime + datetime.timedelta(days=2)

    assert recurrences[-1].start_datetime == start_datetime + datetime.timedelta(days=7)

    for i in range(len(recurrences) - 1):

        assert recurrences[i].start_datetime + datetime.timedelta(days=1) == recurrences[i + 1].start_datetime

    event.repeat = 'weekdays'

    event.save()

    recurrences = event.get_recurrences(

        localtime() + datetime.timedelta(days=1), localtime() + datetime.timedelta(days=7)

    )

    assert len(recurrences) == 4

    assert recurrences[0].start_datetime == start_datetime + datetime.timedelta(days=2)

    assert recurrences[1].start_datetime == start_datetime + datetime.timedelta(days=5)

    assert recurrences[-1].start_datetime == start_datetime + datetime.timedelta(days=7)

    event.repeat = '2-weeks'

    event.save()

    recurrences = event.get_recurrences(

        localtime() + datetime.timedelta(days=3), localtime() + datetime.timedelta(days=45)

    )

    assert len(recurrences) == 3

    assert recurrences[0].start_datetime == start_datetime + datetime.timedelta(days=14)

    assert recurrences[-1].start_datetime == start_datetime + datetime.timedelta(days=14) * len(recurrences)

    for i in range(len(recurrences) - 1):

        assert (

            recurrences[i].start_datetime + datetime.timedelta(days=14) == recurrences[i + 1].start_datetime

        )

    # event recurrences are available

    event = Event.objects.create(

        start_datetime=now(),

        places=10,

        agenda=event_agenda,

        repeat='daily',

    )

    assert len(event_agenda.get_open_events()) == 2

    resp = app.get('/api/agenda/', params={'with_open_events': '1'})

    assert len(resp.json['data']) == 1

        assert len(ctx.captured_queries) == 4

        assert len(ctx.captured_queries) == 4

def test_recurring_events_api(app, user, freezer):

    freezer.move_to('2021-01-12 12:05')  # Tuesday

    agenda = Agenda.objects.create(

        label='Foo bar', kind='events', minimal_booking_delay=1, maximal_booking_delay=30

    )

    event = Event.objects.create(

        slug='abc', label='Test', start_datetime=localtime(), repeat='weekly', places=5, agenda=agenda

    )

    resp = app.get('/api/agenda/%s/datetimes/' % agenda.slug)

    data = resp.json['data']

    assert len(data) == 4

    assert data[0]['id'] == 'abc:2021-01-19-1305'

    assert data[0]['datetime'] == '2021-01-19 13:05:00'

    assert data[0]['text'] == 'Test (Jan. 19, 2021, 1:05 p.m.)'

    assert data[3]['id'] == 'abc:2021-02-09-1305'

    assert Event.objects.count() == 1

    fillslot_url = data[0]['api']['fillslot_url']

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

    # book first event

    resp = app.post(fillslot_url)

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

    assert Event.objects.count() == 2

    event = Booking.objects.get(pk=resp.json['booking_id']).event

    assert event.slug == 'abc-2021-01-19-1305'

    # first event is now a real event in datetimes

    resp = app.get('/api/agenda/%s/datetimes/' % agenda.slug)

    data = resp.json['data']

    assert len(data) == 4

    assert data[0]['id'] == event.slug

    new_fillslot_url = data[0]['api']['fillslot_url']

    # booking again with both old and new urls works

    resp = app.post(fillslot_url)

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

    resp = app.post(new_fillslot_url)

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

    assert Event.objects.count() == 2

    assert event.booking_set.count() == 3

    # status and bookings api also create a real event

    status_url = data[1]['api']['status_url']

    resp = app.get(status_url)

    assert resp.json['places']['total'] == 5

    assert Event.objects.count() == 3

    bookings_url = data[2]['api']['bookings_url']

    resp = app.get(bookings_url, params={'user_external_id': '42'})

    assert resp.json['data'] == []

    assert Event.objects.count() == 4

    # cancelled recurrences do not appear

    event.cancel()

    resp = app.get('/api/agenda/%s/datetimes/' % agenda.slug)

    assert len(resp.json['data']) == 3

    assert resp.json['data'][0]['id'] == 'abc-2021-01-26-1305'

def test_recurring_events_api_various_times(app, user, mock_now):

    agenda = Agenda.objects.create(

        label='Foo bar', kind='events', minimal_booking_delay=0, maximal_booking_delay=30

    )

    event = Event.objects.create(

        slug='abc', start_datetime=localtime(), repeat='weekly', places=5, agenda=agenda

    )

    event.refresh_from_db()

    resp = app.get('/api/agenda/%s/datetimes/' % agenda.slug)

    assert len(resp.json['data']) == 5

    fillslot_url = resp.json['data'][0]['api']['fillslot_url']

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

    resp = app.post(fillslot_url)

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

    new_event = Booking.objects.get(pk=resp.json['booking_id']).event

    assert event.start_datetime == new_event.start_datetime

from dateutil.rrule import DAILY

def test_import_export_recurring_event(app, freezer):

    freezer.move_to('2021-01-12 12:10')

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

    event = Event.objects.create(

        agenda=agenda,

        start_datetime=now(),

        repeat='daily',

        places=10,

    )

    event.refresh_from_db()

    event.get_or_create_event_recurrence(event.start_datetime + datetime.timedelta(days=3))

    assert Event.objects.count() == 2

    output = get_output_of_command('export_site')

    assert len(json.loads(output)['agendas']) == 1

    import_site(data={}, clean=True)

    with tempfile.NamedTemporaryFile() as f:

        f.write(force_bytes(output))

        f.flush()

        call_command('import_site', f.name)

    assert Agenda.objects.count() == 1

    assert Event.objects.count() == 1

    event = Agenda.objects.get(label='Foo Bar').event_set.first()

    assert event.primary_event is None

    assert event.repeat == 'daily'

    assert event.recurrence_rule == {'freq': DAILY}

        assert len(ctx.captured_queries) == 7

    # add recurring event on every Wednesday

    start_datetime = localtime().replace(day=7, month=10, year=2020)

    event = Event.objects.create(

        label='abc', start_datetime=start_datetime, places=10, agenda=agenda, repeat='weekly'

    )

    resp = app.get('/manage/agendas/%s/%s/%s/' % (agenda.id, 2020, 10))

    assert len(resp.pyquery.find('.event-info')) == 4

    assert 'abc' in resp.pyquery.find('.event-info')[0].text

    # 12/2020 has 5 Wednesday

    resp = app.get('/manage/agendas/%s/%s/%s/' % (agenda.id, 2020, 12))

    assert len(resp.pyquery.find('.event-info')) == 5

    # trying to access event recurrence creates it

    event_count = Event.objects.count()

    time = localtime(event.start_datetime).strftime('%H%M')

    resp = resp.click(href='abc:2020-12-02-%s' % time)

    assert Event.objects.count() == event_count + 1

    Event.objects.get(slug='abc-2020-12-02-%s' % time).cancel()

    resp = app.get('/manage/agendas/%s/%s/%s/' % (agenda.id, 2020, 12))

    assert 'Cancelled' in resp.text

    bad_event_url = '/manage/agendas/%s/create_event_recurrence/abc:2020-10-8-1130/' % agenda.id

    resp = app.get(bad_event_url, status=404)

    # weekly recurring event, first recurrence is in the past but second is in range

    event = Event.objects.create(

        label='event G',

        start_datetime=now() - datetime.timedelta(days=3),

        places=10,

        agenda=agenda,

        repeat='weekly',

    )

    assert resp.text.count('event G') == 1