1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from collections import defaultdict
from datetime import datetime
from unittest.mock import patch
from odoo.addons.google_calendar.utils.google_calendar import GoogleCalendarService
from odoo.addons.google_account.models.google_service import GoogleService
from odoo.addons.google_calendar.models.res_users import User
from odoo.addons.google_calendar.models.google_sync import google_calendar_token, GoogleSync
from odoo.tests.common import HttpCase, new_test_user
from freezegun import freeze_time
from contextlib import contextmanager
def patch_api(func):
def patched(self, *args, **kwargs):
with self.mock_google_sync():
return func(self, *args, **kwargs)
return patched
@patch.object(User, '_get_google_calendar_token', lambda user: 'dummy-token')
class TestSyncGoogle(HttpCase):
def setUp(self):
super().setUp()
self.google_service = GoogleCalendarService(self.env['google.service'])
self.env.user.sudo().unpause_google_synchronization()
self.organizer_user = new_test_user(self.env, login="organizer_user")
self.attendee_user = new_test_user(self.env, login='attendee_user')
@contextmanager
def mock_datetime_and_now(self, mock_dt):
"""
Used when synchronization date (using env.cr.now()) is important
in addition to standard datetime mocks. Used mainly to detect sync
issues.
"""
with freeze_time(mock_dt), \
patch.object(self.env.cr, 'now', lambda: mock_dt):
yield
@contextmanager
def mock_google_sync(self, user_id=None):
self._gsync_deleted_ids = []
self._gsync_insert_values = []
self._gsync_patch_values = defaultdict(list)
# as these are normally post-commit hooks, we don't change any state here
def _mock_delete(model, service, google_id, **kwargs):
with google_calendar_token(user_id or model.env.user.sudo()) as token:
if token:
self._gsync_deleted_ids.append(google_id)
def _mock_insert(model, service, values, **kwargs):
if not values:
return
with google_calendar_token(user_id or model.env.user.sudo()) as token:
if token:
self._gsync_insert_values.append((values, kwargs))
def _mock_patch(model, service, google_id, values, **kwargs):
with google_calendar_token(user_id or model.env.user.sudo()) as token:
if token:
self._gsync_patch_values[google_id].append((values, kwargs))
with self.env.cr.savepoint(), \
patch.object(GoogleSync, '_google_insert', autospec=True, wraps=GoogleSync, side_effect=_mock_insert), \
patch.object(GoogleSync, '_google_delete', autospec=True, wraps=GoogleSync, side_effect=_mock_delete), \
patch.object(GoogleSync, '_google_patch', autospec=True, wraps=GoogleSync, side_effect=_mock_patch):
yield
@contextmanager
def mock_google_service(self):
self._gservice_request_uris = []
def _mock_do_request(model, uri, *args, **kwargs):
self._gservice_request_uris.append(uri)
return (200, {}, datetime.now())
with patch.object(GoogleService, '_do_request', autospec=True, wraps=GoogleService, side_effect=_mock_do_request):
yield
def assertGoogleEventDeleted(self, google_id):
self.assertIn(google_id, self._gsync_deleted_ids, "Event should have been deleted")
def assertGoogleEventNotDeleted(self):
self.assertFalse(self._gsync_deleted_ids)
def assertGoogleEventInserted(self, values, timeout=None):
self.assertEqual(len(self._gsync_insert_values), 1)
matching = []
for insert_values, insert_kwargs in self._gsync_insert_values:
if all(insert_values.get(key, False) == value for key, value in values.items()):
matching.append((insert_values, insert_kwargs))
self.assertGreaterEqual(len(matching), 1, 'There must be at least 1 matching insert.')
insert_values, insert_kwargs = matching[0]
self.assertDictEqual(insert_kwargs, {'timeout': timeout} if timeout else {})
def assertGoogleEventNotInserted(self):
self.assertFalse(self._gsync_insert_values)
def assertGoogleEventPatched(self, google_id, values, timeout=None):
patch_values_all = self._gsync_patch_values.get(google_id)
self.assertTrue(patch_values_all)
matching = []
for patch_values, patch_kwargs in patch_values_all:
if all(patch_values.get(key, False) == values[key] for key in values):
matching.append((patch_values, patch_kwargs))
self.assertGreaterEqual(len(matching), 1, 'There must be at least 1 matching patch.')
patch_values, patch_kwargs = matching[0]
self.assertDictEqual(patch_kwargs, {'timeout': timeout} if timeout else {})
def assertGoogleEventNotPatched(self):
self.assertFalse(self._gsync_patch_values)
def assertGoogleAPINotCalled(self):
self.assertGoogleEventNotPatched()
self.assertGoogleEventNotInserted()
self.assertGoogleEventNotDeleted()
def assertGoogleEventSendUpdates(self, expected_value):
self.assertEqual(len(self._gservice_request_uris), 1)
uri = self._gservice_request_uris[0]
uri_parameter = "sendUpdates=%s" % expected_value
self.assertIn(uri_parameter, uri, "The URL should contain %s" % uri_parameter)
def call_post_commit_hooks(self):
"""
manually calls postcommit hooks defined with the decorator @after_commit
"""
funcs = self.env.cr.postcommit._funcs.copy()
while funcs:
func = funcs.popleft()
func()
|
