Produits Entr'ouvert » Chrono

# chrono - agendas system

# Copyright (C) 2016-2017  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# 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 logging

from chrono.agendas.models import Desk, ICSError

from django.core.management.base import BaseCommand, CommandError

class Command(BaseCommand):

    help = 'Synchronize time period exceptions from desks remote ics'

    def handle(self, **options):

        logger = logging.getLogger(__name__)

        for desk in Desk.objects.exclude(timeperiod_exceptions_remote_url=''):

            try:

                desk.create_timeperiod_exceptions_from_remote_ics(desk.timeperiod_exceptions_remote_url)

            except ICSError as e:

                raise CommandError(u'unable to create timeperiod exceptions for "%s": %s' % (desk, e))

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

from __future__ import unicode_literals

from django.db import migrations, models

import datetime

from django.utils.timezone import utc

class Migration(migrations.Migration):

    dependencies = [

        ('agendas', '0019_timeperiodexception'),

    ]

    operations = [

        migrations.AddField(

            model_name='desk',

            name='timeperiod_exceptions_remote_url',

            field=models.URLField(verbose_name='URL to fetch time period exceptions from', blank=True),

        ),

        migrations.AddField(

            model_name='timeperiodexception',

            name='creation_date',

            field=models.DateTimeField(default=datetime.datetime(2017, 10, 19, 17, 29, 53, 280020, tzinfo=utc), auto_now=True),

            preserve_default=False,

        ),

        migrations.AddField(

            model_name='timeperiodexception',

            name='external_id',

            field=models.CharField(max_length=256, null=True, verbose_name='External ID'),

        ),

    ]

import requests

from django.utils.timezone import localtime, now, make_aware, make_naive, is_aware

    timeperiod_exceptions_remote_url = models.URLField(_('URL to fetch time period exceptions from'),

                                    blank=True)

    def create_timeperiod_exceptions_from_remote_ics(self, url):

        try:

            response = requests.get(url)

            response.raise_for_status()

        except requests.HTTPError as e:

            raise ICSError(_('Failed to retrieve remote calendar (HTTP error %s).') % e.response.status_code)

        except requests.RequestException as e:

            raise ICSError(_('Failed to retrieve remote calendar (%s).') % e)

        return self.create_timeperiod_exceptions_from_ics(response.text, store_uid=True)

    def remove_timeperiod_exceptions_from_remote_ics(self):

        TimePeriodException.objects.filter(external_id__startswith='desk-%s:' % self.id).delete()

    def create_timeperiod_exceptions_from_ics(self, data, store_uid=False):

            TimePeriodException.objects.filter(external_id__startswith='desk-%s:' % self.id).delete()

            return total_created

            update_time = now()

                if store_uid:

                    event['external_id'] = 'desk-%s:%s' % (self.id, vevent.contents['uid'][0].value)

                    if not is_aware(start_dt):

                        event['start_datetime'] = make_aware(start_dt)

                    else:

                        event['start_datetime'] = start_dt

                if not is_aware(end_dt):

                    event['end_datetime'] = make_aware(end_dt)

                else:

                    event['end_datetime'] = end_dt

                obj, created = TimePeriodException.objects.update_or_create(desk=self, label=summary, defaults=event)

                # return total_created

            # delete all outdated exceptions from remote calendar

            TimePeriodException.objects.filter(creation_date__lt=update_time,

                                external_id__startswith='desk-%s:' % self.id).delete()

    external_id = models.CharField(_('External ID'), max_length=256, null=True)

    creation_date = models.DateTimeField(auto_now=True)

import requests

                                   TimePeriodException, ICSError)

        exclude = ['slug', 'timeperiod_exceptions_remote_url']

        exclude = ['timeperiod_exceptions_remote_url']

    ics_file = forms.FileField(label=_('ICS File'), required=False,

    ics_url = forms.URLField(label=_('URL'), required=False,

                               help_text=_('URL to remote calendar which will be synchronised hourly'))

  <p class="notice">{% trans "You can upload a file or specify an address to a remote calendar." %}</p>

    def get_initial(self):

        return {'ics_url': self.get_object().timeperiod_exceptions_remote_url}

        exceptions = None

            if form.cleaned_data['ics_file']:

                exceptions = form.instance.create_timeperiod_exceptions_from_ics(form.cleaned_data['ics_file'])

            elif form.cleaned_data['ics_url']:

                exceptions = form.instance.create_timeperiod_exceptions_from_remote_ics(form.cleaned_data['ics_url'])

            else:

                form.instance.remove_timeperiod_exceptions_from_remote_ics()

        form.instance.timeperiod_exceptions_remote_url = form.cleaned_data['ics_url']

        form.instance.save()

        if exceptions is not None:

            message = ungettext('An exception has been imported.',

                                '%(count)d exceptions have been imported.', exceptions)

            message = message % {'count': exceptions}

            messages.info(self.request, message)

MAILTO=root

0 * * * * /sbin/runuser -u chrono /usr/bin/chrono-manage -- tenant_command sync_desks_timeperiod_exceptions --all-tenants

    python-intervaltree,

    python-requests

requests

vobject

        'vobject',

        'requests'

import logging

import mock

import requests

from django.core.management import call_command

from django.core.management.base import CommandError

DTSTART:20170830T180800Z

DTEND:20170831T223400Z

SUMMARY:New Year's Eve

@mock.patch('chrono.agendas.models.requests.get')

def test_timeperiodexception_creation_from_remote_ics(mocked_get):

    agenda = Agenda(label=u'Test 8 agenda')

    agenda.save()

    desk = Desk(label='Test 8 desk', agenda=agenda)

    desk.save()

    mocked_response = mock.Mock()

    mocked_response.text = ICS_SAMPLE

    mocked_get.return_value = mocked_response

    exceptions_count = desk.create_timeperiod_exceptions_from_remote_ics('http://example.com/sample.ics')

    assert exceptions_count == 2

    mocked_response.text = ICS_SAMPLE_WITH_NO_EVENTS

    mocked_get.return_value = mocked_response

    exceptions_count = desk.create_timeperiod_exceptions_from_remote_ics('http://example.com/sample.ics')

    assert exceptions_count == 0

    TimePeriodException.objects.filter(external_id='desk-%s:' % desk.id).count() == 0

@mock.patch('chrono.agendas.models.requests.get')

def test_timeperiodexception_creation_from_unreachable_remote_ics(mocked_get):

    agenda = Agenda(label=u'Test 9 agenda')

    desk = Desk(label='Test 9 desk', agenda=agenda)

    mocked_response = mock.Mock()

    mocked_response.text = ICS_SAMPLE

    mocked_get.return_value = mocked_response

    def mocked_requests_connection_error(*args, **kwargs):

        raise requests.ConnectionError('unreachable')

    mocked_get.side_effect = mocked_requests_connection_error

        exceptions_count = desk.create_timeperiod_exceptions_from_remote_ics('http://example.com/sample.ics')

    assert str(e.value) == "Failed to retrieve remote calendar (unreachable)."

@mock.patch('chrono.agendas.models.requests.get')

def test_timeperiodexception_creation_from_forbidden_remote_ics(mocked_get):

    agenda = Agenda(label=u'Test 10 agenda')

    agenda.save()

    desk = Desk(label='Test 10 desk', agenda=agenda)

    desk.save()

    mocked_response = mock.Mock()

    mocked_response.status_code = 403

    mocked_get.return_value = mocked_response

    def mocked_requests_http_forbidden_error(*args, **kwargs):

        raise requests.HTTPError(response=mocked_response)

    mocked_get.side_effect = mocked_requests_http_forbidden_error

    with pytest.raises(ICSError) as e:

        exceptions_count = desk.create_timeperiod_exceptions_from_remote_ics('http://example.com/sample.ics')

    assert str(e.value) == "Failed to retrieve remote calendar (HTTP error 403)."

@mock.patch('chrono.agendas.models.requests.get')

def test_sync_desks_timeperiod_exceptions_from_ics(mocked_get, caplog):

    agenda = Agenda(label=u'Test 11 agenda')

    agenda.save()

    desk = Desk(label='Test 11 desk', agenda=agenda, timeperiod_exceptions_remote_url='http:example.com/sample.ics')

    desk.save()

    mocked_response = mock.Mock()

    mocked_response.status_code = 403

    mocked_get.return_value = mocked_response

    def mocked_requests_http_forbidden_error(*args, **kwargs):

        raise requests.HTTPError(response=mocked_response)

    mocked_get.side_effect = mocked_requests_http_forbidden_error

    with pytest.raises(CommandError) as e:

        call_command('sync_desks_timeperiod_exceptions')

    assert str(e.value) == 'unable to create timeperiod exceptions for "Test 11 desk": Failed to retrieve remote calendar (HTTP error 403).'

@mock.patch('chrono.agendas.models.requests.get')

def test_sync_desks_timeperiod_exceptions_from_changing_ics(mocked_get, caplog):

    agenda = Agenda(label=u'Test 11 agenda')

    agenda.save()

    desk = Desk(label='Test 11 desk', agenda=agenda, timeperiod_exceptions_remote_url='http:example.com/sample.ics')

    desk.save()

    mocked_response = mock.Mock()

    mocked_response.text = ICS_SAMPLE

    mocked_get.return_value = mocked_response

    call_command('sync_desks_timeperiod_exceptions')

    assert TimePeriodException.objects.filter(desk=desk).count() == 2

    mocked_response = mock.Mock()

    mocked_response.text = """BEGIN:VCALENDAR

VERSION:2.0

PRODID:-//foo.bar//EN

BEGIN:VEVENT

DTSTAMP:20180824T082855Z

UID:new-and-unique-uid

DTSTART:20180831T170800Z

DTEND:20180831T203400Z

SUMMARY:Wonderfull event

END:VEVENT

END:VCALENDAR"""

    mocked_get.return_value = mocked_response

    call_command('sync_desks_timeperiod_exceptions')

    assert TimePeriodException.objects.filter(desk=desk).count() == 1

    exception = TimePeriodException.objects.get(desk=desk)

    assert exception.external_id == 'desk-%s:new-and-unique-uid' % desk.id

import mock

import requests

    assert "You can upload a file or specify an address to a remote calendar." in resp

    resp = resp.form.submit(status=302)

    resp = app.get('/manage/agendas/%d/' % agenda.pk)

    resp = resp.click('upload')

SUMMARY:New Year's Eve

SUMMARY:New Year's Eve

    assert 'Event "New Year's Eve" has no start date.' in resp.content

    resp = resp.form.submit(status=302)

    resp = resp.follow()

    assert "0 exceptions have been imported." in resp.content

SUMMARY:New Year's Eve

    resp = app.get('/manage/agendas/%d/' % agenda.pk)

    resp = resp.click('upload')

    assert TimePeriodException.objects.filter(desk=desk).count() == 1

@mock.patch('chrono.agendas.models.requests.get')

def test_agenda_import_time_period_exception_with_remote_ics(mocked_get, app, admin_user):

    agenda = Agenda.objects.create(label='New Example', kind='meetings')

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

    MeetingType(agenda=agenda, label='Bar').save()

    login(app)

    resp = app.get('/manage/agendas/%d/' % agenda.pk)

    assert 'Import exceptions from .ics' not in resp.content

    TimePeriod.objects.create(weekday=1, desk=desk,

                start_time=datetime.time(10, 0), end_time=datetime.time(12, 0))

    resp = app.get('/manage/agendas/%d/' % agenda.pk)

    resp = resp.click('upload')

    assert 'ics_file' in resp.form.fields

    assert 'ics_url' in resp.form.fields

    resp.form['ics_url'] = 'http://example.com/foo.ics'

    mocked_response = mock.Mock()

    mocked_response.text =  """BEGIN:VCALENDAR

VERSION:2.0

PRODID:-//foo.bar//EN

BEGIN:VEVENT

UID:random-event-id

DTSTART:20180101

DTEND:20180101

SUMMARY:New Year's Eve

END:VEVENT

END:VCALENDAR"""

    mocked_get.return_value = mocked_response

    resp = resp.form.submit(status=302)

    assert TimePeriodException.objects.filter(desk=desk).count() == 1

    exception = TimePeriodException.objects.get(desk=desk)

    assert exception.external_id == 'desk-%s:random-event-id' % desk.id

    resp = app.get('/manage/agendas/%d/' % agenda.pk)

    resp = resp.click('upload')

    resp.form['ics_url'] = ''

    resp = resp.form.submit(status=302)

    assert TimePeriodException.objects.filter(desk=desk,

                    external_id='desk-%s:random-event-id' % desk.id).count() == 0

@mock.patch('chrono.agendas.models.requests.get')

def test_agenda_update_time_period_exception_from_remote_ics(mocked_get, app, admin_user):

    agenda = Agenda.objects.create(label='New Example', kind='meetings')

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

    MeetingType(agenda=agenda, label='Bar').save()

    login(app)

    resp = app.get('/manage/agendas/%d/' % agenda.pk)

    assert 'Import exceptions from .ics' not in resp.content

    TimePeriod.objects.create(weekday=1, desk=desk,

                start_time=datetime.time(10, 0), end_time=datetime.time(12, 0))

    resp = app.get('/manage/agendas/%d/' % agenda.pk)

    resp = resp.click('upload')

    resp.form['ics_url'] = 'http://example.com/foo.ics'

    mocked_response = mock.Mock()

    mocked_response.text =  """BEGIN:VCALENDAR

VERSION:2.0

PRODID:-//foo.bar//EN

BEGIN:VEVENT

UID:first-eventrandom-event-id

DTSTART:20180101

DTEND:20180101

SUMMARY:First test event

END:VEVENT

BEGIN:VEVENT

UID:second-eventrandom-event-id

DTSTART:20190101

DTEND:20190101

SUMMARY:Second test event

END:VEVENT

END:VCALENDAR"""

    mocked_get.return_value = mocked_response

    resp = resp.form.submit(status=302)

    assert TimePeriodException.objects.filter(desk=desk).count() == 2

    resp = app.get('/manage/agendas/%d/' % agenda.pk)

    resp = resp.click('upload')

    resp.form['ics_url'] = 'http://example.com/foo.ics'

    mocked_response.text = """BEGIN:VCALENDAR

VERSION:2.0

PRODID:-//foo.bar//EN

BEGIN:VEVENT

UID:secord-eventrandom-event-id

DTSTART:20190101

DTEND:20190101

SUMMARY:Second test event

END:VEVENT

END:VCALENDAR"""

    mocked_get.return_value = mocked_response

    resp = resp.form.submit(status=302)

    assert TimePeriodException.objects.filter(desk=desk).count() == 1

@mock.patch('chrono.agendas.models.requests.get')

def test_agenda_import_time_period_exception_from_remote_ics_with_connection_error(mocked_get, app, admin_user):

    agenda = Agenda.objects.create(label='New Example', kind='meetings')

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

    MeetingType(agenda=agenda, label='Bar').save()

    login(app)

    resp = app.get('/manage/agendas/%d/' % agenda.pk)

    assert 'Import exceptions from .ics' not in resp.content

    TimePeriod.objects.create(weekday=1, desk=desk,

                start_time=datetime.time(10, 0), end_time=datetime.time(12, 0))

    resp = app.get('/manage/agendas/%d/' % agenda.pk)

    resp = resp.click('upload')

    assert 'ics_file' in resp.form.fields

    assert 'ics_url' in resp.form.fields

    resp.form['ics_url'] = 'http://example.com/foo.ics'

    mocked_response = mock.Mock()

    mocked_get.return_value = mocked_response

    def mocked_requests_connection_error(*args, **kwargs):

        raise requests.exceptions.ConnectionError('unreachable')

    mocked_get.side_effect = mocked_requests_connection_error

    resp = resp.form.submit(status=200)

    assert 'Failed to retrieve remote calendar (unreachable).' in resp.content

@mock.patch('chrono.agendas.models.requests.get')

def test_agenda_import_time_period_exception_from_forbidden_remote_ics(mocked_get, app, admin_user):

    agenda = Agenda.objects.create(label='New Example', kind='meetings')

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

    MeetingType(agenda=agenda, label='Bar').save()

    login(app)

    resp = app.get('/manage/agendas/%d/' % agenda.pk)

    assert 'Import exceptions from .ics' not in resp.content

    TimePeriod.objects.create(weekday=1, desk=desk,

                start_time=datetime.time(10, 0), end_time=datetime.time(12, 0))

    resp = app.get('/manage/agendas/%d/' % agenda.pk)

    resp = resp.click('upload')

    resp.form['ics_url'] = 'http://example.com/foo.ics'

    mocked_response = mock.Mock()

    mocked_response.status_code = 403

    mocked_get.return_value = mocked_response

    def mocked_requests_http_forbidden_error(*args, **kwargs):

        raise requests.exceptions.HTTPError(response=mocked_response)

    mocked_get.side_effect = mocked_requests_http_forbidden_error

    resp = resp.form.submit(status=200)

    assert 'Failed to retrieve remote calendar (HTTP error 403).' in resp.content

@mock.patch('chrono.agendas.models.requests.get')

def test_agenda_import_time_period_exception_from_remote_ics_with_ssl_error(mocked_get, app, admin_user):

    agenda = Agenda.objects.create(label='New Example', kind='meetings')

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

    MeetingType(agenda=agenda, label='Bar').save()

    login(app)

    resp = app.get('/manage/agendas/%d/' % agenda.pk)

    assert 'Import exceptions from .ics' not in resp.content

    TimePeriod.objects.create(weekday=1, desk=desk,

                start_time=datetime.time(10, 0), end_time=datetime.time(12, 0))

    resp = app.get('/manage/agendas/%d/' % agenda.pk)

    resp = resp.click('upload')

    resp.form['ics_url'] = 'https://example.com/foo.ics'

    mocked_response = mock.Mock()

    mocked_get.return_value = mocked_response

    def mocked_requests_http_ssl_error(*args, **kwargs):

        raise requests.exceptions.SSLError('SSL error')

    mocked_get.side_effect = mocked_requests_http_ssl_error

    resp = resp.form.submit(status=200)

    assert 'Failed to retrieve remote calendar (SSL error).' in resp.content