Produits Entr'ouvert » Chrono

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

# Generated by Django 1.11.18 on 2020-10-21 11:56

from __future__ import unicode_literals

from django.db import migrations, models

from django.db.models import Count, F

def remove_broken_exceptions(apps, schema_editor):

    TimePeriodException = apps.get_model('agendas', 'TimePeriodException')

    # an exception is broken if its desk in not the same at the desk of its source

    qs = TimePeriodException.objects.filter(source__isnull=False)

    qs.exclude(source__desk=F('desk')).delete()

def remove_duplicate_sources(apps, schema_editor):

    Desk = apps.get_model('agendas', 'Desk')

    TimePeriodExceptionSource = apps.get_model('agendas', 'TimePeriodExceptionSource')

    for desk in Desk.objects.all():

        duplicate_source_slugs = (

            desk.timeperiodexceptionsource_set.values('settings_slug')

            .annotate(count=Count('settings_slug'))

            .order_by()

            .filter(count__gt=1)

        )

        if not duplicate_source_slugs:

            continue

        for source in duplicate_source_slugs:

            settings_slug = source['settings_slug']

            duplicate_sources = desk.timeperiodexceptionsource_set.filter(settings_slug=settings_slug)

            # remove duplicates, keeping the one that has related time period exceptions, if any

            source_to_keep = duplicate_sources.filter(timeperiodexception__isnull=False).first()

            if not source_to_keep:

                source_to_keep = duplicate_sources.first()

            duplicate_sources.exclude(pk=source_to_keep.pk).delete()

class Migration(migrations.Migration):

    dependencies = [

        ('agendas', '0065_unavailability_calendar'),

    ]

    operations = [

        migrations.RunPython(remove_broken_exceptions, migrations.RunPython.noop),

        migrations.RunPython(remove_duplicate_sources, migrations.RunPython.noop),

    ]

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

# Generated by Django 1.11.18 on 2020-10-21 15:46

from __future__ import unicode_literals

from django.db import migrations

class Migration(migrations.Migration):

    dependencies = [

        ('agendas', '0066_timeperiodexceptionsource_unique_settings_slug'),

    ]

    operations = [

        migrations.AlterUniqueTogether(

            name='timeperiodexceptionsource', unique_together=set([('desk', 'settings_slug')]),

        ),

    ]

    class Meta:

        unique_together = ['desk', 'settings_slug']

from django.db.migrations.executor import MigrationExecutor

def test_clean_time_period_exceptions(transactional_db):

    app = 'agendas'

    migrate_from = [(app, '0065_unavailability_calendar')]

    migrate_to = [(app, '0066_timeperiodexceptionsource_unique_settings_slug')]

    executor = MigrationExecutor(connection)

    old_apps = executor.loader.project_state(migrate_from).apps

    executor.migrate(migrate_from)

    Agenda = old_apps.get_model(app, 'Agenda')

    Desk = old_apps.get_model(app, 'Desk')

    TimePeriodException = old_apps.get_model(app, 'TimePeriodException')

    TimePeriodExceptionSource = old_apps.get_model(app, 'TimePeriodExceptionSource')

    agenda = Agenda.objects.create(label='Agenda')

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

    # add normal time period exception to Desk

    source_desk = TimePeriodExceptionSource.objects.create(desk=desk, settings_slug='holidays', enabled=True)

    start_datetime = datetime.datetime(year=2020, month=1, day=2)

    end_datetime = datetime.datetime(year=2020, month=1, day=3)

    for i in range(5):

        TimePeriodException.objects.create(

            desk=desk,

            source=source_desk,

            external=True,

            start_datetime=start_datetime,

            end_datetime=end_datetime,

        )

    # now simulate broken state (desk duplication)

    new_desk = Desk.objects.create(label='New Desk', slug='new-desk', agenda=agenda)

    # normal source and exceptions

    source_new_desk = TimePeriodExceptionSource.objects.create(

        desk=new_desk, settings_slug='holidays', enabled=True

    )

    for i in range(5):

        TimePeriodException.objects.create(

            desk=new_desk,

            source=source_new_desk,

            external=True,

            start_datetime=start_datetime,

            end_datetime=end_datetime,

        )

    # wrong duplicate of source

    TimePeriodExceptionSource.objects.create(desk=new_desk, settings_slug='holidays', enabled=True)

    # wrong duplicate of exceptions, referencing original desk source

    for i in range(5):

        TimePeriodException.objects.create(

            desk=new_desk,

            source=source_desk,

            external=True,

            start_datetime=start_datetime,

            end_datetime=end_datetime,

        )

    # ensure migration fixes state

    executor = MigrationExecutor(connection)

    executor.migrate(migrate_to)

    executor.loader.build_graph()

    apps = executor.loader.project_state(migrate_to).apps

    Desk = apps.get_model(app, 'Desk')

    TimePeriodException = apps.get_model(app, 'TimePeriodException')

    TimePeriodExceptionSource = apps.get_model(app, 'TimePeriodExceptionSource')

    # original desk hasn't been touched

    desk = Desk.objects.get(pk=desk.pk)

    assert desk.timeperiodexception_set.count() == 5

    assert desk.timeperiodexceptionsource_set.filter(settings_slug='holidays').count() == 1

    # duplicated desk has correct exceptions

    new_desk = Desk.objects.get(pk=new_desk.pk)

    assert new_desk.timeperiodexception_set.count() == 5

    assert new_desk.timeperiodexceptionsource_set.filter(settings_slug='holidays').count() == 1

    exc = new_desk.timeperiodexception_set.first()

    assert exc.source == new_desk.timeperiodexceptionsource_set.get(settings_slug='holidays')