passerelle/tests/test_caldav.py

import datetime
import re
import urllib

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'

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>
'''

FAKE_ICS = '''BEGIN:VCALENDAR
VERSION:2.0
PRODID:-//EGroupware//NONSGML EGroupware Calendar 17.1.003//FR
BEGIN:VTIMEZONE
TZID:Europe/Paris
BEGIN:DAYLIGHT
TZOFFSETFROM:+0100
TZOFFSETTO:+0200
TZNAME:CEST
DTSTART:19700329T020000
RRULE:FREQ=YEARLY;BYMONTH=3;BYDAY=-1SU
END:DAYLIGHT
BEGIN:STANDARD
TZOFFSETFROM:+0200
TZOFFSETTO:+0100
TZNAME:CET
DTSTART:19701025T030000
RRULE:FREQ=YEARLY;BYMONTH=10;BYDAY=-1SU
END:STANDARD
END:VTIMEZONE
BEGIN:VEVENT
DTSTART;VALUE=DATE:20240227
DTEND;VALUE=DATE:20240228
X-MICROSOFT-CDO-ALLDAYEVENT:TRUE
SUMMARY:Yann Weber off
RRULE:FREQ=WEEKLY;BYDAY=TU
TRANSP:TRANSPARENT
CATEGORIES:Journées off
UID:01234567-abcd
STATUS:CONFIRMED
CREATED:20240320T095450Z
LAST-MODIFIED:20240320T095450Z
DTSTAMP:20240320T095513Z
END:VEVENT
END:VCALENDAR
'''

# Sometime EGW will add this line somewhere in its replies
SQL_LOG = '==> SQL => SELECT d.adnum as num, d.adsrc as def from pg_attrdef d, pg_class c where d.adrelid=c.oid and c.relname=\'egw_cal\' order by d.adnum<br>'

SOME_EVENTS = [
    {'DTSTART': '2020-02-20T20:02', 'DTEND': '2020-02-22T22:43', 'SUMMARY': 'Test'},
    {
        'DTSTART': '2020-02-20',
        'DTEND': '2020-02-22',
        'SUMMARY': 'Test',
        'TRANSP': False,
        'CATEGORY': 'Foobar',
        'RRULE': {'FREQ': 'MONTHLY', 'BYDAY': ['FR', 'MO']},
    },
    {
        'DTSTART': '2020-02-20',
        'DTEND': '2020-02-22',
        'SUMMARY': 'Test',
        'LOCATION': 'Foo bar',
        'TRANSP': True,
        'DESCRIPTION': 'test',
    },
    {
        'DTSTART': '2020-02-20T07:00',
        'DTEND': '2020-02-20T19:00',
        'SUMMARY': 'Test',
        'TRANSP': True,
        'CATEGORY': 'Test',
    },
]

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):
    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 qs_url(url, params):
    '''Append query string to URL'''
    return url + '?' + urllib.parse.urlencode(params)


def app_post(app, url, params, data, status=None):
    '''Allows to post data on an URL with query string params and JSON data'''
    return app.post_json(qs_url(url, params), params=data, status=status)


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)


def add_egw_sql_log(content, lineno=0):
    '''Adds the EGW's random SQL log to some content
    - lineno : indicate where to insert the line (-1 for last line)
    '''
    spl = content.split('\n')
    if lineno == -1:
        spl.append(SQL_LOG)
    else:
        if lineno < 0:
            lineno += 1
        spl.insert(lineno, SQL_LOG)
    return '\n'.join(spl)


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


@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


@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'
    qs_params = {'username': username}
    responses.add(responses.PUT, event_url_regex(username), status=204, body='')

    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]

    expt_event = event.copy()
    expt_event['SEQUENCE'] = 0
    assert_match_vevent(vevent, event)


@pytest.mark.parametrize(
    'expt', [ConnectionError('refused'), ConnectTimeout('timeout'), ReadTimeout('timeout')]
)
@responses.activate
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(
    'dtstart,dtend,until',
    [
        ('2020-13-13', '2021-01-01', '2021-12-01'),
        ('2020-01-01T01:02:03', '2020-01-01T02:64:02', '2021-12-01'),
        ('23-01-12', '2023-01-13', '2021-12-01'),
        ('13:30', '2023-01-01T14:00', '2021-12-01'),
        ('2023-10-02', '2023-10-03', '2024-13-12'),
    ],
)
def test_caldav_event_create_bad_dates(app, caldav_conn, dtstart, dtend, until):
    url = tests.utils.generic_endpoint_url('caldav', 'event/create')
    username = 'toto'
    event = {
        'DTSTART': dtstart,
        'DTEND': dtend,
        'SUMMARY': 'hello',
        'RRULE': {'FREQ': 'WEEKLY', 'BYDAY': ['FR'], 'UNTIL': until},
    }
    qs_params = {'username': username}

    resp = app_post(app, url, qs_params, event, status=400)

    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')


@responses.activate
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_url = DAV_URL + get_event_path(username, evt_id)

    qs_params = {'username': username, 'event_id': evt_id}

    responses.add(responses.GET, evt_url, body=FAKE_ICS)
    responses.add(responses.PUT, evt_url, status=204, body='')

    resp = app_post(app, url, qs_params, event_update)
    assert resp.json['err'] == 0
    assert resp.json['data']['event_id'] == evt_id

    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(username, evt_id)

    qs_params = {'username': username, 'event_id': evt_id}

    responses.add(responses.GET, evt_url, body=FAKE_ICS)
    responses.add(responses.PUT, evt_url, status=204, body='')

    # testing SEQUENCE initialization : if missing we add it
    dtend = datetime.date.fromisoformat('2024-02-29')
    resp = app_post(app, url, qs_params, {'DTEND': dtend.isoformat()})
    assert resp.json['err'] == 0
    assert len(responses.calls) == 2
    put_mock = responses.calls[1]
    assert put_mock.request.method == 'PUT'
    raw_ics = put_mock.request.body
    calendar = icalendar.Calendar.from_ical(raw_ics)
    vevent = calendar.walk('VEVENT')[0]
    assert vevent['SEQUENCE'] == 1

    for expt_sequence in range(2, 10):
        # testing SEQUENCE incrementation on modification
        responses.reset()
        responses.add(responses.GET, evt_url, body=raw_ics)
        responses.add(responses.PUT, evt_url, status=204, body='')

        dtend += datetime.timedelta(days=1)
        resp = app_post(app, url, qs_params, {'DTEND': dtend.isoformat()})
        assert resp.json['err'] == 0
        assert len(responses.calls) == 2
        assert responses.calls[1].request.method == 'PUT'
        raw_ics = responses.calls[1].request.body
        calendar = icalendar.Calendar.from_ical(raw_ics)
        vevent = calendar.walk('VEVENT')[0]
        assert vevent['SEQUENCE'] == expt_sequence


@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(username, evt_id)

    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}

    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)

    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}

    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, 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(username, evt_id)
    qs_params = {'username': username, 'event_id': evt_id}

    responses.add(responses.GET, DAV_URL + evt_path, body=FAKE_ICS)
    responses.add(responses.DELETE, DAV_URL + evt_path, status=204, body='')

    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, 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)

    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'

    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_class'] == APIERROR_CLS
    assert resp.json['err_desc'] == 'Unable to get event %r in calendar owned by %r' % (evt_id, username)


@pytest.mark.parametrize('lineno', (0, 10, -1))
@responses.activate
def test_egw_sql_log_propfind(app, caldav_conn, lineno):
    response_body = add_egw_sql_log(PROPFIND_REPLY, lineno)

    responses.add('PROPFIND', DAV_URL, body=response_body, content_type='text/xml')

    caldav_conn.check_status()
    assert len(responses.calls) == 1
    assert responses.calls[0].request.method == 'PROPFIND'
    assert responses.calls[0].request.url == DAV_URL


@pytest.mark.parametrize('lineno', (0, 10, -1))
@responses.activate
def test_egw_sql_log_update(app, caldav_conn, lineno):
    event_update = SOME_UPDATES[0]

    response_body = add_egw_sql_log(FAKE_ICS, lineno)

    url = tests.utils.generic_endpoint_url('caldav', 'event/update')
    username = 'toto'
    evt_id = '01234567-abcd'
    evt_url = DAV_URL + get_event_path(username, evt_id)

    qs_params = {'username': username, 'event_id': evt_id}

    responses.add(responses.GET, evt_url, body=response_body)
    responses.add(responses.PUT, evt_url, status=204, body='')

    resp = app_post(app, url, qs_params, event_update)
    assert resp.json['err'] == 0
    assert resp.json['data']['event_id'] == evt_id

    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)