chrono/tests/test_import_export.py

from cStringIO import StringIO
import datetime
import json
import os
import shutil
import sys
import tempfile

import pytest
from django.core.management import call_command
from django.utils.timezone import make_aware

from chrono.agendas.models import (Agenda, Event, MeetingType, TimePeriod,
                                   Desk, TimePeriodException)
from chrono.manager.utils import export_site, 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):
    # add exception to meeting agenda
    desk = meetings_agenda.desk_set.first()
    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)))
    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')
    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(output)
        f.flush()
        call_command('import_site', f.name)

    assert Agenda.objects.count() == 3

    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(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=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)