caldav: add caldav connector (#87227) #472

Merged
yweber merged 1 commits from wip/87227-create-caldav-connector into main 2024-03-05 14:34:29 +01:00
8 changed files with 793 additions and 0 deletions

1
debian/control vendored
View File

@ -16,6 +16,7 @@ Architecture: all
Depends: ghostscript,
pdftk,
poppler-utils,
python3-caldav,
python3-cmislib,
python3-cryptography,
python3-dateutil,

View File

View File

@ -0,0 +1,47 @@
# Generated by Django 3.2.18 on 2024-02-20 15:41
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
('base', '0030_resourcelog_base_resour_appname_298cbc_idx'),
]
operations = [
migrations.CreateModel(
name='CalDAV',
fields=[
(
'id',
models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID'),
),
('title', models.CharField(max_length=50, verbose_name='Title')),
('slug', models.SlugField(unique=True, verbose_name='Identifier')),
('description', models.TextField(verbose_name='Description')),
(
'dav_url',
models.URLField(
help_text='DAV root URL (such as https://test.egw/groupdav.php/)',
verbose_name='DAV root URL',
),
),
('dav_login', models.CharField(max_length=128, verbose_name='DAV username')),
('dav_password', models.CharField(max_length=512, verbose_name='DAV password')),
(
'users',
models.ManyToManyField(
blank=True,
related_name='_caldav_caldav_users_+',
related_query_name='+',
to='base.ApiUser',
),
),
],
options={
'verbose_name': 'CalDAV',
},
),
]

View File

@ -0,0 +1,305 @@
# passerelle - uniform access to multiple data sources and services
# Copyright (C) 2024 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 functools
import urllib.parse
import caldav
import requests
from django.db import models
from django.utils.dateparse import parse_date, parse_datetime
from django.utils.translation import gettext_lazy as _
from passerelle.base.models import BaseResource
from passerelle.utils.api import endpoint
from passerelle.utils.conversion import exception_to_text
from passerelle.utils.jsonresponse import APIError
EVENT_SCHEMA_PART = {
'type': 'object',
'description': _('Ical event properties ( VEVENT RFC 5545 3.6.1 )'),
'required': ['DTSTART', 'DTEND'],
'properties': {
'DTSTART': {
'type': 'string',
'description': _('Event start (included) ISO-8601 date-time or date (for allday event)'),
},
'DTEND': {
'type': 'string',
'description': _('Event end (excluded) ISO-8601 date-time or date (for allday event)'),
},
'SUMMARY': {
'type': 'string',
'description': 'RFC 5545 3.8.1.12',
},
'DESCRIPTION': {
'type': 'string',
'description': 'RFC 5545 3.8.2.5',
},
'LOCATION': {
'type': 'string',
'description': 'RFC 5545 3.8.1.7',
},
'TRANSP': {
'type': 'boolean',
'description': 'Transparent if true else opaque (RFC 5545 3.8.2.7)',
},
'RRULE': {
'description': _('Recurrence rule (RFC 5545 3.8.5.3)'),
'type': 'object',
'properties': {
'FREQ': {
'type': 'string',
'enum': ['WEEKLY', 'MONTHLY', 'YEARLY'],
},
'BYDAY': {
'type': 'array',
'items': {
'type': 'string',
'enum': ['MO', 'TU', 'WE', 'TH', 'FR', 'SA', 'SU'],
},
},
'BYMONTH': {
'type': 'array',
'items': {
'type': 'integer',
'minimum': 1,
'maximum': 12,
},
},
'COUNT': {
'type': 'integer',
'minimum': 1,
},
},
},
},
}
USERNAME_PARAM = {
'description': _('The calendar\'s owner username'),
'type': 'string',
}
EVENT_UID_PARAM = {
'description': _('An event UID'),
'type': 'string',
}
# Action's request body schema
EVENT_SCHEMA = {
yweber marked this conversation as resolved Outdated

Je vois l'ambition d'avoir un truc générique mais je pense qu'il faut en premier lieu nous simplifier la vie (après tout c'est la raison d'être de ce connecteur) : permettre d'appeler avec un paramètre calendar_name (dans lequel on passera le username) et un paramètre event_id, et construire le chemin dans le connecteur.

Je vois l'ambition d'avoir un truc générique mais je pense qu'il faut en premier lieu nous simplifier la vie (après tout c'est la raison d'être de ce connecteur) : permettre d'appeler avec un paramètre calendar_name (dans lequel on passera le username) et un paramètre event_id, et construire le chemin dans le connecteur.

Le truc, c'est que si j'ai bien compris la RFC (c'est pas certain), il n'y a pas de restriction sur les URL des événements :
https://www.rfc-editor.org/rfc/rfc4791#section-5.3.2

The URL for each calendar object resource is entirely arbitrary and does not need to bear a specific relationship to the calendar object resource's iCalendar properties or other metadata.

Si je comprend bien, malgré la présence d'un UID, un évènement est identifié par son URL (d’ailleurs EGW n'arrive pas a renvoyer un évènement en fonction de son UID).
Ensuite, au final, l'utilisateur ne devrait pas avoir à construire lui même l'URL vers un évènement : l'info est renvoyé par le endpoint de création.

Mais au final oui, on peut très bien forger les URL dans le connecteur en fonction du username et de l'UID, mais ça en ferait un connecteur EGW et pas CalDAV.

Le truc, c'est que si j'ai bien compris la RFC (c'est pas certain), il n'y a pas de restriction sur les URL des événements : https://www.rfc-editor.org/rfc/rfc4791#section-5.3.2 > The URL for each calendar object resource is entirely arbitrary and does not need to bear a specific relationship to the calendar object resource's iCalendar properties or other metadata. Si je comprend bien, malgré la présence d'un UID, un évènement est identifié par son URL (d’ailleurs EGW n'arrive pas a renvoyer un évènement en fonction de son UID). Ensuite, au final, l'utilisateur ne devrait pas avoir à construire lui même l'URL vers un évènement : l'info est renvoyé par le endpoint de création. Mais au final oui, on peut très bien forger les URL dans le connecteur en fonction du username et de l'UID, mais ça en ferait un connecteur EGW et pas CalDAV.

Mais au final oui, on peut très bien forger les URL dans le connecteur en fonction du username et de l'UID, mais ça en ferait un connecteur EGW et pas CalDAV.

Yep c'est bien ça l'idée, la seule conséquence sera de rendre les choses plus simples pour le seul usage imaginé de ce connecteur.

Ce qui ne l'empêche pas de continuer à l'appeler CalDAV : en vérité aucun connecteur n'est 100% générique, toute utilisation avec une nouvelle instance du logiciel cible entraîne des devs supplémentaires, qu'on gérera ici si le besoin se présente :)

> Mais au final oui, on peut très bien forger les URL dans le connecteur en fonction du username et de l'UID, mais ça en ferait un connecteur EGW et pas CalDAV. Yep c'est bien ça l'idée, la seule conséquence sera de rendre les choses plus simples pour le seul usage imaginé de ce connecteur. Ce qui ne l'empêche pas de continuer à l'appeler CalDAV : en vérité aucun connecteur n'est 100% générique, toute utilisation avec une nouvelle instance du logiciel cible entraîne des devs supplémentaires, qu'on gérera ici si le besoin se présente :)

Ca me va :)

Ca me va :)
'$schema': 'http://json-schema.org/draft-04/schema#',
'title': _('Event description schema'),
'unflatten': True,
**EVENT_SCHEMA_PART,
}
class CalDAV(BaseResource):
dav_url = models.URLField(
blank=False,
verbose_name=_('DAV root URL'),
help_text=_('DAV root URL (such as https://test.egw/groupdav.php/)'),
)
dav_login = models.CharField(max_length=128, verbose_name=_('DAV username'), blank=False)
dav_password = models.CharField(max_length=512, verbose_name=_('DAV password'), blank=False)
category = _('Misc')
class Meta:
verbose_name = _('CalDAV')
@functools.cached_property
def dav_client(self):
'''Instanciate a caldav.DAVCLient and return the instance'''
yweber marked this conversation as resolved Outdated

Ces attributs sont à déplacer au seul endroit de leur utilisation respective :)

Ces attributs sont à déplacer au seul endroit de leur utilisation respective :)

Soit :P

Soit :P
return caldav.DAVClient(self.dav_url, username=self.dav_login, password=self.dav_password)
def check_status(self):
'''Attempt a propfind on DAV root URL'''
try:
rep = self.dav_client.propfind()
rep.find_objects_and_props()
except caldav.lib.error.AuthorizationError:
raise Exception(_('Not authorized: bad login/password ?'))
@endpoint(
name='event',
yweber marked this conversation as resolved Outdated

Il me semble que le décorateur @cached_property fait tout ça à moindre frais

Il me semble que le décorateur @cached_property fait tout ça à moindre frais

Bien vu ! Merci :)

Bien vu ! Merci :)

Tu as rajouté le décorateur mais tu n'as pas enlevé la mise en cache manuelle de l'objet via self.__client, pour moi cached_property permet de le faire sauter

Tu as rajouté le décorateur mais tu n'as pas enlevé la mise en cache manuelle de l'objet via `self.__client`, pour moi cached_property permet de le faire sauter

Navré -_- ! Merci !

Navré -_- ! Merci !
pattern='^create$',
example_pattern='create',
yweber marked this conversation as resolved Outdated

Pas besoin de rattraper, c'est géré dans BaseResource.availability()

Pas besoin de rattraper, c'est géré dans BaseResource.availability()

ok :)

ok :)
methods=['post'],
post={'request_body': {'schema': {'application/json': EVENT_SCHEMA}}},
yweber marked this conversation as resolved Outdated

Cette méthode peut-être largement simplifiée j'ai l'impression : il n'y a pas besoin que ce soit un endpoint, et il n'y a même pas besoin de rattraper les exceptions (voir ce qu'il se passe dans BaseResource.availability()).

Donc

def check_status(self):
    rep = self.dav_client.propfind()
    rep.find_objects_and_props()

me semble faire le job !

Cette méthode peut-être largement simplifiée j'ai l'impression : il n'y a pas besoin que ce soit un endpoint, et il n'y a même pas besoin de rattraper les exceptions (voir ce qu'il se passe dans BaseResource.availability()). Donc ``` def check_status(self): rep = self.dav_client.propfind() rep.find_objects_and_props() ``` me semble faire le job !

Bien vu, merci :)

Bien vu, merci :)
parameters={
'username': USERNAME_PARAM,
},
)
yweber marked this conversation as resolved Outdated

Sur la vue du connecteur ça indique comme url /caldav/egw/event/event/create alors que la bonne est juste /caldav/egw/event/create

Sur la vue du connecteur ça indique comme url ` /caldav/egw/event/event/create` alors que la bonne est juste `/caldav/egw/event/create`

En effet !

En effet !
def create_event(self, request, username, post_data):
'''Event creation endpoint'''
cal = self.get_calendar(username)
self._process_event_properties(post_data)
try:
evt = cal.save_event(**post_data)
except requests.exceptions.RequestException as expt:
raise APIError(
_('Error sending creation request to caldav server'),
data={
'expt_class': str(type(expt)),
'expt': str(expt),
vdeniaud marked this conversation as resolved Outdated

Idéalement il faudrait nommer cette clé event_id (un œil de CPF a de forte chances de glisser sur u de uid qui n'est pas une dénomination habituelle)

Idéalement il faudrait nommer cette clé event_id (un œil de CPF a de forte chances de glisser sur u de uid qui n'est pas une dénomination habituelle)

Je fais ça :)

J'avais peut être mal compris, mais il me semblait qu'il vallait mieux éviter les _ dans les retour à wcs (vu que ce caractère est dédié a la séparation de "namespace").

Je fais ça :) J'avais peut être mal compris, mais il me semblait qu'il vallait mieux éviter les `_` dans les retour à wcs (vu que ce caractère est dédié a la séparation de "namespace").
'username': username,
},
)
return {'data': {'event_id': evt.id}}
# Patch do not support request_body validation, using post instead
@endpoint(
name='event',
pattern='^update$',
example_pattern='update',
methods=['post'],
yweber marked this conversation as resolved Outdated

Pareil event_id ici serait moins étonnant

Pareil event_id ici serait moins étonnant

ok, c'est fait :)

ok, c'est fait :)
post={'request_body': {'schema': {'application/json': EVENT_SCHEMA}}},
parameters={
'username': USERNAME_PARAM,
'event_id': EVENT_UID_PARAM,
yweber marked this conversation as resolved Outdated

Pas besoin de variable intermédiaire ici

Pas besoin de variable intermédiaire ici
},
)
def update_event(self, request, username, event_id, post_data):
'''Event update endpoint'''
self._process_event_properties(post_data)
ical = self.get_event(username, event_id)
vevent = ical.icalendar_instance.walk('VEVENT')
if not len(vevent) == 1:
raise APIError(
_('Given event (user:%r uid:%r) do not contains VEVENT component') % (username, event_id),
data={
'username': username,
'event_uid': event_id,
'VEVENT': str(vevent),
},
)
vevent = vevent[0]
# vevent.update(post_data) do not convert values as expected
for k, v in post_data.items():
vevent[k] = v
try:
# do not use ical.save(no_create=True) : no_create fails on some calDAV
ical.save()
except requests.exceptions.RequestException as expt:
raise APIError(
_('Error sending update request to caldav server'),
yweber marked this conversation as resolved Outdated

Je trouverais mieux sans abréviation ici, juste nommer la variable vevent

Je trouverais mieux sans abréviation ici, juste nommer la variable `vevent`

Fait

Fait
data={
'expt_class': str(type(expt)),
'expt': str(expt),
'username': username,
'event_id': event_id,
},
)
return {'data': {'event_id': ical.id}}
yweber marked this conversation as resolved Outdated

Je pense qu'on peut faire sans ces commentaires

Je pense qu'on peut faire sans ces commentaires

Je trouve que la tentation de faire un vevent.update() et de passer le no_create=True un peu forte ;)

Ca te vas si je les réduis à une seule ligne de commentaire ?

Je trouve que la tentation de faire un `vevent.update()` et de passer le `no_create=True` un peu forte ;) Ca te vas si je les réduis à une seule ligne de commentaire ?

Il y a une tentation pour toi, mais pas pour les autres lecteurs de ce code à qui ces commentaires vont plutôt alourdir la lecture : ignorance is bliss :)

Mais une ligne c'est déjà mieux en effet, tu peux laisser comme ça

Il y a une tentation pour toi, mais pas pour les autres lecteurs de ce code à qui ces commentaires vont plutôt alourdir la lecture : ignorance is bliss :) Mais une ligne c'est déjà mieux en effet, tu peux laisser comme ça
@endpoint(
name='event',
pattern='^delete$',
example_pattern='delete',
methods=['delete'],
parameters={
'username': USERNAME_PARAM,
'event_id': EVENT_UID_PARAM,
},
)
def delete_event(self, request, username, event_id):
ical = self.get_event(username, event_id)
try:
ical.delete()
except requests.exceptions.RequestException as expt:
raise APIError(
_('Error sending deletion request to caldav server'),
yweber marked this conversation as resolved Outdated

Cette méthode est juste utilisée pour les retours du connecteur, mais puisqu'on ne prend plus de chemin en entrée, je pense qu'on peut les retirer de la sortie également

Cette méthode est juste utilisée pour les retours du connecteur, mais puisqu'on ne prend plus de chemin en entrée, je pense qu'on peut les retirer de la sortie également

ok :)

ok :)
data={
'expt_class': str(type(expt)),
'expt': str(expt),
'username': username,
'event_id': event_id,
},
)
return {}
def get_event(self, username, event_uid):
'''Fetch an event given a username and an event_uid
Arguments:
- username: Calendar owner's username
- event_uid: The event's UID
Returns an caldav.Event instance
'''
event_path = '%s/calendar/%s.ics' % (urllib.parse.quote(username), urllib.parse.quote(str(event_uid)))
yweber marked this conversation as resolved Outdated

Ça pourrait plutôt être un commentaire à côté de la ligne en question url = str(obj.url.canonical()) # normalize URL

Ça pourrait plutôt être un commentaire à côté de la ligne en question `url = str(obj.url.canonical()) # normalize URL`

J'avais mis ce commentaire vu que j'étais pas sur de moi. La classe URL de caldav indique en commentaire :

It's used internally in the library, end users should not need to know anything about this class.

En réalité, en testant et en y pensant, c'est un truc "au cas ou", plutôt inutile à priori...

J'avais mis ce commentaire vu que j'étais pas sur de moi. La classe URL de caldav indique en commentaire : > It's used internally in the library, end users should not need to know anything about this class. En réalité, en testant et en y pensant, c'est un truc "au cas ou", plutôt inutile à priori...
cal = self.get_calendar(username)
try:
ical = cal.event_by_url(event_path)
except caldav.lib.error.DAVError as expt:
yweber marked this conversation as resolved Outdated

Peut-être à faire une seule fois dans le __init__ ?

Peut-être à faire une seule fois dans le `__init__` ?
raise APIError(
vdeniaud marked this conversation as resolved Outdated

Peut-être que la méthode cal.object_by_uid fonctionne et permet de ne pas avoir à construire le chemin vers l'évènement ?

Peut-être que la méthode cal.object_by_uid fonctionne et permet de ne pas avoir à construire le chemin vers l'évènement ?

Tristement non :/ Même la RFC explique qu'il faut s'attendre à ce que ça ne fonctionne pas ...

EDIT : Désolé, je retrouve plus la source... La seule trace c'est ce warning dans la doc de caldav : https://caldav.readthedocs.io/en/latest/caldav/objects.html#caldav.objects.CalendarObjectResource.save

Some servers don’t support searching for an object uid without explicitly specifying what kind of object it should be, hence obj_type can be passed.

Mais les tests que j'avais fais autour de object_by_uid (y compris en passant un comp_filter) n'ont rien donnés sur EGW :/

Tristement non :/ Même la RFC explique qu'il faut s'attendre à ce que ça ne fonctionne pas ... EDIT : Désolé, je retrouve plus la source... La seule trace c'est ce warning dans la doc de caldav : https://caldav.readthedocs.io/en/latest/caldav/objects.html#caldav.objects.CalendarObjectResource.save > Some servers don’t support searching for an object uid without explicitly specifying what kind of object it should be, hence obj_type can be passed. Mais les tests que j'avais fais autour de object_by_uid (y compris en passant un comp_filter) n'ont rien donnés sur EGW :/
_('Unable to get event %r in calendar owned by %r') % (event_uid, username),
data={
'expt': exception_to_text(expt),
'expt_cls': str(type(expt)),
'username': username,
'event_uid': event_uid,
},
)
except requests.exceptions.RequestException as expt:
raise APIError(
_('Unable to communicate with caldav server while fetching event'),
data={
'expt': exception_to_text(expt),
'expt_class': str(type(expt)),
'username': username,
'event_uid': event_uid,
},
)
return ical
def get_calendar(self, username):
'''Given a username returns the associated calendar set
Arguments:
- username: Calendar owner's username
Returns A caldav.Calendar instance
yweber marked this conversation as resolved Outdated

On peut faire un sort à cette méthode en construisant l'URL au seul endroit où il y en a besoin il me semble, dans get_calendar

On peut faire un sort à cette méthode en construisant l'URL au seul endroit où il y en a besoin il me semble, dans get_calendar

Et la même chose pour get_event_path du coup :)

Et la même chose pour get_event_path du coup :)
Note: do not raise any caldav exception before a method trying to make
a request is called
'''
path = '%s/calendar' % urllib.parse.quote(username)
calendar = caldav.Calendar(client=self.dav_client, url=path)
return calendar
def _process_event_properties(self, data):
'''Handles verification & convertion of event properties
@note Modify given data dict inplace
'''
if 'TRANSP' in data:
data['TRANSP'] = 'TRANSPARENT' if data['TRANSP'] else 'OPAQUE'
for dt_field in ('DTSTART', 'DTEND'):
yweber marked this conversation as resolved Outdated

Ici tu as constaté que la validation induite par la clé « pattern » dans le schéma json n'était pas appliquée ? Ça paraît improbable vu le nombre de connecteurs qui utilisent cette fonctionnalité non ?

Ici tu as constaté que la validation induite par la clé « pattern » dans le schéma json n'était pas appliquée ? Ça paraît improbable vu le nombre de connecteurs qui utilisent cette fonctionnalité non ?

Non pas les pattern dans le schéma JSON, dans la spécification des paramètres de la query string.

Non pas les pattern dans le schéma JSON, dans la spécification des paramètres de la query string.

Ah oui, c'est le nom SCHEMA qui m'avait induit en erreur, ça fait penser à du JSON alors que non. Mais cette partie devrait disparaître si on passe direct nom du calendrier + id de l'événement

Ah oui, c'est le nom SCHEMA qui m'avait induit en erreur, ça fait penser à du JSON alors que non. Mais cette partie devrait disparaître si on passe direct nom du calendrier + id de l'événement

Exactement

Exactement
value = data[dt_field]
try:
data[dt_field] = parse_date(value) or parse_datetime(value)
except ValueError:
data[dt_field] = None
if not data[dt_field]:
raise APIError(
_('Unable to convert field %s=%r: not a valid date nor date-time') % (dt_field, value),
http_status=400,
)

View File

@ -143,6 +143,7 @@ INSTALLED_APPS = (
'passerelle.apps.base_adresse',
'passerelle.apps.bbb',
'passerelle.apps.bdp',
'passerelle.apps.caldav',
'passerelle.apps.cartads_cs',
'passerelle.apps.choosit',
'passerelle.apps.cityweb',

View File

@ -138,6 +138,7 @@ setup(
scripts=['manage.py'],
include_package_data=True,
install_requires=[
'caldav',
'django >= 3.2, <3.3',
'django-model-utils<4.3',
'requests',

438
tests/test_caldav.py Normal file
View File

@ -0,0 +1,438 @@
import datetime
import urllib
from unittest.mock import Mock, patch
import caldav
import icalendar
import pytest
import responses
from django.contrib.contenttypes.models import ContentType
from requests.exceptions import ConnectionError, ConnectTimeout, ReadTimeout
import tests.utils
from passerelle.apps.caldav.models import CalDAV
from passerelle.base.models import AccessRight, ApiUser
DAV_URL = 'http://test.caldav.notld/somedav/'
EVENT_PATH_FMT = '%(username)s/calendar/%(uid)s.ics'
PROPFIND_REPLY = '''<?xml version="1.0" encoding="utf-8"?>
<D:multistatus xmlns:D="DAV:">
<D:response xmlns:ns0="urn:uuid:02481234-abcd-4321-abcd-00aa00bb00cc/" xmlns:ns2="urn:ietf:params:xml:ns:carddav">
<D:href>/somedave/</D:href>
<D:propstat>
<D:prop>
<D:displayname>Test Caldav</D:displayname>
<D:owner/>
<D:resourcetype><D:collection /></D:resourcetype>
<D:getcontenttype>httpd/unix-directory</D:getcontenttype>
<D:current-user-principal><D:href>/somedav/principals/users/apiaccess/</D:href></D:current-user-principal>
<D:principal-collection-set><D:href>/somedav/principals/</D:href></D:principal-collection-set>
<D:getetag>"none"</D:getetag>
<D:getcontentlength/>
<D:getlastmodified/>
<ns2:addressbook-home-set><D:href>/somedav/apiaccess/</D:href></ns2:addressbook-home-set>
<ns2:principal-address/>
<ns2:directory-gateway><D:href>/somedav/addressbook/</D:href></ns2:directory-gateway>
</D:prop>
<D:status>HTTP/1.1 200 OK</D:status>
</D:propstat>
</D:response>
</D:multistatus>
'''
SOME_EVENTS = [
{'DTSTART': '2020-02-20T20:02', 'DTEND': '2020-02-22T22:43', 'SUMMARY': 'Test'},
{
'DTSTART': '2020-02-20',
'DTEND': '2020-02-22',
'SUMMARY': 'Test',
'RRULE': {'FREQ': 'MONTHLY', 'BYDAY': ['FR', 'MO']},
},
{
'DTSTART': '2020-02-20',
'DTEND': '2020-02-22',
'SUMMARY': 'Test',
'LOCATION': 'Foo bar',
'TRANSP': True,
'DESCRIPTION': 'test',
},
]
@pytest.fixture()
def caldav_conn(db):
user = ApiUser.objects.create(username='all', keytype='', key='')
cdav = CalDAV.objects.create(dav_url=DAV_URL, slug='test', dav_login='foo', dav_password='hackmeplz')
content_type = ContentType.objects.get_for_model(cdav)
AccessRight.objects.create(
codename='can_access', apiuser=user, resource_type=content_type, resource_pk=cdav.pk
)
return cdav
def get_fake_event(caldav_conn):
vevent = icalendar.Event()
for k, v in {
'DTSTART': datetime.date(2023, 1, 1),
'DTEND': datetime.date(2023, 2, 1),
'SUMMARY': 'Test',
'UID': '12345678-abcd',
}.items():
vevent.add(k, v)
ical = icalendar.Calendar()
ical.add_component(vevent)
event = caldav.Event(client=caldav_conn.dav_client, data=ical)
event.url = caldav.lib.url.URL(DAV_URL + 'foo/calendar/12345678-abcd.ics')
calendar = caldav.Calendar(client=caldav_conn.dav_client)
calendar.url = caldav.lib.url.URL(DAV_URL + 'foo/calendar')
event.parent = calendar
return event
yweber marked this conversation as resolved Outdated

Pour moi cette partie induit trop d'indirections, l'objectif pour un test est d'être lisible au premier coup d'œil, là mon œil a besoin de beaucoup de coups :)

Pour simplifier je propose de remplacer la fixture par une simple méthode, get_mocked_calendar(url=..., id=..., ...).

Ça me semble convenir pour remplacer

    get_mock, conf = mocks
    conf['Event']['url'] = DAV_URL + cal_path
    conf['Event']['id'] = 42
    cal_mock = get_mock(conf)

par

calendar = get_mocked_calendar(url=DAV_URL + cal_path, id=42)
Pour moi cette partie induit trop d'indirections, l'objectif pour un test est d'être lisible au premier coup d'œil, là mon œil a besoin de beaucoup de coups :) Pour simplifier je propose de remplacer la fixture par une simple méthode, get_mocked_calendar(url=..., id=..., ...). Ça me semble convenir pour remplacer ``` get_mock, conf = mocks conf['Event']['url'] = DAV_URL + cal_path conf['Event']['id'] = 42 cal_mock = get_mock(conf) ``` par ``` calendar = get_mocked_calendar(url=DAV_URL + cal_path, id=42) ```

Bien vu ! Merci

Bien vu ! Merci
def get_calendar_mock(event_path, event_uid):
'''Return a mock configured to be usable as a Calendar instance in
the caldav connector
'''
event_url = DAV_URL + event_path
url = caldav.lib.url.URL(event_url)
evt_conf = {'url': url, 'id': event_uid}
mock_event = Mock(**evt_conf)
ical = icalendar.Calendar()
event = icalendar.Event()
conf = {
'url': DAV_URL,
'id': '1337',
'props': {
'dtstart': datetime.date.fromisoformat('2023-02-02'),
'dtend': datetime.date.fromisoformat('2023-02-28'),
'summary': 'test',
},
}
for k, v in conf.items():
event.add(k, v)
ical.add_component(event)
get_evt_conf = {
'url': url,
'id': event_uid,
'save': Mock(),
'delete': Mock(),
'icalendar_instance': ical,
}
mock_get_event = Mock(**get_evt_conf)
cal_conf = {'save_event.return_value': mock_event, 'event_by_url.return_value': mock_get_event}
return Mock(**cal_conf)
def qs_url(url, params):
'''Append query string to URL'''
return url + '?' + urllib.parse.urlencode(params)
yweber marked this conversation as resolved Outdated

Il est possible d'indiquer le nom de l'erreur plutôt que son numéro : disable=assignment-from-none

Mais ici pylint a plutôt raison, vérifier responses.calls comme tu le fais déjà semble être la manière recommandée, pourquoi cette double vérification ?

Il est possible d'indiquer le nom de l'erreur plutôt que son numéro : disable=assignment-from-none Mais ici pylint a plutôt raison, vérifier responses.calls comme tu le fais déjà semble être la manière recommandée, pourquoi cette double vérification ?

Mais ici pylint a plutôt raison

Ben bof, visiblement responses.add ne renvoie pas None ;)

vérifier responses.calls comme tu le fais déjà semble être la manière recommandée, pourquoi cette double vérification ?

Pour vérifier qu'il n'y a pas eu d'autres appels que celui prévu (et que l'appel unique qui a été fait était bien l'appel prévu). C'est pas la bonne manière de faire ?

> Mais ici pylint a plutôt raison Ben bof, visiblement `responses.add` ne renvoie pas None ;) > vérifier responses.calls comme tu le fais déjà semble être la manière recommandée, pourquoi cette double vérification ? Pour vérifier qu'il n'y a pas eu d'autres appels que celui prévu (et que l'appel unique qui a été fait était bien l'appel prévu). C'est pas la bonne manière de faire ?

C'est pas la bonne manière de faire ?

À partir du moment où ce n'est pas documenté, alors que pour le coup la doc de responses est plutôt touffue, on peut dire que non (et par extension que ça risque de casser sans prévenir au gré des mises à jour de la lib)

Pour vérifier qu'il n'y a pas eu d'autres appels que celui prévu (et que l'appel unique qui a été fait était bien l'appel prévu)

À mon avis le test échoue si responses voit que la réponse que tu as ajouté n'a pas été utilisée, donc juste vérifier len(responses.calls) garantit bien ces deux points

> C'est pas la bonne manière de faire ? À partir du moment où ce n'est pas documenté, alors que pour le coup la doc de responses est plutôt touffue, on peut dire que non (et par extension que ça risque de casser sans prévenir au gré des mises à jour de la lib) > Pour vérifier qu'il n'y a pas eu d'autres appels que celui prévu (et que l'appel unique qui a été fait était bien l'appel prévu) À mon avis le test échoue si responses voit que la réponse que tu as ajouté n'a pas été utilisée, donc juste vérifier `len(responses.calls)` garantit bien ces deux points

À partir du moment où ce n'est pas documenté, alors que pour le coup la doc de responses est plutôt touffue, on peut dire que non (et par extension que ça risque de casser sans prévenir au gré des mises à jour de la lib)

C'est dans la doc ;) https://github.com/getsentry/responses/blob/master/README.rst#assert-based-on-response-object

À mon avis le test échoue si responses voit que la réponse que tu as ajouté n'a pas été utilisée, donc juste vérifier len(responses.calls) garantit bien ces deux points

Alors oui mais non :P Le test va fail vu que caldav va se prendre une ConnectionRefused de requests, mais si des responses sont définit et ne sont jamais appelées ça ne fait pas fail le test.

> À partir du moment où ce n'est pas documenté, alors que pour le coup la doc de responses est plutôt touffue, on peut dire que non (et par extension que ça risque de casser sans prévenir au gré des mises à jour de la lib) C'est dans la doc ;) https://github.com/getsentry/responses/blob/master/README.rst#assert-based-on-response-object > À mon avis le test échoue si responses voit que la réponse que tu as ajouté n'a pas été utilisée, donc juste vérifier `len(responses.calls)` garantit bien ces deux points Alors oui mais non :P Le test va fail vu que caldav va se prendre une ConnectionRefused de requests, mais si des responses sont définit et ne sont jamais appelées ça ne fait pas fail le test.
def app_post(app, url, params, data, status=None):
yweber marked this conversation as resolved Outdated

Il me semble que plutôt que expect_errors on peut direct dire le statut d'erreur attendu, genre status=400 fait la même chose que expect_errors=True en plus précis

Aussi utiliser post_json et passer params=data évite d'encoder data explicitement avec json.dumps

Il me semble que plutôt que expect_errors on peut direct dire le statut d'erreur attendu, genre `status=400` fait la même chose que `expect_errors=True` en plus précis Aussi utiliser post_json et passer `params=data` évite d'encoder data explicitement avec json.dumps

Bien vu, merci ! Du coup ça a levé une série de tests qui n'était pas ouf. C'est corrigé.

Bien vu, merci ! Du coup ça a levé une série de tests qui n'était pas ouf. C'est corrigé.
'''Allows to post data on an URL with query string params and JSON data'''
yweber marked this conversation as resolved Outdated

On peut se passer de cette variable intermédiaire

On peut se passer de cette variable intermédiaire
return app.post_json(qs_url(url, params), params=data, status=status)
def get_event_path(caldav_conn, username, uid):
return EVENT_PATH_FMT % {
'username': urllib.parse.quote(username),
'uid': urllib.parse.quote(str(uid)),
}
#
# Tests
#
@responses.activate
def test_caldav_check_status_ok(app, caldav_conn):
responses.add('PROPFIND', DAV_URL, headers={'Content-Type': 'text/xml'}, body=PROPFIND_REPLY)
caldav_conn.check_status()
assert len(responses.calls) == 1
assert responses.calls[0].request.method == 'PROPFIND'
assert responses.calls[0].request.url == DAV_URL
@responses.activate
def test_caldav_check_status_login_failed(app, caldav_conn):
responses.add(
'PROPFIND',
DAV_URL,
headers={'Content-Type': 'text/xml'},
body=PROPFIND_REPLY,
status=401,
)
with pytest.raises(Exception) as expt:
caldav_conn.check_status()
assert str(expt) == 'Not authorized: bad login/password ?'
assert len(responses.calls) == 1
assert responses.calls[0].request.method == 'PROPFIND'
assert responses.calls[0].request.url == DAV_URL
@responses.activate
def test_caldav_check_status_fails(app, caldav_conn):
responses.add('PROPFIND', DAV_URL, status=500)
with pytest.raises(Exception):
caldav_conn.check_status()
assert len(responses.calls) == 1
assert responses.calls[0].request.method == 'PROPFIND'
assert responses.calls[0].request.url == DAV_URL
vdeniaud marked this conversation as resolved Outdated

Ça serait vraiment plus lisible d'avoir

        resp = app.post(url + '?cal_path=foo/calendar', json.dumps(event)) 

et je pense qu'on peut même

        resp = app.post_json(url + '?cal_path=foo/calendar', params=event)
Ça serait vraiment plus lisible d'avoir ``` resp = app.post(url + '?cal_path=foo/calendar', json.dumps(event)) ``` et je pense qu'on peut même ``` resp = app.post_json(url + '?cal_path=foo/calendar', params=event) ```

Je suis passé par une petite fonction utilitaire, je trouve effectivement ça plus lisible, j'espère que toi aussi.

Je suis passé par une petite fonction utilitaire, je trouve effectivement ça plus lisible, j'espère que toi aussi.
@pytest.mark.parametrize('requests_expt', (ConnectTimeout, ReadTimeout, ConnectionError))
@responses.activate
def test_caldav_check_status_timeout(app, caldav_conn, requests_expt):
propfind_mock = responses.add( # pylint: disable=assignment-from-none
'PROPFIND', DAV_URL, body=requests_expt('Do not work as expected')
)
get_mock = responses.add( # pylint: disable=assignment-from-none
'GET', DAV_URL, body=requests_expt('Do not work as expected')
)
with pytest.raises(Exception):
caldav_conn.check_status()
assert len(responses.calls) == 2
assert len(propfind_mock.calls) == 1
assert len(get_mock.calls) == 1
@responses.activate
def test_caldav_event_create_allday_ok(app, caldav_conn):
url = tests.utils.generic_endpoint_url('caldav', 'event/create')
username = 'foo'
evt_id = 42
evt_path = get_event_path(caldav_conn, username, evt_id)
cal_mock = get_calendar_mock(evt_path, evt_id)
event = {'DTSTART': '2020-02-20', 'DTEND': '2020-03-30', 'SUMMARY': 'foobar'}
qs_params = {'username': username}
with patch('caldav.Calendar', return_value=cal_mock):
resp = app_post(app, url, qs_params, event)
assert resp.json['err'] == 0
assert resp.json['data']['event_id'] == evt_id
event_conv = {
k: v if k not in ('DTSTART', 'DTEND') else datetime.date.fromisoformat(v)
for k, v in event.items()
}
cal_mock.save_event.assert_called_once_with(**event_conv)
@pytest.mark.parametrize(
'event',
SOME_EVENTS,
)
@responses.activate
def test_caldav_event_create_ok(app, caldav_conn, event):
url = tests.utils.generic_endpoint_url('caldav', 'event/create')
username = 'foo'
evt_id = '12345678-abcd'
evt_path = get_event_path(caldav_conn, username, evt_id)
cal_mock = get_calendar_mock(evt_path, evt_id)
qs_params = {'username': username}
with patch('caldav.Calendar', return_value=cal_mock):
resp = app_post(app, url, qs_params, event)
assert resp.json['err'] == 0
assert resp.json['data']['event_id'] == evt_id
save_args = event.copy()
caldav_conn._process_event_properties(save_args)
cal_mock.save_event.assert_called_once_with(**save_args)
@pytest.mark.parametrize(
'event',
SOME_EVENTS,
)
@responses.activate
def test_caldav_event_create_ok_nomock(app, caldav_conn, event):
'''Testing that caldav is able to create the event instance before sending it'''
url = tests.utils.generic_endpoint_url('caldav', 'event/create')
username = 'toto'
qs_params = {'username': username}
resp = app_post(app, url, qs_params, event)
assert resp.json['err'] != 0
assert resp.json['err_desc'] == 'Error sending creation request to caldav server'
@pytest.mark.parametrize(
'event',
SOME_EVENTS,
)
@responses.activate
def test_caldav_event_create_ok_net_err(app, caldav_conn, event):
with patch('passerelle.apps.caldav.models.CalDAV.get_event', return_value=get_fake_event(caldav_conn)):
url = tests.utils.generic_endpoint_url('caldav', 'event/create')
username = 'toto'
qs_params = {'username': username}
resp = app_post(app, url, qs_params, event)
assert resp.json['err'] != 0
assert resp.json['err_desc'] == 'Error sending creation request to caldav server'
@pytest.mark.parametrize(
'dtstart,dtend',
[
('2020-13-13', '2021-01-01'),
('2020-01-01T01:02:03', '2020-01-01T02:64:02'),
('23-01-12', '2023-01-13'),
('13:30', '2023-01-01T14:00'),
],
)
def test_caldav_event_create_bad_dates(app, caldav_conn, dtstart, dtend):
url = tests.utils.generic_endpoint_url('caldav', 'event/create')
username = 'toto'
event = {'DTSTART': dtstart, 'DTEND': dtend, 'SUMMARY': 'hello'}
qs_params = {'username': username}
resp = app_post(app, url, qs_params, event, status=400)
assert resp.status_code == 400
assert resp.json['err'] != 0
@pytest.mark.parametrize('transp', [True, False])
@responses.activate
def test_caldav_event_update_ok(app, transp, caldav_conn):
url = tests.utils.generic_endpoint_url('caldav', 'event/update')
username = 'toto'
evt_id = '01234567-abcd'
evt_path = get_event_path(caldav_conn, username, evt_id)
cal_mock = get_calendar_mock(evt_path, evt_id)
event = {'DTSTART': '2020-02-20', 'DTEND': '2020-03-30', 'SUMMARY': 'foobar', 'TRANSP': transp}
qs_params = {'username': username, 'event_id': evt_id}
with patch('caldav.Calendar', return_value=cal_mock):
resp = app_post(app, url, qs_params, event)
assert resp.json['err'] == 0
assert resp.json['data']['event_id'] == evt_id
cal_mock.event_by_url.assert_called_once_with(evt_path)
cal_mock.event_by_url().save.assert_called_once_with()
ical = cal_mock.event_by_url().icalendar_instance
evt = ical.walk('VEVENT')[0]
for k, v in event.items():
if k == 'TRANSP':
if v:
assert str(evt[k]) == 'TRANSPARENT'
else:
assert str(evt[k]) == 'OPAQUE'
else:
assert str(evt[k]) == v
@responses.activate
def test_caldav_event_update_notfound(app, caldav_conn):
url = tests.utils.generic_endpoint_url('caldav', 'event/update')
username = 'foo-user'
evt_id = '42'
evt_path = get_event_path(caldav_conn, username, evt_id)
cal_mock = get_calendar_mock(evt_path, evt_id)
cal_mock.event_by_url().icalendar_instance = icalendar.Calendar() # empty evt
event = {'DTSTART': '2020-02-20', 'DTEND': '2020-03-30', 'SUMMARY': 'foobar'}
qs_params = {'username': username, 'event_id': evt_id}
with patch('caldav.Calendar', return_value=cal_mock):
resp = app_post(app, url, qs_params, event)
assert resp.json['err'] == 1
assert resp.json['err_desc'] == 'Given event (user:%r uid:%r) do not contains VEVENT component' % (
username,
evt_id,
)
@responses.activate
def test_caldav_event_update_bad_evt(app, caldav_conn):
url = tests.utils.generic_endpoint_url('caldav', 'event/update')
username = 'toto'
evt_id = '012345678'
evt_path = get_event_path(caldav_conn, username, evt_id)
cal_mock = get_calendar_mock(evt_path, evt_id)
cal_mock.event_by_url.side_effect = caldav.lib.error.NotFoundError('NotFoundError at \'404 Not Found')
event = {'DTSTART': '2020-02-20', 'DTEND': '2020-03-30', 'SUMMARY': 'foobar'}
qs_params = {'username': username, 'event_id': evt_id}
with patch('caldav.Calendar', return_value=cal_mock):
resp = app_post(app, url, qs_params, event)
assert resp.json['err'] == 1
cal_mock.event_by_url.assert_called_once_with(evt_path)
assert cal_mock.event_by_url.save.call_count == 0
@responses.activate
def test_caldav_event_update_net_err(app, caldav_conn):
with patch('passerelle.apps.caldav.models.CalDAV.get_event', return_value=get_fake_event(caldav_conn)):
url = tests.utils.generic_endpoint_url('caldav', 'event/update')
username = 'toto'
evt_id = '012345678'
event = {'DTSTART': '2020-02-20', 'DTEND': '2020-03-30', 'SUMMARY': 'foobar'}
qs_params = {'username': username, 'event_id': evt_id}
resp = app_post(app, url, qs_params, event)
assert resp.json['err'] == 1
assert resp.json['err_desc'] == 'Error sending update request to caldav server'
@responses.activate
def test_caldav_event_delete_ok(app, caldav_conn):
url = tests.utils.generic_endpoint_url('caldav', 'event/delete')
username = 'toto'
evt_id = '012345678-dcba'
evt_path = get_event_path(caldav_conn, username, evt_id)
cal_mock = get_calendar_mock(evt_path, evt_id)
qs_params = {'username': username, 'event_id': evt_id}
with patch('caldav.Calendar', return_value=cal_mock):
resp = app.delete(qs_url(url, qs_params))
assert resp.json['err'] == 0
cal_mock.event_by_url.assert_called_once_with(evt_path)
cal_mock.event_by_url().delete.assert_called_once_with()
@responses.activate
def test_caldav_event_delete_net_err(app, caldav_conn):
with patch('passerelle.apps.caldav.models.CalDAV.get_event', return_value=get_fake_event(caldav_conn)):
url = tests.utils.generic_endpoint_url('caldav', 'event/delete')
username = 'toto'
evt_id = '012345678-dcba'
qs_params = {'username': username, 'event_id': evt_id}
resp = app.delete(qs_url(url, qs_params))
assert resp.json['err'] != 0
assert resp.json['err_desc'] == 'Error sending deletion request to caldav server'
@responses.activate
def test_caldav_event_delete_get_event_fail(app, caldav_conn):
url = tests.utils.generic_endpoint_url('caldav', 'event/delete')
username = 'toto'
evt_id = '012345678-dcba'
qs_params = {'username': username, 'event_id': evt_id}
resp = app.delete(qs_url(url, qs_params))
assert resp.json['err'] != 0
assert resp.json['err_desc'] == 'Unable to communicate with caldav server while fetching event'