odoo_source/addons/microsoft_calendar/tests/test_microsoft_service.py

import json
import requests
from unittest.mock import patch, call, MagicMock

from odoo import fields
from odoo.addons.microsoft_calendar.utils.microsoft_calendar import MicrosoftCalendarService
from odoo.addons.microsoft_calendar.utils.microsoft_event import MicrosoftEvent
from odoo.addons.microsoft_account.models.microsoft_service import MicrosoftService
from odoo.tests import TransactionCase


DEFAULT_TIMEOUT = 20


class TestMicrosoftService(TransactionCase):

    def _do_request_result(self, data):
        """ _do_request returns a tuple (status, data, time) but only the data part is used """
        return (None, data, None)

    def setUp(self):
        super(TestMicrosoftService, self).setUp()

        self.service = MicrosoftCalendarService(self.env["microsoft.service"])
        self.fake_token = "MY_TOKEN"
        self.fake_sync_token = "MY_SYNC_TOKEN"
        self.fake_next_sync_token = "MY_NEXT_SYNC_TOKEN"
        self.fake_next_sync_token_url = f"https://graph.microsoft.com/v1.0/me/calendarView/delta?$deltatoken={self.fake_next_sync_token}"

        self.header_prefer = 'outlook.body-content-type="html", odata.maxpagesize=50'
        self.header = {'Content-type': 'application/json', 'Authorization': 'Bearer %s' % self.fake_token}
        self.call_with_sync_token = call(
            "/v1.0/me/calendarView/delta",
            {"$deltatoken": self.fake_sync_token},
            {**self.header, 'Prefer': self.header_prefer},
            method="GET", timeout=DEFAULT_TIMEOUT,
        )
        self.call_without_sync_token = call(
            "/v1.0/me/calendarView/delta",
            {
                'startDateTime': fields.Datetime.subtract(fields.Datetime.now(), days=365).strftime("%Y-%m-%dT00:00:00Z"),
                'endDateTime': fields.Datetime.add(fields.Datetime.now(), days=365).strftime("%Y-%m-%dT00:00:00Z"),
            },
            {**self.header, 'Prefer': self.header_prefer},
            method="GET", timeout=DEFAULT_TIMEOUT,
        )

    def test_get_events_delta_without_token(self):
        """
        if no token is provided, an exception is raised
        """
        with self.assertRaises(AttributeError):
            self.service._get_events_delta()

    @patch.object(MicrosoftService, "_do_request")
    def test_get_events_unexpected_exception(self, mock_do_request):
        """
        When an unexpected exception is raised, just propagate it.
        """
        mock_do_request.side_effect = Exception()

        with self.assertRaises(Exception):
            self.service._get_events_delta(token=self.fake_token, timeout=DEFAULT_TIMEOUT)

    @patch.object(MicrosoftService, "_do_request")
    def test_get_events_delta_token_error(self, mock_do_request):
        """
        When the provided sync token is invalid, an exception should be raised and then
        a full sync should be done.
        """
        mock_do_request.side_effect = [
            requests.HTTPError(response=MagicMock(status_code=410, content="fullSyncRequired")),
            self._do_request_result({"value": []}),
        ]

        events, next_token = self.service._get_events_delta(
            token=self.fake_token, sync_token=self.fake_sync_token, timeout=DEFAULT_TIMEOUT
        )

        self.assertEqual(next_token, None)
        self.assertFalse(events)
        mock_do_request.assert_has_calls([self.call_with_sync_token, self.call_without_sync_token])

    @patch.object(MicrosoftService, "_do_request")
    def test_get_events_delta_without_sync_token(self, mock_do_request):
        """
        when no sync token is provided, a full sync should be done
        """
        # returns empty data without any next sync token
        mock_do_request.return_value = self._do_request_result({"value": []})

        events, next_token = self.service._get_events_delta(token=self.fake_token, timeout=DEFAULT_TIMEOUT)

        self.assertEqual(next_token, None)
        self.assertFalse(events)
        mock_do_request.assert_has_calls([self.call_without_sync_token])

    @patch.object(MicrosoftService, "_do_request")
    def test_get_events_delta_with_sync_token(self, mock_do_request):
        """
        when a sync token is provided, we should retrieve the sync token to use for the next sync.
        """
        # returns empty data with a next sync token
        mock_do_request.return_value = self._do_request_result({
            "value": [],
            "@odata.deltaLink": self.fake_next_sync_token_url
        })

        events, next_token = self.service._get_events_delta(
            token=self.fake_token, sync_token=self.fake_sync_token, timeout=DEFAULT_TIMEOUT
        )

        self.assertEqual(next_token, "MY_NEXT_SYNC_TOKEN")
        self.assertFalse(events)
        mock_do_request.assert_has_calls([self.call_with_sync_token])

    @patch.object(MicrosoftService, "_do_request")
    def test_get_events_one_page(self, mock_do_request):
        """
        When all events are on one page, just get them.
        """
        mock_do_request.return_value = self._do_request_result({
            "value": [
                {"id": 1, "type": "singleInstance", "subject": "ev1"},
                {"id": 2, "type": "singleInstance", "subject": "ev2"},
                {"id": 3, "type": "singleInstance", "subject": "ev3"},
            ],
        })
        events, _ = self.service._get_events_delta(token=self.fake_token, timeout=DEFAULT_TIMEOUT)

        self.assertEqual(events, MicrosoftEvent([
            {"id": 1, "type": "singleInstance", "subject": "ev1"},
            {"id": 2, "type": "singleInstance", "subject": "ev2"},
            {"id": 3, "type": "singleInstance", "subject": "ev3"},
        ]))
        mock_do_request.assert_has_calls([self.call_without_sync_token])

    @patch.object(MicrosoftService, "_do_request")
    def test_get_events_loop_over_pages(self, mock_do_request):
        """
        Loop over pages to retrieve all the events.
        """
        mock_do_request.side_effect = [
            self._do_request_result({
                "value": [{"id": 1, "type": "singleInstance", "subject": "ev1"}],
                "@odata.nextLink": "link_1"
            }),
            self._do_request_result({
                "value": [{"id": 2, "type": "singleInstance", "subject": "ev2"}],
                "@odata.nextLink": "link_2"
            }),
            self._do_request_result({
                "value": [{"id": 3, "type": "singleInstance", "subject": "ev3"}],
            }),
        ]

        events, _ = self.service._get_events_delta(token=self.fake_token, timeout=DEFAULT_TIMEOUT)

        self.assertEqual(events, MicrosoftEvent([
            {"id": 1, "type": "singleInstance", "subject": "ev1"},
            {"id": 2, "type": "singleInstance", "subject": "ev2"},
            {"id": 3, "type": "singleInstance", "subject": "ev3"},
        ]))
        mock_do_request.assert_has_calls([
            self.call_without_sync_token,
            call(
                "link_1",
                {},
                {**self.header, 'Prefer': self.header_prefer},
                preuri='', method="GET", timeout=DEFAULT_TIMEOUT
            ),
            call(
                "link_2",
                {},
                {**self.header, 'Prefer': self.header_prefer},
                preuri='', method="GET", timeout=DEFAULT_TIMEOUT
            ),
        ])

    @patch.object(MicrosoftService, "_do_request")
    def test_get_events_filter_out_occurrences(self, mock_do_request):
        """
        When all events are on one page, just get them.
        """
        mock_do_request.return_value = self._do_request_result({
            "value": [
                {"id": 1, "type": "singleInstance", "subject": "ev1"},
                {"id": 2, "type": "occurrence", "subject": "ev2"},
                {"id": 3, "type": "seriesMaster", "subject": "ev3"},
            ],
        })
        events, _ = self.service._get_events_delta(token=self.fake_token, timeout=DEFAULT_TIMEOUT)

        self.assertEqual(events, MicrosoftEvent([
            {"id": 1, "type": "singleInstance", "subject": "ev1"},
            {"id": 3, "type": "seriesMaster", "subject": "ev3"},
        ]))
        mock_do_request.assert_has_calls([self.call_without_sync_token])

    def test_get_occurrence_details_token_error(self):
        """
        if no token is provided, an exception is raised
        """
        with self.assertRaises(AttributeError):
            self.service._get_occurrence_details(1)

    @patch.object(MicrosoftService, "_do_request")
    def test_get_occurrence_details(self, mock_do_request):
        mock_do_request.return_value = self._do_request_result({
            "value": [
                {"id": 1, "type": "singleInstance", "subject": "ev1"},
                {"id": 2, "type": "occurrence", "subject": "ev2"},
                {"id": 3, "type": "seriesMaster", "subject": "ev3"},
            ],
        })
        events = self.service._get_occurrence_details(123, token=self.fake_token, timeout=DEFAULT_TIMEOUT)

        self.assertEqual(events, MicrosoftEvent([
            {"id": 1, "type": "singleInstance", "subject": "ev1"},
            {"id": 2, "type": "occurrence", "subject": "ev2"},
            {"id": 3, "type": "seriesMaster", "subject": "ev3"},
        ]))

        mock_do_request.assert_called_with(
            "/v1.0/me/events/123/instances",
            {
                'startDateTime': fields.Datetime.subtract(fields.Datetime.now(), days=365).strftime("%Y-%m-%dT00:00:00Z"),
                'endDateTime': fields.Datetime.add(fields.Datetime.now(), days=365).strftime("%Y-%m-%dT00:00:00Z"),
            },
            {**self.header, 'Prefer': self.header_prefer},
            method='GET', timeout=DEFAULT_TIMEOUT,
        )

    def test_get_events_token_error(self):
        """
        if no token is provided, an exception is raised
        """
        with self.assertRaises(AttributeError):
            self.service.get_events()

    @patch.object(MicrosoftService, "_do_request")
    def test_get_events_no_serie_master(self, mock_do_request):
        """
        When there is no serie master, just retrieve the list of events.
        """
        mock_do_request.return_value = self._do_request_result({
            "value": [
                {"id": 1, "type": "singleInstance", "subject": "ev1"},
                {"id": 2, "type": "singleInstance", "subject": "ev2"},
                {"id": 3, "type": "singleInstance", "subject": "ev3"},
            ],
        })

        events, _ = self.service.get_events(token=self.fake_token, timeout=DEFAULT_TIMEOUT)

        self.assertEqual(events, MicrosoftEvent([
            {"id": 1, "type": "singleInstance", "subject": "ev1"},
            {"id": 2, "type": "singleInstance", "subject": "ev2"},
            {"id": 3, "type": "singleInstance", "subject": "ev3"},
        ]))

    @patch.object(MicrosoftService, "_do_request")
    def test_get_events_with_one_serie_master(self, mock_do_request):
        """
        When there is a serie master, retrieve the list of events and event occurrences linked to the serie master
        """
        mock_do_request.side_effect = [
            self._do_request_result({
                "value": [
                    {"id": 1, "type": "singleInstance", "subject": "ev1"},
                    {"id": 2, "type": "seriesMaster", "subject": "ev2"},
                ],
            }),
            self._do_request_result({
                "value": [
                    {"id": 3, "type": "occurrence", "subject": "ev3"},
                ],
            }),
        ]

        events, _ = self.service.get_events(token=self.fake_token, timeout=DEFAULT_TIMEOUT)

        self.assertEqual(events, MicrosoftEvent([
            {"id": 1, "type": "singleInstance", "subject": "ev1"},
            {"id": 2, "type": "seriesMaster", "subject": "ev2"},
            {"id": 3, "type": "occurrence", "subject": "ev3"},
        ]))

    def test_insert_token_error(self):
        """
        if no token is provided, an exception is raised
        """
        with self.assertRaises(AttributeError):
            self.service.insert({})


    @patch.object(MicrosoftService, "_do_request")
    def test_insert(self, mock_do_request):

        mock_do_request.return_value = self._do_request_result({'id': 1, 'iCalUId': 2})

        instance_id, event_id = self.service.insert({"subject": "ev1"}, token=self.fake_token, timeout=DEFAULT_TIMEOUT)

        self.assertEqual(instance_id, 1)
        self.assertEqual(event_id, 2)
        mock_do_request.assert_called_with(
            "/v1.0/me/calendar/events",
            json.dumps({"subject": "ev1"}),
            self.header, method="POST", timeout=DEFAULT_TIMEOUT
        )

    def test_patch_token_error(self):
        """
        if no token is provided, an exception is raised
        """
        with self.assertRaises(AttributeError):
            self.service.patch(123, {})

    @patch.object(MicrosoftService, "_do_request")
    def test_patch_returns_false_if_event_does_not_exist(self, mock_do_request):
        event_id = 123
        values = {"subject": "ev2"}
        mock_do_request.return_value = (404, "", None)

        res = self.service.patch(event_id, values, token=self.fake_token, timeout=DEFAULT_TIMEOUT)

        self.assertFalse(res)
        mock_do_request.assert_called_with(
            f"/v1.0/me/calendar/events/{event_id}",
            json.dumps(values),
            self.header, method="PATCH", timeout=DEFAULT_TIMEOUT
        )

    @patch.object(MicrosoftService, "_do_request")
    def test_patch_an_existing_event(self, mock_do_request):
        event_id = 123
        values = {"subject": "ev2"}
        mock_do_request.return_value = (200, "", None)

        res = self.service.patch(event_id, values, token=self.fake_token, timeout=DEFAULT_TIMEOUT)

        self.assertTrue(res)
        mock_do_request.assert_called_with(
            f"/v1.0/me/calendar/events/{event_id}",
            json.dumps(values),
            self.header, method="PATCH", timeout=DEFAULT_TIMEOUT
        )

    def test_delete_token_error(self):
        """
        if no token is provided, an exception is raised
        """
        with self.assertRaises(AttributeError):
            self.service.delete(123)

    @patch.object(MicrosoftService, "_do_request")
    def test_delete_returns_false_if_event_does_not_exist(self, mock_do_request):
        event_id = 123
        mock_do_request.return_value = (404, "", None)

        res = self.service.delete(event_id, token=self.fake_token, timeout=DEFAULT_TIMEOUT)

        self.assertFalse(res)
        mock_do_request.assert_called_with(
            f"/v1.0/me/calendar/events/{event_id}",
            {}, headers={'Authorization': 'Bearer %s' % self.fake_token}, method="DELETE", timeout=DEFAULT_TIMEOUT
        )

    @patch.object(MicrosoftService, "_do_request")
    def test_delete_an_already_cancelled_event(self, mock_do_request):
        """
        When an event has already been cancelled, Outlook may return a status code equals to 403 or 410.
        In this case, the delete method should return True.
        """
        event_id = 123

        for status in (403, 410):
            mock_do_request.return_value = (status, "", None)

            res = self.service.delete(event_id, token=self.fake_token, timeout=DEFAULT_TIMEOUT)

            self.assertTrue(res)
            mock_do_request.assert_called_with(
                f"/v1.0/me/calendar/events/{event_id}",
                {}, headers={'Authorization': 'Bearer %s' % self.fake_token}, method="DELETE", timeout=DEFAULT_TIMEOUT
            )


    @patch.object(MicrosoftService, "_do_request")
    def test_delete_an_existing_event(self, mock_do_request):
        event_id = 123
        mock_do_request.return_value = (200, "", None)

        res = self.service.delete(event_id, token=self.fake_token, timeout=DEFAULT_TIMEOUT)

        self.assertTrue(res)
        mock_do_request.assert_called_with(
            f"/v1.0/me/calendar/events/{event_id}",
            {}, headers={'Authorization': 'Bearer %s' % self.fake_token}, method="DELETE", timeout=DEFAULT_TIMEOUT
        )

    def test_answer_token_error(self):
        """
        if no token is provided, an exception is raised
        """
        with self.assertRaises(AttributeError):
            self.service.answer(123, 'ok', {})

    @patch.object(MicrosoftService, "_do_request")
    def test_answer_returns_false_if_event_does_not_exist(self, mock_do_request):
        event_id = 123
        answer = "accept"
        values = {"a": 1, "b": 2}
        mock_do_request.return_value = (404, "", None)

        res = self.service.answer(event_id, answer, values, token=self.fake_token, timeout=DEFAULT_TIMEOUT)

        self.assertFalse(res)
        mock_do_request.assert_called_with(
            f"/v1.0/me/calendar/events/{event_id}/{answer}",
            json.dumps(values),
            self.header, method="POST", timeout=DEFAULT_TIMEOUT
        )

    @patch.object(MicrosoftService, "_do_request")
    def test_answer_to_an_existing_event(self, mock_do_request):
        event_id = 123
        answer = "decline"
        values = {"a": 1, "b": 2}
        mock_do_request.return_value = (200, "", None)

        res = self.service.answer(event_id, answer, values, token=self.fake_token, timeout=DEFAULT_TIMEOUT)

        self.assertTrue(res)
        mock_do_request.assert_called_with(
            f"/v1.0/me/calendar/events/{event_id}/{answer}",
            json.dumps(values),
            self.header, method="POST", timeout=DEFAULT_TIMEOUT
        )