Produits Entr'ouvert » Chrono

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

from __future__ import unicode_literals

from django.db import migrations, models

class Migration(migrations.Migration):

    dependencies = [

        ('agendas', '0019_timeperiodexception'),

    ]

    operations = [

        migrations.AddField(

            model_name='desk',

            name='timeperiod_exceptions_remote_url',

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

        ),

    ]

import requests

def get_remote_calendar(url):

    try:

        response = requests.get(url)

        response.raise_for_status()

    except requests.exceptions.HTTPError as e:

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

    except requests.exceptions.ConnectionError:

        raise ICSError(_('Failed to retrieve remote calendar (connection error).'))

    except requests.exceptions.Timeout:

        raise ICSError(_('Failed to retrieve remote calendar (HTTP timeout).'))

    return response.text

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

                                    null=True, blank=True)

    def create_timeperiod_exceptions_from_remote_ics(self, url):

        return self.create_timeperiod_exceptions_from_ics(get_remote_calendar(url))

import requests

                                   TimePeriodException, ICSError, get_remote_calendar)

class DeskForm(forms.ModelForm):

        exclude = []

    def is_valid(self):

        if not self.data['timeperiod_exceptions_remote_url']:

            return super(DeskForm, self).is_valid()

        try:

            get_remote_calendar(self.data['timeperiod_exceptions_remote_url'])

        except ICSError as e:

            self.add_error('timeperiod_exceptions_remote_url', e)

            return False

        return super(DeskForm, self).is_valid()

class NewDeskForm(DeskForm):

        exclude = ['slug']

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

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

    def clean(self):

        cleaned_data = super(ExceptionsImportForm, self).clean()

        if not cleaned_data['ics_file'] and not cleaned_data['ics_url']:

            raise ValidationError(_('A file or an url should be filled.'))

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

import logging

from django.core.management.base import BaseCommand

from chrono.agendas.models import Desk, ICSError

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.filter(timeperiod_exceptions_remote_url__isnull=False):

            try:

                desk.create_timeperiod_exceptions_from_remote_ics(desk.timeperiod_exceptions_remote_url)

            except ICSError as e:

                logger.warning(u'unable to create timeperiod exceptions for "%s": %s' % (desk, e))

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

            if form.cleaned_data['ics_file']:

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

            else:

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

#!/bin/sh

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

    python-intervaltree,

    python-requests

requests

        'vobject',

        'requests'

import mock

import requests

import logging

from django.core.management import call_command

@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

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

def test_timeperiodexception_creation_from_unreachable_remote_ics(mocked_get):

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

    agenda.save()

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

    desk.save()

    mocked_response = mock.Mock()

    mocked_response.content.return_value = ICS_SAMPLE

    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

    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 (connection error)."

@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.exceptions.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_timeperiodexception_creation_from_remote_ics_with_timeout_error(mocked_get):

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

    agenda.save()

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

    desk.save()

    mocked_response = mock.Mock()

    mocked_get.return_value = mocked_response

    def mocked_requests_http_timeout_error(*args, **kwargs):

        raise requests.exceptions.Timeout(response=mocked_response)

    mocked_get.side_effect = mocked_requests_http_timeout_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 timeout)."

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

def test_timeperiodexception_creation_from_remote_ics_with_ssl_error(mocked_get):

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

    agenda.save()

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

    desk.save()

    mocked_response = mock.Mock()

    mocked_get.return_value = mocked_response

    def mocked_requests_http_ssl_error(*args, **kwargs):

        raise requests.exceptions.SSLError(response=mocked_response)

    mocked_get.side_effect = mocked_requests_http_ssl_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 (connection error)."

@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.exceptions.HTTPError(response=mocked_response)

    mocked_get.side_effect = mocked_requests_http_forbidden_error

    call_command('sync_desks_timeperiod_exceptions')

    records = caplog.records()

    assert len(records) == 1

    for record in records:

        assert record.name == 'chrono.manager.management.commands.sync_desks_timeperiod_exceptions'

        assert record.levelno == logging.WARNING

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

import mock

import requests

@mock.patch('chrono.manager.forms.requests.get')

def test_meetings_agenda_add_desk_with_exceptions_url(mocked_get, app, admin_user):

    app = login(app)

    resp = app.get('/manage/', status=200)

    resp = resp.click('New')

    resp.form['label'] = 'Foo bar'

    resp.form['kind'] = 'meetings'

    resp = resp.form.submit()

    agenda = Agenda.objects.get(slug='foo-bar')

    resp = app.get('/manage/agendas/%s/' % agenda.id, status=200)

    ICS_SAMPLE = """BEGIN:VCALENDAR

VERSION:2.0

PRODID:-//foo.bar//EN

BEGIN:VEVENT

DTSTART:20180101

DTEND:20180101

SUMMARY:New eve

END:VEVENT

END:VCALENDAR"""

    mocked_response = mock.Mock()

    mocked_response.content.return_value = ICS_SAMPLE

    mocked_get.return_value = mocked_response

    resp = resp.click('New Desk')

    resp.form['label'] = 'Desk A'

    resp.form['timeperiod_exceptions_remote_url'] = 'http://nowhere.com/unknown.ics'

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

@mock.patch('chrono.manager.forms.requests.get')

def test_meetings_agenda_add_desk_with_non_existing_exceptions_url(mocked_get, app, admin_user):

    app = login(app)

    resp = app.get('/manage/', status=200)

    resp = resp.click('New')

    resp.form['label'] = 'Foo bar'

    resp.form['kind'] = 'meetings'

    resp = resp.form.submit()

    agenda = Agenda.objects.get(slug='foo-bar')

    resp = app.get('/manage/agendas/%s/' % agenda.id, status=200)

    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.click('New Desk')

    resp.form['label'] = 'Desk A'

    resp.form['timeperiod_exceptions_remote_url'] = 'http://nowhere.com/unknown.ics'

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

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

@mock.patch('chrono.manager.forms.requests.get')

def test_meetings_agenda_add_desk_with_unreachable_exceptions_url(mocked_get, app, admin_user):

    app = login(app)

    resp = app.get('/manage/', status=200)

    resp = resp.click('New')

    resp.form['label'] = 'Foo bar'

    resp.form['kind'] = 'meetings'

    resp = resp.form.submit()

    agenda = Agenda.objects.get(slug='foo-bar')

    resp = app.get('/manage/agendas/%s/' % agenda.id, status=200)

    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.click('New Desk')

    resp.form['label'] = 'Desk A'

    resp.form['timeperiod_exceptions_remote_url'] = 'http://nowhere.com/unknown.ics'

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

    assert 'Failed to retrieve remote calendar (connection error).' in resp.text

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

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

    assert "A file or an url should be filled." in resp.body

@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 (connection error).' 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_timeout_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'] = 'http://example.com/foo.ics'

    mocked_response = mock.Mock()

    mocked_get.return_value = mocked_response

    def mocked_requests_http_timeout_error(*args, **kwargs):

        raise requests.exceptions.Timeout(response=mocked_response)

    mocked_get.side_effect = mocked_requests_http_timeout_error

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

    assert 'Failed to retrieve remote calendar (HTTP timeout).' 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'] = 'http://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

    mocked_get.side_effect = mocked_requests_http_ssl_error

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

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