chrono/tests/test_import_export.py

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

from __future__ import unicode_literals

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.utils.encoding import force_bytes
from django.utils.timezone import make_aware

from chrono.agendas.models import (Agenda, Event, TimePeriod, Desk,
        TimePeriodException, AgendaImportError)
from chrono.manager.utils import import_site

from test_api import some_data, meetings_agenda, time_zone, mock_now

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, some_data, meetings_agenda):
    first_event = Agenda.objects.get(label='Foo bar').event_set.first()

    # add exception to meeting agenda
    desk = meetings_agenda.desk_set.first()
    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']) == 3
    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() == 3
    first_imported_event = Agenda.objects.get(label='Foo bar').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=u'Foo bar')
    agenda2 = Agenda.objects.get(label=u'Foo bar Meeting')
    event = Event(agenda=agenda1, start_datetime=make_aware(datetime.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=make_aware(datetime.datetime.now()), places=10)
    event.save()
    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() == 3

    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_event_description(app, some_data, meetings_agenda):
    first_event = Agenda.objects.get(label='Foo bar').event_set.first()
    first_event.description = 'description'
    first_event.save()

    output = get_output_of_command('export_site')
    assert len(json.loads(output)['agendas']) == 3
    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() == 3
    first_imported_event = Agenda.objects.get(label='Foo bar').event_set.first()
    assert first_imported_event.description == 'description'


def test_import_export_permissions(app, some_data, meetings_agenda):
    group1 = Group(name=u'gé1')
    group1.save()
    group2 = Group(name=u'gé2')
    group2.save()

    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']) == 3
    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 u'%s' % excinfo.value == u'Missing "gé1" role'

    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 "gé2" role'

    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 "gé2" role'

    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