chrono/tests/test_import_export.py

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

from __future__ import unicode_literals

import copy
import datetime
from io import StringIO
import json
import os
import shutil
import sys
import tempfile

import pytest
from django.contrib.auth.models import Group
from django.core.management import call_command, CommandError
from django.test import override_settings
from django.utils.encoding import force_bytes
from django.utils.timezone import make_aware, now

from chrono.agendas.models import (
    Agenda,
    Category,
    Desk,
    Event,
    Resource,
    TimePeriod,
    TimePeriodException,
    TimePeriodExceptionSource,
    AgendaImportError,
    MeetingType,
    VirtualMember,
    AgendaNotificationsSettings,
    AgendaReminderSettings,
)
from chrono.manager.utils import import_site

pytestmark = pytest.mark.django_db


def get_output_of_command(command, *args, **kwargs):
    old_stdout = sys.stdout
    output = sys.stdout = StringIO()
    call_command(command, *args, **kwargs)
    sys.stdout = old_stdout
    return output.getvalue()


def test_import_export(app):
    agenda_events = Agenda.objects.create(label='Events Agenda', kind='events')
    first_event = Event.objects.create(
        agenda=agenda_events, start_datetime=make_aware(datetime.datetime(2020, 7, 21, 16, 42, 35)), places=10
    )
    agenda_meetings = Agenda.objects.create(label='Meetings Agenda', kind='meetings')
    MeetingType.objects.create(agenda=agenda_meetings, label='Meeting Type', duration=30)
    desk = Desk.objects.create(agenda=agenda_meetings, label='Desk')

    # add exception to meeting agenda
    tpx_start = make_aware(datetime.datetime(2017, 5, 22, 8, 0))
    tpx_end = make_aware(datetime.datetime(2017, 5, 22, 12, 30))
    TimePeriodException.objects.create(desk=desk, start_datetime=tpx_start, end_datetime=tpx_end)
    output = get_output_of_command('export_site')
    assert len(json.loads(output)['agendas']) == 2
    import_site(data={}, clean=True)
    empty_output = get_output_of_command('export_site')
    assert len(json.loads(empty_output)['agendas']) == 0

    Agenda(label=u'test').save()
    old_stdin = sys.stdin
    sys.stdin = StringIO(json.dumps({}))
    assert Agenda.objects.count() == 1
    try:
        call_command('import_site', '-', clean=True)
    finally:
        sys.stdin = old_stdin
    assert Agenda.objects.count() == 0

    with tempfile.NamedTemporaryFile() as f:
        f.write(force_bytes(output))
        f.flush()
        call_command('import_site', f.name)

    assert Agenda.objects.count() == 2
    first_imported_event = Agenda.objects.get(label='Events Agenda').event_set.first()
    assert first_imported_event.start_datetime == first_event.start_datetime
    assert TimePeriodException.objects.get().start_datetime == tpx_start
    assert TimePeriodException.objects.get().end_datetime == tpx_end

    agenda1 = Agenda.objects.get(label='Events Agenda')
    agenda2 = Agenda.objects.get(label='Meetings Agenda')
    event = Event(agenda=agenda1, start_datetime=now(), places=10)
    event.save()
    desk, _ = Desk.objects.get_or_create(agenda=agenda2, label='Desk A', slug='desk-a')
    timeperiod = TimePeriod(
        desk=desk, weekday=2, start_time=datetime.time(10, 0), end_time=datetime.time(11, 0)
    )
    timeperiod.save()
    exception = TimePeriodException(
        desk=desk,
        start_datetime=make_aware(datetime.datetime(2017, 5, 22, 8, 0)),
        end_datetime=make_aware(datetime.datetime(2017, 5, 22, 12, 30)),
    )
    exception.save()

    import_site(json.loads(output), overwrite=True)
    assert Event.objects.filter(id=event.id).count() == 0
    assert Desk.objects.filter(slug='desk-a').count() == 0
    assert TimePeriod.objects.filter(id=timeperiod.id).count() == 0
    assert TimePeriodException.objects.filter(id=exception.id).count() == 0

    event = Event(agenda=agenda1, start_datetime=now(), places=10)
    event.save()
    desk, _ = Desk.objects.get_or_create(agenda=agenda2, label='Desk A', slug='desk-a')
    timeperiod = TimePeriod(
        weekday=2, desk=desk, start_time=datetime.time(10, 0), end_time=datetime.time(11, 0)
    )
    timeperiod.save()
    exception = TimePeriodException(
        desk=desk,
        start_datetime=make_aware(datetime.datetime(2017, 5, 22, 8, 0)),
        end_datetime=make_aware(datetime.datetime(2017, 5, 22, 12, 30)),
    )
    exception.save()
    import_site(json.loads(output), overwrite=False)
    assert Event.objects.filter(id=event.id).count() == 1
    assert TimePeriod.objects.filter(id=timeperiod.id).count() == 1
    assert TimePeriodException.objects.filter(id=exception.id).count() == 1

    import_site(data={}, if_empty=True)
    assert Agenda.objects.count() == 2

    import_site(data={}, clean=True)
    tempdir = tempfile.mkdtemp('chrono-test')
    empty_output = get_output_of_command('export_site', output=os.path.join(tempdir, 't.json'))
    assert os.path.exists(os.path.join(tempdir, 't.json'))
    shutil.rmtree(tempdir)


def test_import_export_events_agenda_options(app):
    agenda = Agenda.objects.create(
        label='Foo Bar',
        kind='events',
        default_view='open_events',
        booking_form_url='{{ eservices_url }}backoffice/submission/inscription-aux-activites/',
    )

    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
    agenda = Agenda.objects.first()
    assert agenda.default_view == 'open_events'
    assert agenda.booking_form_url == '{{ eservices_url }}backoffice/submission/inscription-aux-activites/'


def test_import_export_event_details(app):
    agenda = Agenda.objects.create(label='Foo Bar', kind='events')
    Event.objects.create(
        agenda=agenda,
        description='description',
        pricing='100',
        url='https://example.net/',
        publication_date=datetime.date(2020, 5, 11),
        places=42,
        start_datetime=now(),
    )

    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
    first_imported_event = Agenda.objects.get(label='Foo Bar').event_set.first()
    assert first_imported_event.description == 'description'
    assert first_imported_event.pricing == '100'
    assert first_imported_event.url == 'https://example.net/'
    assert first_imported_event.publication_date == datetime.date(2020, 5, 11)


def test_import_export_permissions(app):
    meetings_agenda = Agenda.objects.create(label='Foo Bar', kind='meetings')
    group1 = Group.objects.create(name=u'gé1')
    group2 = Group.objects.create(name=u'gé2')
    meetings_agenda.view_role = group1
    meetings_agenda.edit_role = group2
    meetings_agenda.save()
    output = get_output_of_command('export_site')
    assert len(json.loads(output)['agendas']) == 1
    import_site(data={}, clean=True)
    assert Agenda.objects.count() == 0
    Group.objects.all().delete()

    with pytest.raises(AgendaImportError) as excinfo:
        import_site(json.loads(output), overwrite=True)
    assert 'gé1' in str(excinfo.value) and 'gé2' in str(excinfo.value)

    group1 = Group(name=u'gé1')
    group1.save()
    with pytest.raises(AgendaImportError) as excinfo:
        import_site(json.loads(output), overwrite=True)
    assert u'%s' % excinfo.value == u'Missing roles: "gé2"'

    with tempfile.NamedTemporaryFile() as f:
        f.write(force_bytes(output))
        f.flush()
        with pytest.raises(CommandError) as excinfo:
            call_command('import_site', f.name)
        assert u'%s' % excinfo.value == u'Missing roles: "gé2"'

    group2 = Group(name=u'gé2')
    group2.save()
    import_site(json.loads(output), overwrite=True)

    agenda = Agenda.objects.get(slug=meetings_agenda.slug)
    assert agenda.view_role == group1
    assert agenda.edit_role == group2


def test_import_export_resources(app):
    meetings_agenda = Agenda.objects.create(label='Foo Bar', kind='meetings')
    resource = Resource.objects.create(label='foo')
    meetings_agenda.resources.add(resource)
    output = get_output_of_command('export_site')

    import_site(data={}, clean=True)
    assert Agenda.objects.count() == 0
    resource.delete()

    with pytest.raises(AgendaImportError) as excinfo:
        import_site(json.loads(output), overwrite=True)
    assert str(excinfo.value) == 'Missing "foo" resource'

    resource = Resource.objects.create(label='foobar')
    with pytest.raises(AgendaImportError) as excinfo:
        import_site(json.loads(output), overwrite=True)
    assert str(excinfo.value) == 'Missing "foo" resource'

    resource = Resource.objects.create(label='foo')
    import_site(json.loads(output), overwrite=True)
    agenda = Agenda.objects.get(slug=meetings_agenda.slug)
    assert list(agenda.resources.all()) == [resource]


def test_import_export_categorys(app):
    category = Category.objects.create(label='foo')
    agenda = Agenda.objects.create(label='Foo Bar', category=category)
    output = get_output_of_command('export_site')

    import_site(data={}, clean=True)
    assert Agenda.objects.count() == 0
    category.delete()

    with pytest.raises(AgendaImportError) as excinfo:
        import_site(json.loads(output), overwrite=True)
    assert str(excinfo.value) == 'Missing "foo" category'

    category = Category.objects.create(label='foobar')
    with pytest.raises(AgendaImportError) as excinfo:
        import_site(json.loads(output), overwrite=True)
    assert str(excinfo.value) == 'Missing "foo" category'

    category = Category.objects.create(label='foo')
    import_site(json.loads(output), overwrite=True)
    agenda = Agenda.objects.get(slug=agenda.slug)
    assert agenda.category == category


def test_import_export_virtual_agenda(app):
    virtual_agenda = Agenda.objects.create(label='Virtual Agenda', kind='virtual')
    output = get_output_of_command('export_site')
    payload = json.loads(output)
    assert len(payload['agendas']) == 1
    virtual_agenda.delete()
    import_site(json.loads(output), overwrite=True)

    virtual_agenda = Agenda.objects.get(label='Virtual Agenda', slug='virtual-agenda', kind='virtual')
    assert virtual_agenda.minimal_booking_delay is None
    assert virtual_agenda.maximal_booking_delay is None
    assert virtual_agenda.real_agendas.count() == 0
    assert virtual_agenda.excluded_timeperiods.count() == 0

    # add booking delay
    virtual_agenda.minimal_booking_delay = 2
    virtual_agenda.maximal_booking_delay = 10
    virtual_agenda.save()
    output = get_output_of_command('export_site')
    payload = json.loads(output)
    virtual_agenda.delete()
    import_site(json.loads(output), overwrite=True)
    virtual_agenda = Agenda.objects.get(label='Virtual Agenda', slug='virtual-agenda', kind='virtual')
    assert virtual_agenda.minimal_booking_delay == 2
    assert virtual_agenda.maximal_booking_delay == 10

    # add excluded timeperiods
    tp1 = TimePeriod.objects.create(
        agenda=virtual_agenda, weekday=1, start_time=datetime.time(10, 0), end_time=datetime.time(11, 0)
    )
    tp2 = TimePeriod.objects.create(
        agenda=virtual_agenda, weekday=2, start_time=datetime.time(12, 0), end_time=datetime.time(13, 0)
    )
    output = get_output_of_command('export_site')
    payload = json.loads(output)
    virtual_agenda.delete()
    tp1.delete()
    tp2.delete()
    import_site(json.loads(output), overwrite=True)
    virtual_agenda = Agenda.objects.get(label='Virtual Agenda', slug='virtual-agenda', kind='virtual')
    assert virtual_agenda.excluded_timeperiods.count() == 2
    tp1 = TimePeriod.objects.get(
        agenda=virtual_agenda, weekday=1, start_time=datetime.time(10, 0), end_time=datetime.time(11, 0)
    )
    tp2 = TimePeriod.objects.get(
        agenda=virtual_agenda, weekday=2, start_time=datetime.time(12, 0), end_time=datetime.time(13, 0)
    )


def test_import_export_virtual_agenda_with_included_agenda(app):
    virtual_agenda = Agenda.objects.create(label='Virtual Agenda', kind='virtual')
    foo_agenda = Agenda.objects.create(label='Foo', kind='meetings')
    bar_agenda = Agenda.objects.create(label='Bar', kind='meetings')
    mt1 = MeetingType.objects.create(agenda=foo_agenda, label='Meeting Type', duration=30)
    mt2 = MeetingType.objects.create(agenda=bar_agenda, label='Meeting Type', duration=30)
    VirtualMember.objects.create(virtual_agenda=virtual_agenda, real_agenda=foo_agenda)
    VirtualMember.objects.create(virtual_agenda=virtual_agenda, real_agenda=bar_agenda)
    output = get_output_of_command('export_site')
    payload = json.loads(output)
    assert len(payload['agendas']) == 3
    virtual_agenda.delete()
    foo_agenda.delete()
    bar_agenda.delete()
    mt1.delete()
    mt2.delete()
    import_site(payload, overwrite=True)

    virtual_agenda = Agenda.objects.get(label='Virtual Agenda', slug='virtual-agenda', kind='virtual')
    assert virtual_agenda.real_agendas.count() == 2
    assert virtual_agenda.real_agendas.filter(label='Foo').count() == 1
    assert virtual_agenda.real_agendas.filter(label='Bar').count() == 1

    # add incompatible meetingtype
    bar_agenda = Agenda.objects.get(label='Bar', kind='meetings')
    mt2 = MeetingType.objects.get(agenda=bar_agenda, label='Meeting Type')
    mt2.duration = 10
    mt2.save()
    output = get_output_of_command('export_site')
    payload = json.loads(output)
    virtual_agenda.delete()
    with pytest.raises(AgendaImportError) as excinfo:
        import_site(json.loads(output), overwrite=False)
    assert (
        'This agenda does not have the same meeting types provided by the virtual agenda.'
        in '%s' % excinfo.value
    )


def test_import_export_virtual_agenda_with_missing_agenda(app):
    virtual_agenda = Agenda.objects.create(label='Virtual Agenda', kind='virtual')
    foo_agenda = Agenda.objects.create(label='Foo', kind='meetings')
    VirtualMember.objects.create(virtual_agenda=virtual_agenda, real_agenda=foo_agenda)
    output = get_output_of_command('export_site')
    payload = json.loads(output)
    del payload['agendas'][0]  # remove real agenda

    virtual_agenda.delete()
    foo_agenda.delete()
    with pytest.raises(AgendaImportError) as excinfo:
        import_site(payload, overwrite=False)
    assert 'The real agenda "foo" does not exist.' in '%s' % excinfo.value


def test_import_export_desk_missing_fields(app):
    meetings_agenda = Agenda.objects.create(label='Foo Bar', kind='meetings')
    MeetingType.objects.create(agenda=meetings_agenda, label='Meeting Type', duration=30)
    desk = Desk.objects.create(agenda=meetings_agenda, label='Desk')
    TimePeriod.objects.create(
        desk=desk, weekday=2, start_time=datetime.time(10, 0), end_time=datetime.time(11, 0)
    )
    TimePeriodException.objects.create(
        desk=desk,
        start_datetime=make_aware(datetime.datetime(2017, 5, 22, 8, 0)),
        end_datetime=make_aware(datetime.datetime(2017, 5, 22, 12, 30)),
    )
    output = get_output_of_command('export_site')
    payload = json.loads(output)
    del payload['agendas'][0]['desks'][0]['timeperiods']
    del payload['agendas'][0]['desks'][0]['exceptions']
    Agenda.objects.all().delete()
    import_site(payload)
    assert TimePeriod.objects.exists() is False
    assert TimePeriodException.objects.exists() is False


def test_import_export_desk_unknown_fields(app):
    events_agenda = Agenda.objects.create(label='Events Agenda', kind='events')
    Event.objects.create(agenda=events_agenda, start_datetime=now(), places=10)
    meetings_agenda = Agenda.objects.create(label='Meetings Agenda', kind='meetings')
    MeetingType.objects.create(agenda=meetings_agenda, label='Meeting Type', duration=30)
    Desk.objects.create(agenda=meetings_agenda, label='Desk')
    desk = meetings_agenda.desk_set.first()
    TimePeriod.objects.create(
        desk=desk, weekday=2, start_time=datetime.time(10, 0), end_time=datetime.time(11, 0)
    )
    TimePeriodException.objects.create(
        desk=desk,
        start_datetime=make_aware(datetime.datetime(2017, 5, 22, 8, 0)),
        end_datetime=make_aware(datetime.datetime(2017, 5, 22, 12, 30)),
    )

    output = get_output_of_command('export_site')
    payload = json.loads(output)

    Agenda.objects.all().delete()
    assert Agenda.objects.exists() is False
    assert MeetingType.objects.exists() is False
    assert Desk.objects.exists() is False
    assert TimePeriod.objects.exists() is False
    assert TimePeriodException.objects.exists() is False

    # add unknown fields everywhere
    payload['agendas'][0]['unknown_field'] = 'foobar'
    payload['agendas'][0]['events'][0]['unknown_field'] = 'foobar'
    payload['agendas'][1]['unknown_field'] = 'foobar'
    payload['agendas'][1]['meetingtypes'][0]['unknown_field'] = 'foobar'
    payload['agendas'][1]['desks'][0]['unknown_field'] = 'foobar'
    payload['agendas'][1]['desks'][0]['timeperiods'][0]['unknown_field'] = 'foobar'
    payload['agendas'][1]['desks'][0]['exceptions'][0]['unknown_field'] = 'foobar'

    import_site(payload)
    assert Agenda.objects.exists() is True
    assert MeetingType.objects.exists() is True
    assert Desk.objects.exists() is True
    assert TimePeriod.objects.exists() is True
    assert TimePeriodException.objects.exists() is True


def test_import_export_slug_fields(app):
    agenda_events = Agenda.objects.create(label='Events Agenda', kind='events')
    Event.objects.create(agenda=agenda_events, start_datetime=now(), places=10)
    agenda_meetings = Agenda.objects.create(label='Meetings Agenda', kind='meetings')
    MeetingType.objects.create(agenda=agenda_meetings, label='Meeting Type', duration=30)
    Desk.objects.create(agenda=agenda_meetings, label='Desk')

    output = get_output_of_command('export_site')
    original_payload = json.loads(output)

    Agenda.objects.all().delete()
    assert Agenda.objects.exists() is False
    assert MeetingType.objects.exists() is False
    assert Desk.objects.exists() is False
    assert Event.objects.exists() is False

    payload = copy.deepcopy(original_payload)
    payload['agendas'][0]['slug'] = 'events-agenda&'
    with pytest.raises(AgendaImportError) as excinfo:
        import_site(payload)
    assert str(excinfo.value) == 'Bad slug format "events-agenda&"'

    payload = copy.deepcopy(original_payload)
    payload['agendas'][0]['events'][0]['slug'] = 'events-agenda-event&'
    with pytest.raises(AgendaImportError) as excinfo:
        import_site(payload)
    assert str(excinfo.value) == 'Bad slug format "events-agenda-event&"'

    payload = copy.deepcopy(original_payload)
    payload['agendas'][1]['desks'][0]['slug'] = 'desk&'
    with pytest.raises(AgendaImportError) as excinfo:
        import_site(payload)
    assert str(excinfo.value) == 'Bad slug format "desk&"'

    payload = copy.deepcopy(original_payload)
    payload['agendas'][1]['meetingtypes'][0]['slug'] = 'meeting-type&'
    with pytest.raises(AgendaImportError) as excinfo:
        import_site(payload)
    assert str(excinfo.value) == 'Bad slug format "meeting-type&"'


def test_import_export_notification_settings():
    agenda = Agenda.objects.create(label='Foo bar', kind='events')
    AgendaNotificationsSettings.objects.create(
        agenda=agenda,
        almost_full_event=AgendaNotificationsSettings.EDIT_ROLE,
        full_event=AgendaNotificationsSettings.VIEW_ROLE,
        cancelled_event=AgendaNotificationsSettings.EMAIL_FIELD,
        cancelled_event_emails=['hop@entrouvert.com', 'top@entrouvert.com'],
    )
    output = get_output_of_command('export_site')
    payload = json.loads(output)

    agenda.delete()
    assert not AgendaNotificationsSettings.objects.exists()

    import_site(payload)
    agenda = Agenda.objects.first()
    AgendaNotificationsSettings.objects.get(
        agenda=agenda,
        almost_full_event=AgendaNotificationsSettings.EDIT_ROLE,
        full_event=AgendaNotificationsSettings.VIEW_ROLE,
        cancelled_event=AgendaNotificationsSettings.EMAIL_FIELD,
        cancelled_event_emails=['hop@entrouvert.com', 'top@entrouvert.com'],
    )

    # again - check OneToOneField
    import_site(payload)


def test_import_export_reminder_settings():
    agenda = Agenda.objects.create(label='Foo bar', kind='events')
    AgendaReminderSettings.objects.create(
        agenda=agenda, days=2, send_email=True, send_sms=False, email_extra_info='test',
    )
    output = get_output_of_command('export_site')
    payload = json.loads(output)

    agenda.delete()
    assert not AgendaReminderSettings.objects.exists()

    import_site(payload)
    agenda = Agenda.objects.first()
    AgendaReminderSettings.objects.get(
        agenda=agenda, days=2, send_email=True, send_sms=False, email_extra_info='test',
    )

    # again - check OneToOneField
    import_site(payload)


@override_settings(
    EXCEPTIONS_SOURCES={'holidays': {'class': 'workalendar.europe.France', 'label': 'Holidays'},}
)
def test_import_export_time_period_exception_source():
    agenda = Agenda.objects.create(label='Foo bar', kind='meetings')
    desk = Desk.objects.create(slug='test', agenda=agenda)
    desk.import_timeperiod_exceptions_from_settings(enable=True)

    output = get_output_of_command('export_site')
    payload = json.loads(output)

    agenda.delete()
    assert not TimePeriodExceptionSource.objects.exists()

    import_site(payload)
    desk = Desk.objects.get(slug='test')
    source = desk.timeperiodexceptionsource_set.first()
    assert source.enabled
    assert desk.timeperiodexception_set.exists()

    source.disable()
    output = get_output_of_command('export_site')
    payload = json.loads(output)
    Agenda.objects.all().delete()

    import_site(payload)
    desk = Desk.objects.get(slug='test')
    source = desk.timeperiodexceptionsource_set.first()
    assert not source.enabled
    assert not desk.timeperiodexception_set.exists()