caldav: remove caldav lib mocks from tests (#88463)
gitea/passerelle/pipeline/head This commit looks good Details

This commit is contained in:
Yann Weber 2024-03-21 16:51:03 +01:00
parent 96c1e49e23
commit 1fa2c0f9a7
2 changed files with 166 additions and 265 deletions

View File

@ -174,6 +174,11 @@ class CalDAV(BaseResource):
'username': username,
},
)
except caldav.lib.error.DAVError as expt:
raise APIError(
_('Error creating event'),
data={'expt_class': str(type(expt)), 'expt': exception_to_text(expt), 'username': username},
)
return {'data': {'event_id': evt.id}}
# Patch do not support request_body validation, using post instead

View File

@ -1,18 +1,19 @@
import datetime
import re
import urllib
from unittest.mock import Mock, patch
import caldav
import icalendar
import pytest
import responses
from django.contrib.contenttypes.models import ContentType
from django.utils.dateparse import parse_date, parse_datetime
from requests.exceptions import ConnectionError, ConnectTimeout, ReadTimeout
import tests.utils
from passerelle.apps.caldav.models import CalDAV
from passerelle.base.models import AccessRight, ApiUser
APIERROR_CLS = 'passerelle.utils.jsonresponse.APIError'
DAV_URL = 'http://test.caldav.notld/somedav/'
EVENT_PATH_FMT = '%(username)s/calendar/%(uid)s.ics'
@ -84,6 +85,8 @@ SOME_EVENTS = [
'DTSTART': '2020-02-20',
'DTEND': '2020-02-22',
'SUMMARY': 'Test',
'TRANSP': False,
'CATEGORY': 'Foobar',
'RRULE': {'FREQ': 'MONTHLY', 'BYDAY': ['FR', 'MO']},
},
{
@ -96,6 +99,15 @@ SOME_EVENTS = [
},
]
SOME_UPDATES = [
*SOME_EVENTS,
{'DTSTART': '2020-01-01'},
{'RRULE': {'FREQ': 'WEEKLY', 'BYDAY': ['FR'], 'UNTIL': '2025-01-01'}},
{'SUMMARY': 'Foobar', 'CATEGORY': 'Toto'},
{'TRANSP': False},
{'CATEGORY': 'SomeCategory'},
]
@pytest.fixture()
def caldav_conn(db):
@ -108,64 +120,6 @@ def caldav_conn(db):
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
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)
@ -176,11 +130,38 @@ def app_post(app, url, params, data, status=None):
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)),
}
def get_event_path(username, uid=None):
ret = '%s/calendar/' % username
if uid:
ret += '%s.ics' % uid
return ret
def event_url_regex(username):
return re.compile(DAV_URL + username + '/calendar/([^/]+)$')
def assert_match_vevent(vevent, expt_event):
for k, v in expt_event.items():
if k in ('DTSTART', 'DTEND'):
assert (parse_date(v) or parse_datetime(v)) == vevent.decoded(k)
elif k == 'TRANSP':
assert (b'TRANSPARENT' if v else b'OPAQUE') == vevent.decoded(k)
elif k == 'RRULE':
rrule = vevent.decoded(k)
for rk, rv in v.items():
if rk == 'UNTIL':
assert [(parse_date(rv) or parse_datetime(rv))] == rrule[rk]
elif isinstance(rrule[rk], list) and not isinstance(rv, list):
assert rv == rrule[rk][0]
else:
assert set(rv) == set(rrule[rk])
elif k == 'CATEGORY':
assert v.encode() == vevent['CATEGORIES'].to_ical()
elif isinstance(v, str):
assert v.encode() == vevent.decoded(k)
else:
assert v == vevent.decoded(k)
#
@ -240,55 +221,6 @@ def test_caldav_check_status_timeout(app, caldav_conn, requests_expt):
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()
}
event_conv['SEQUENCE'] = 0
cal_mock.save_event.assert_called_once_with(**event_conv)
@responses.activate
def test_caldav_event_categories(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', 'CATEGORY': '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()
}
event_conv['CATEGORIES'] = [event_conv.pop('CATEGORY')]
event_conv['SEQUENCE'] = 0
cal_mock.save_event.assert_called_once_with(**event_conv)
@pytest.mark.parametrize(
'event',
SOME_EVENTS,
@ -297,55 +229,37 @@ def test_caldav_event_categories(app, caldav_conn):
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}
responses.add(responses.PUT, event_url_regex(username), status=204, body='')
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
resp = app_post(app, url, qs_params, event)
assert resp.json['err'] == 0
assert len(responses.calls) == 1
calendar = icalendar.Calendar.from_ical(responses.calls[0].request.body)
vevent = calendar.walk('VEVENT')[0]
save_args = event.copy()
caldav_conn._process_event_properties(save_args)
save_args['SEQUENCE'] = 0
cal_mock.save_event.assert_called_once_with(**save_args)
expt_event = event.copy()
expt_event['SEQUENCE'] = 1 # Bug with caldav not able to save event with SEQUENCE:0
assert_match_vevent(vevent, event)
@pytest.mark.parametrize(
'event',
SOME_EVENTS,
'expt', [ConnectionError('refused'), ConnectTimeout('timeout'), ReadTimeout('timeout')]
)
@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'''
def test_caldav_event_create_ok_net_err(app, caldav_conn, expt):
event = SOME_EVENTS[0]
url = tests.utils.generic_endpoint_url('caldav', 'event/create')
username = 'toto'
qs_params = {'username': username}
responses.add(responses.PUT, event_url_regex(username), body=expt)
resp = app_post(app, url, qs_params, event)
assert resp.json['err'] != 0
assert resp.json['err_class'] == APIERROR_CLS
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,until',
[
@ -371,79 +285,32 @@ def test_caldav_event_create_bad_dates(app, caldav_conn, dtstart, dtend, until):
assert resp.status_code == 400
assert resp.json['err'] != 0
assert resp.json['err_class'] == APIERROR_CLS
assert resp.json['err_desc'].startswith('Unable to convert field')
@pytest.mark.parametrize('transp', [True, False])
@responses.activate
def test_caldav_event_update_ok(app, transp, caldav_conn):
"""This test mock de caldav lib in order to check if the
icalendar lib raises error on event modifications
"""
def test_caldav_event_create_bad_username(app, caldav_conn):
url = tests.utils.generic_endpoint_url('caldav', 'event/create')
username = 'foo'
qs_params = {'username': username}
responses.add(responses.PUT, event_url_regex(username), status=404)
event = SOME_EVENTS[0]
resp = app_post(app, url, qs_params, event)
assert resp.json['err'] != 0
assert resp.json['err_class'] == APIERROR_CLS
assert resp.json['err_desc'] == 'Error creating event'
assert len(responses.calls) == 2
@pytest.mark.parametrize('event_update', SOME_UPDATES)
@responses.activate
def test_caldav_event_update_ok(app, caldav_conn, event_update):
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,
'RRULE': {'FREQ': 'MONTHLY', 'BYDAY': ['FR'], 'UNTIL': '2020-10-01'},
}
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'
elif k == 'RRULE':
assert isinstance(evt[k], icalendar.vRecur)
for rk, rv in event['RRULE'].items():
if rk == 'UNTIL':
assert evt[k][rk] == datetime.date.fromisoformat(rv)
else:
assert evt[k][rk] == rv
elif k in ('DTSTART', 'DTEND'):
assert evt.decoded(k) == datetime.date.fromisoformat(v)
else:
assert str(evt[k]) == v
@pytest.mark.parametrize(
'event_update',
(
{'DTSTART': '2020-02-20', 'DTEND': '2020-03-30', 'SUMMARY': 'Foobar'},
{'TRANSP': True},
{'RRULE/FREQ': 'MONTHLY', 'RRULE/BYDAY': ['TU'], 'RRULE/UNTIL': '2020-10-01'},
{'DTSTART': '2020-02-20', 'TRANSP': True},
),
)
@responses.activate
def test_caldav_event_update_ok_nomock(app, caldav_conn, event_update):
"""This test let the caldav lib think it is able to retrieve
an event in order to modify it.
This can trigger errors from icalendar lib when caldav
attempt to save the event
"""
url = tests.utils.generic_endpoint_url('caldav', 'event/update')
username = 'toto'
evt_id = '01234567-abcd'
evt_url = DAV_URL + get_event_path(caldav_conn, username, evt_id)
evt_url = DAV_URL + get_event_path(username, evt_id)
qs_params = {'username': username, 'event_id': evt_id}
@ -457,13 +324,18 @@ def test_caldav_event_update_ok_nomock(app, caldav_conn, event_update):
assert len(responses.calls) == 2
assert responses.calls[1].request.method == 'PUT'
calendar = icalendar.Calendar.from_ical(responses.calls[1].request.body)
vevent = calendar.walk('VEVENT')[0]
assert_match_vevent(vevent, event_update)
@responses.activate
def test_caldav_event_update_sequence_ok(app, caldav_conn):
url = tests.utils.generic_endpoint_url('caldav', 'event/update')
username = 'foobar'
evt_id = '1234567890'
evt_url = DAV_URL + get_event_path(caldav_conn, username, evt_id)
evt_url = DAV_URL + get_event_path(username, evt_id)
qs_params = {'username': username, 'event_id': evt_id}
@ -504,98 +376,122 @@ 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)
evt_path = get_event_path(username, evt_id)
cal_mock.event_by_url().icalendar_instance = icalendar.Calendar() # empty evt
responses.add(responses.GET, DAV_URL + evt_path, status=404)
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,
)
resp = app_post(app, url, qs_params, event)
assert resp.json['err'] == 1
assert resp.json['err_desc'] == 'Unable to get event %r in calendar owned by %r' % (
evt_id,
username,
)
@responses.activate
def test_caldav_event_update_bad_evt(app, caldav_conn):
url = tests.utils.generic_endpoint_url('caldav', 'event/update')
username = 'foo-user'
evt_id = '42'
evt_path = get_event_path(username, evt_id)
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')
responses.add(
responses.GET, DAV_URL + evt_path, status=200, body=b'BEGIN:VCALENDAR\nVERSION:2.0\nEND:VCALENDAR'
)
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
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,
)
@pytest.mark.parametrize(
'expt', [ConnectionError('refused'), ConnectTimeout('timeout'), ReadTimeout('timeout')]
)
@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'
def test_caldav_event_update_net_err(app, caldav_conn, expt):
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}
responses.add(responses.GET, DAV_URL + get_event_path(username, evt_id), body=expt)
resp = app_post(app, url, qs_params, event)
assert resp.json['err'] == 1
assert resp.json['err_desc'] == 'Unable to communicate with caldav server while fetching event'
responses.reset()
responses.add(responses.GET, DAV_URL + get_event_path(username, evt_id), body=FAKE_ICS)
responses.add(responses.PUT, DAV_URL + get_event_path(username, evt_id), body=expt)
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)
evt_path = get_event_path(username, 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
responses.add(responses.GET, DAV_URL + evt_path, body=FAKE_ICS)
responses.add(responses.DELETE, DAV_URL + evt_path, status=204, body='')
cal_mock.event_by_url.assert_called_once_with(evt_path)
cal_mock.event_by_url().delete.assert_called_once_with()
resp = app.delete(qs_url(url, qs_params))
assert resp.json['err'] == 0
assert len(responses.calls) == 2
assert responses.calls[1].request.method == 'DELETE'
@pytest.mark.parametrize(
'expt', [ConnectionError('refused'), ConnectTimeout('timeout'), ReadTimeout('timeout')]
)
@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')
def test_caldav_event_delete_net_err(app, caldav_conn, expt):
url = tests.utils.generic_endpoint_url('caldav', 'event/delete')
username = 'toto'
evt_id = '012345678-dcba'
qs_params = {'username': username, 'event_id': evt_id}
evt_path = get_event_path(username, evt_id)
username = 'toto'
evt_id = '012345678-dcba'
responses.add(responses.GET, DAV_URL + evt_path, body=expt)
resp = app.delete(qs_url(url, qs_params))
assert resp.json['err'] != 0
assert resp.json['err_class'] == APIERROR_CLS
assert resp.json['err_desc'] == 'Unable to communicate with caldav server while fetching event'
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.add(responses.GET, DAV_URL + evt_path, body=FAKE_ICS)
responses.add(responses.DELETE, DAV_URL + evt_path, body=expt)
resp = app.delete(qs_url(url, qs_params))
assert resp.json['err'] != 0
assert resp.json['err_class'] == APIERROR_CLS
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}
evt_path = get_event_path(username, evt_id)
responses.add(responses.GET, DAV_URL + evt_path, body='', status=404)
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'
assert resp.json['err_class'] == APIERROR_CLS
assert resp.json['err_desc'] == 'Unable to get event %r in calendar owned by %r' % (evt_id, username)