Produits Entr'ouvert » Chrono

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

from __future__ import unicode_literals

from django.db import migrations, transaction

from django.db.utils import InternalError, OperationalError, ProgrammingError

sql_forwards = """

-- Add technical columns

ALTER TABLE agendas_event ADD COLUMN _end_datetime TIMESTAMP WITH TIME ZONE;

ALTER TABLE agendas_event ADD COLUMN _ignore_reason CHARACTER VARYING(20);

-- trigger and procedure to maintain _ignore_reason from bookings

CREATE OR REPLACE FUNCTION set_ignore_reason() RETURNS TRIGGER AS $$

    BEGIN

        IF (TG_OP = 'INSERT') THEN

            IF NEW.cancellation_datetime IS NOT NULL THEN

                UPDATE agendas_event SET _ignore_reason = 'cancel' WHERE id = NEW.event_id;

            END IF;

            RETURN NEW;

        ELSE

            PERFORM 1 FROM agendas_booking b WHERE b.event_id = OLD.event_id AND b.cancellation_datetime IS NOT NULL;

            IF FOUND THEN

                UPDATE agendas_event SET _ignore_reason = 'cancel' WHERE id = OLD.event_id;

            ELSE

                UPDATE agendas_event SET _ignore_reason = NULL WHERE id = OLD.event_id;

            END IF;

            IF (TG_OP = 'UPDATE') THEN

                RETURN NEW;

            ELSE

                RETURN OLD;

            END IF;

        END IF;

    END;

$$ LANGUAGE plpgsql;

DROP TRIGGER IF EXISTS set_ignore_reason_trg ON agendas_booking;

CREATE TRIGGER set_ignore_reason_trg

    AFTER INSERT OR UPDATE OR DELETE ON agendas_booking

    FOR EACH ROW

    EXECUTE PROCEDURE set_ignore_reason();

-- triggers to maintain _end_datetime

CREATE OR REPLACE FUNCTION update_duration() RETURNS TRIGGER AS $$

    BEGIN

        UPDATE agendas_event SET _end_datetime = start_datetime + (NEW.duration ||' minutes')::interval WHERE meeting_type_id = NEW.id;

        RETURN NEW;

    END;

$$ LANGUAGE plpgsql;

DROP TRIGGER IF EXISTS update_duration_trg ON agendas_meetingtype;

CREATE TRIGGER update_duration_trg

    AFTER UPDATE ON agendas_meetingtype

    FOR EACH ROW

    WHEN (OLD.duration != NEW.duration)

    EXECUTE PROCEDURE update_duration();

CREATE OR REPLACE FUNCTION set_end_datetime() RETURNS TRIGGER AS $$

    BEGIN

        IF NEW.meeting_type_id IS NULL THEN

            NEW._end_datetime = NULL;

        ELSE

            NEW._end_datetime = NEW.start_datetime + ((

                SELECT mt.duration FROM agendas_meetingtype mt WHERE mt.id = NEW.meeting_type_id) ||' minutes')::interval;

        END IF;

        RETURN NEW;

    END;

$$ LANGUAGE plpgsql;

DROP TRIGGER IF EXISTS set_end_datetime_trg ON agendas_event;

CREATE TRIGGER set_end_datetime_trg

    BEFORE INSERT OR UPDATE ON agendas_event

    FOR EACH ROW

    EXECUTE PROCEDURE set_end_datetime();

-- Init legacy

UPDATE agendas_event SET _ignore_reason = 'history';

UPDATE

    agendas_event

SET

    _end_datetime = agendas_event.start_datetime + (agendas_meetingtype.duration ||' minutes')::interval

FROM

    agendas_meetingtype

WHERE

    agendas_meetingtype.id = agendas_event.meeting_type_id;

"""

sql_backwards = """

DROP TRIGGER IF EXISTS set_ignore_reason_trg ON agendas_booking;

DROP FUNCTION IF EXISTS set_ignore_reason;

DROP TRIGGER IF EXISTS update_duration_trg ON agendas_meetingtype;

DROP FUNCTION IF EXISTS update_duration;

DROP TRIGGER IF EXISTS set_end_datetime_trg ON agendas_event;

DROP FUNCTION IF EXISTS set_end_datetime;

ALTER TABLE agendas_event DROP COLUMN _end_datetime;

ALTER TABLE agendas_event DROP COLUMN _ignore_reason;

"""

class Migration(migrations.Migration):

    dependencies = [

        ('agendas', '0050_event_slug'),

    ]

    operations = [migrations.RunSQL(sql=sql_forwards, reverse_sql=sql_backwards)]

    def _check_db(self, project_state, schema_editor):

        if schema_editor.connection.vendor != 'postgresql':

            return project_state

        try:

            with transaction.atomic():

                try:

                    # will fail if extension does not exist and can not be created

                    schema_editor.execute('CREATE EXTENSION IF NOT EXISTS btree_Gist SCHEMA public')

                except (OperationalError, ProgrammingError):

                    # if no extension, do not create columns and triggers

                    return project_state

        except InternalError:

            return project_state

    def apply(self, project_state, schema_editor, *args, **kwargs):

        result = self._check_db(project_state, schema_editor)

        return result or super().apply(project_state, schema_editor, *args, **kwargs)

    def unapply(self, project_state, schema_editor, *args, **kwargs):

        result = self._check_db(project_state, schema_editor)

        return result or super().unapply(project_state, schema_editor, *args, **kwargs)

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

from __future__ import unicode_literals

from django.db import migrations, transaction

from django.db.utils import InternalError, OperationalError, ProgrammingError

sql_forwards = """

ALTER TABLE agendas_event

ADD CONSTRAINT tstzrange_constraint

EXCLUDE USING GIST(desk_id WITH =, tstzrange(start_datetime, _end_datetime) WITH &&)

    WHERE (_ignore_reason IS NULL AND _end_datetime IS NOT NULL and desk_id IS NOT NULL);

"""

sql_backwards = """

ALTER TABLE agendas_event DROP CONSTRAINT tstzrange_constraint;

"""

class Migration(migrations.Migration):

    dependencies = [

        ('agendas', '0051_event_date_range_constraint'),

    ]

    operations = [migrations.RunSQL(sql=sql_forwards, reverse_sql=sql_backwards)]

    def _check_db(self, project_state, schema_editor):

        if schema_editor.connection.vendor != 'postgresql':

            return project_state

        try:

            with transaction.atomic():

                try:

                    # will fail if extension does not exist and can not be created

                    schema_editor.execute('CREATE EXTENSION IF NOT EXISTS btree_Gist SCHEMA public')

                except (OperationalError, ProgrammingError):

                    # if no extension, columns and triggers does not exist and constraint can not be added

                    return project_state

        except InternalError:

            return project_state

    def apply(self, project_state, schema_editor, *args, **kwargs):

        result = self._check_db(project_state, schema_editor)

        return result or super().apply(project_state, schema_editor, *args, **kwargs)

    def unapply(self, project_state, schema_editor, *args, **kwargs):

        result = self._check_db(project_state, schema_editor)

        return result or super().unapply(project_state, schema_editor, *args, **kwargs)

                day=middle_day.day - middle_day.weekday() + 2, hour=hour, minute=minute

                day=middle_day.day - middle_day.weekday() + 2, hour=hour, minute=minute

import datetime

from django.db import IntegrityError

from django.db import ProgrammingError

from django.db import connection

from django.db import transaction

from django.utils.timezone import now

import pytest

from chrono.agendas.models import Agenda

from chrono.agendas.models import Booking

from chrono.agendas.models import Desk

from chrono.agendas.models import Event

from chrono.agendas.models import MeetingType

pytestmark = pytest.mark.django_db

def check_ignore_reason(event, value):

    with connection.cursor() as cursor:

        cursor.execute("SELECT _ignore_reason FROM agendas_event WHERE id = %s", [event.pk])

        row = cursor.fetchone()

        assert row[0] == value

def set_ignore_reason(event, value):

    with connection.cursor() as cursor:

        cursor.execute("UPDATE agendas_event SET _ignore_reason = %s WHERE id = %s", [value, event.pk])

def check_end_datetime(event, value):

    with connection.cursor() as cursor:

        cursor.execute("SELECT _end_datetime FROM agendas_event WHERE id = %s", [event.pk])

        row = cursor.fetchone()

        assert row[0] == value

def test_event_ignore_reason():

    if connection.vendor != 'postgresql':

        pytest.skip('postgresql required')

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

    meeting_type = MeetingType.objects.create(agenda=agenda, label='Foo', duration=60)

    desk = Desk.objects.create(agenda=agenda, label='Desk')

    event = Event.objects.create(

        start_datetime=now(), meeting_type=meeting_type, places=10, agenda=agenda, desk=desk

    )

    try:

        check_ignore_reason(event, None)

    except ProgrammingError:

        pytest.skip('btree_Gist extension required')

    Booking.objects.create(event=event)

    check_ignore_reason(event, None)

    Booking.objects.create(event=event, cancellation_datetime=now())

    check_ignore_reason(event, 'cancel')

    Booking.objects.update(cancellation_datetime=None)

    check_ignore_reason(event, None)

    Booking.objects.update(cancellation_datetime=now())

    check_ignore_reason(event, 'cancel')

    Booking.objects.first().delete()

    check_ignore_reason(event, 'cancel')

    Booking.objects.all().delete()

    check_ignore_reason(event, None)

def test_event_end_datetime():

    if connection.vendor != 'postgresql':

        pytest.skip('postgresql required')

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

    meeting_type1 = MeetingType.objects.create(agenda=agenda, label='Foo', duration=60)

    meeting_type2 = MeetingType.objects.create(agenda=agenda, label='Foo', duration=45)

    event = Event.objects.create(start_datetime=now(), places=10, agenda=agenda)

    try:

        check_end_datetime(event, None)

    except ProgrammingError:

        pytest.skip('btree_Gist extension required')

    event.meeting_type = meeting_type1

    event.save()

    check_end_datetime(event, event.start_datetime + datetime.timedelta(minutes=60))

    event.meeting_type = meeting_type2

    event.save()

    check_end_datetime(event, event.start_datetime + datetime.timedelta(minutes=45))

    event.start_datetime = now()

    event.save()

    check_end_datetime(event, event.start_datetime + datetime.timedelta(minutes=45))

    event.meeting_type = None

    event.save()

    check_end_datetime(event, None)

    event.meeting_type = meeting_type2

    event.save()

    check_end_datetime(event, event.start_datetime + datetime.timedelta(minutes=45))

    event2 = Event.objects.create(start_datetime=now(), meeting_type=meeting_type1, places=10, agenda=agenda)

    check_end_datetime(event2, event2.start_datetime + datetime.timedelta(minutes=60))

    meeting_type1.duration = 42

    meeting_type1.save()

    check_end_datetime(event2, event2.start_datetime + datetime.timedelta(minutes=42))

    check_end_datetime(event, event.start_datetime + datetime.timedelta(minutes=45))

def test_meeting_event_exclusion_constraint():

    if connection.vendor != 'postgresql':

        pytest.skip('postgresql required')

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

    meeting_type1 = MeetingType.objects.create(agenda=agenda, label='Foo 1', duration=60)

    meeting_type2 = MeetingType.objects.create(agenda=agenda, label='Foo 2', duration=30)

    desk1 = Desk.objects.create(agenda=agenda, label='Desk 1')

    desk2 = Desk.objects.create(agenda=agenda, label='Desk 2')

    # create an event

    event1 = Event.objects.create(

        start_datetime=now(), meeting_type=meeting_type1, places=10, agenda=agenda, desk=desk1

    )

    try:

        check_ignore_reason(event1, None)

    except ProgrammingError:

        pytest.skip('btree_Gist extension required')

    # create a event with the same date range on other desk => no error

    Event.objects.create(

        start_datetime=event1.start_datetime, meeting_type=meeting_type1, places=10, agenda=agenda, desk=desk2

    )

    # no check if no meeting_type set

    Event.objects.create(start_datetime=event1.start_datetime, places=10, agenda=agenda, desk=desk1)

    # no check if no desk set

    Event.objects.create(

        start_datetime=event1.start_datetime, meeting_type=meeting_type1, places=10, agenda=agenda

    )

    # create an event just after the first one => ok

    Event.objects.create(

        start_datetime=event1.start_datetime + datetime.timedelta(minutes=60),

        meeting_type=meeting_type1,

        places=10,

        agenda=agenda,

        desk=desk1,

    )

    # try to create an event with the same date range and the same desk => error

    with pytest.raises(IntegrityError):

        with transaction.atomic():

            Event.objects.create(

                start_datetime=event1.start_datetime,

                meeting_type=meeting_type1,

                places=10,

                agenda=agenda,

                desk=desk1,

            )

    # try to create an event with the same start date, other duration and the same desk => error

    with pytest.raises(IntegrityError):

        with transaction.atomic():

            Event.objects.create(

                start_datetime=event1.start_datetime,

                meeting_type=meeting_type2,

                places=10,

                agenda=agenda,

                desk=desk1,

            )

    # try to create an event overlaping the first one on the same desk => error

    with pytest.raises(IntegrityError):

        with transaction.atomic():

            Event.objects.create(

                start_datetime=event1.start_datetime + datetime.timedelta(minutes=10),

                meeting_type=meeting_type1,

                places=10,

                agenda=agenda,

                desk=desk1,

            )

    with pytest.raises(IntegrityError):

        with transaction.atomic():

            Event.objects.create(

                start_datetime=event1.start_datetime + datetime.timedelta(minutes=10),

                meeting_type=meeting_type2,

                places=10,

                agenda=agenda,

                desk=desk1,

            )

    with pytest.raises(IntegrityError):

        with transaction.atomic():

            Event.objects.create(

                start_datetime=event1.start_datetime - datetime.timedelta(minutes=10),

                meeting_type=meeting_type1,

                places=10,

                agenda=agenda,

                desk=desk1,

            )

    with pytest.raises(IntegrityError):

        with transaction.atomic():

            Event.objects.create(

                start_datetime=event1.start_datetime - datetime.timedelta(minutes=10),

                meeting_type=meeting_type2,

                places=10,

                agenda=agenda,

                desk=desk1,

            )

    # but if event1 is cancelled it's ok

    Booking.objects.create(event=event1, cancellation_datetime=now())

    Event.objects.create(

        start_datetime=event1.start_datetime, meeting_type=meeting_type1, places=10, agenda=agenda, desk=desk1

    )