passerelle/tests/test_caldav.py

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


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)


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


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