agendas: import/export all exception sources (#80219)

chrono/agendas/models.py

@@ -14,6 +14,7 @@
 # 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 base64
 import collections
 import copy
 import dataclasses
@@ -35,6 +36,7 @@ from django.contrib.auth.models import Group
 from django.contrib.humanize.templatetags.humanize import ordinal
 from django.contrib.postgres.fields import ArrayField
 from django.core.exceptions import ValidationError
+from django.core.files.base import ContentFile
 from django.core.validators import MaxValueValidator, MinValueValidator
 from django.db import connection, models, transaction
 from django.db.models import (
@@ -2934,14 +2936,15 @@ class Desk(models.Model):
             desk.unavailability_calendars.add(target_calendar)
 
     def export_json(self):
-        time_period_exceptions = self.timeperiodexception_set.filter(source__settings_slug__isnull=True)
-        time_period_exception_sources = self.timeperiodexceptionsource_set.filter(settings_slug__isnull=False)
+        time_period_exceptions = self.timeperiodexception_set.filter(source__isnull=True)
         return {
             'label': self.label,
             'slug': self.slug,
             'timeperiods': [time_period.export_json() for time_period in self.timeperiod_set.filter()],
             'exceptions': [exception.export_json() for exception in time_period_exceptions],
-            'exception_sources': [source.export_json() for source in time_period_exception_sources],
+            'exception_sources': [
+                source.export_json() for source in self.timeperiodexceptionsource_set.all()
+            ],
             'unavailability_calendars': [{'slug': x.slug} for x in self.unavailability_calendars.all()],
         }
 
@@ -3365,15 +3368,31 @@ class TimePeriodExceptionSource(models.Model):
     @classmethod
     def import_json(cls, data):
         data = clean_import_data(cls, data)
+
+        if data.get('ics_file'):
+            data['ics_file'] = ContentFile(base64.b64decode(data['ics_file']), name=data['ics_filename'])
+
         desk = data.pop('desk')
         settings_slug = data.pop('settings_slug')
-        source, _ = cls.objects.update_or_create(desk=desk, settings_slug=settings_slug, defaults=data)
-        if source.enabled:
-            source.enable()
+        ics_url = data.pop('ics_url', None)
+        ics_filename = data.pop('ics_filename', None)
+        source, _ = cls.objects.update_or_create(
+            desk=desk, settings_slug=settings_slug, ics_filename=ics_filename, ics_url=ics_url, defaults=data
+        )
+        if settings_slug:
+            if source.enabled:
+                source.enable()
+        else:
+            try:
+                source.refresh_timeperiod_exceptions_from_ics()
+            except ICSError:
+                pass
 
     def export_json(self):
-        '''Export only sources from settings.'''
         return {
+            'ics_filename': self.ics_filename,
+            'ics_file': base64.b64encode(self.ics_file.read()).decode() if self.ics_file else None,
+            'ics_url': self.ics_url,
             'settings_slug': self.settings_slug,
             'settings_label': self.settings_label,
             'enabled': self.enabled,

tests/test_import_export.py

@@ -10,6 +10,7 @@ from unittest import mock
 
 import pytest
 from django.contrib.auth.models import Group
+from django.core.files.base import ContentFile
 from django.core.management import CommandError, call_command
 from django.test import override_settings
 from django.utils.encoding import force_bytes
@@ -35,6 +36,8 @@ from chrono.agendas.models import (
 from chrono.manager.utils import import_site
 from chrono.utils.timezone import make_aware, now
 
+from .test_agendas import ICS_SAMPLE, ICS_SAMPLE_WITH_NO_EVENTS
+
 pytestmark = pytest.mark.django_db
 
 
@@ -799,6 +802,99 @@ def test_import_export_time_period_exception_source_enabled():
     assert source.settings_slug == 'holidays'
 
 
+@mock.patch('chrono.agendas.models.requests.get')
+def test_import_export_time_period_exception_source_remote_ics(mocked_get):
+    agenda = Agenda.objects.create(label='Foo bar', kind='meetings')
+    desk = Desk.objects.create(label='Desk', agenda=agenda)
+    source = TimePeriodExceptionSource.objects.create(desk=desk, ics_url='http://example.com/sample.ics')
+
+    mocked_response = mock.Mock()
+    mocked_response.text = ICS_SAMPLE
+    mocked_get.return_value = mocked_response
+
+    source.refresh_timeperiod_exceptions_from_ics()
+    assert TimePeriodException.objects.count() == 2
+
+    output = get_output_of_command('export_site')
+    payload = json.loads(output)
+
+    Agenda.objects.all().delete()
+    assert not TimePeriodExceptionSource.objects.exists()
+    assert not TimePeriodException.objects.exists()
+
+    import_site(payload)
+    source = TimePeriodExceptionSource.objects.get()
+    assert source.timeperiodexception_set.count() == 2
+    assert TimePeriodException.objects.count() == 2
+
+    # import again changes nothing
+    import_site(payload)
+    assert TimePeriodExceptionSource.objects.count() == 1
+    assert TimePeriodException.objects.count() == 2
+
+    # empty remote ics
+    mocked_response.text = ICS_SAMPLE_WITH_NO_EVENTS
+
+    Agenda.objects.all().delete()
+    import_site(payload)
+
+    assert TimePeriodExceptionSource.objects.count() == 1
+    assert TimePeriodException.objects.count() == 0
+
+
+@mock.patch('chrono.agendas.models.requests.get')
+def test_import_export_time_period_exception_source_ics_file(mocked_get):
+    agenda = Agenda.objects.create(label='Foo bar', kind='meetings')
+    desk = Desk.objects.create(label='Desk', agenda=agenda)
+    source = TimePeriodExceptionSource.objects.create(
+        desk=desk, ics_filename='sample.ics', ics_file=ContentFile(ICS_SAMPLE, name='sample.ics')
+    )
+
+    source.refresh_timeperiod_exceptions_from_ics()
+    assert TimePeriodException.objects.count() == 2
+
+    output = get_output_of_command('export_site')
+    payload = json.loads(output)
+
+    Agenda.objects.all().delete()
+    assert not TimePeriodExceptionSource.objects.exists()
+    assert not TimePeriodException.objects.exists()
+
+    import_site(payload)
+    source = TimePeriodExceptionSource.objects.get()
+    assert source.timeperiodexception_set.count() == 2
+    assert TimePeriodException.objects.count() == 2
+
+    # import again changes nothing
+    import_site(payload)
+    assert TimePeriodExceptionSource.objects.count() == 1
+    assert TimePeriodException.objects.count() == 2
+
+
+@override_settings(
+    EXCEPTIONS_SOURCES={
+        'holidays': {'class': 'workalendar.europe.France', 'label': 'Holidays'},
+    }
+)
+def test_import_export_time_period_exception_legacy_file():
+    agenda = Agenda.objects.create(label='Foo bar', kind='events')
+    desk = Desk.objects.create(agenda=agenda, slug='_exceptions_holder')
+    desk.import_timeperiod_exceptions_from_settings()
+
+    output = get_output_of_command('export_site')
+    payload = json.loads(output)
+    source = payload['agendas'][0]['exceptions_desk']['exception_sources'][0]
+    del source['ics_file']
+    del source['ics_filename']
+    del source['ics_url']
+
+    agenda.delete()
+    assert not TimePeriodExceptionSource.objects.exists()
+
+    import_site(payload)
+    assert TimePeriodExceptionSource.objects.count() == 1
+
+
 def test_import_export_do_not_duplicate_timeperiod_and_exceptions():
     agenda = Agenda.objects.create(label='Foo bar', kind='meetings')
     desk = Desk.objects.create(slug='test', agenda=agenda)