[FIX] google_calendar,microsoft_calendar: use ReadonlyDict
To avoid developers to update the dict of `_events` in their overrides, to not alter by mistake the default behavior. Using a frozendict will force them to create a copy of the dict. closes odoo/odoo#123261 Signed-off-by: Arnaud Joset (arj) <arj@odoo.com>
This commit is contained in:
committed by
Lefebvre Thomas (thle)
parent
b75ff67d10
commit
7a33a52371
@@ -91,7 +91,7 @@ class GoogleService(models.AbstractModel):
|
||||
raise self.env['res.config.settings'].get_config_warning(error_msg)
|
||||
|
||||
@api.model
|
||||
def _do_request(self, uri, params=None, headers=None, method='POST', preuri="https://www.googleapis.com", timeout=TIMEOUT):
|
||||
def _do_request(self, uri, params=None, headers=None, method='POST', preuri=GOOGLE_API_BASE_URL, timeout=TIMEOUT):
|
||||
""" Execute the request to Google API. Return a tuple ('HTTP_CODE', 'HTTP_RESPONSE')
|
||||
:param uri : the url to contact
|
||||
:param params : dict or already encoded parameters for the request to make
|
||||
@@ -104,6 +104,10 @@ class GoogleService(models.AbstractModel):
|
||||
if headers is None:
|
||||
headers = {}
|
||||
|
||||
assert urls.url_parse(preuri + uri).host in [
|
||||
urls.url_parse(url).host for url in (GOOGLE_TOKEN_ENDPOINT, GOOGLE_API_BASE_URL)
|
||||
]
|
||||
|
||||
# Remove client_secret key from logs
|
||||
if isinstance(params, str):
|
||||
_log_params = json.loads(params) or {}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# Part of Odoo. See LICENSE file for full copyright and licensing details.
|
||||
|
||||
from . import test_google_event
|
||||
from . import test_sync_common
|
||||
from . import test_sync_google2odoo
|
||||
from . import test_sync_odoo2google
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# Part of Odoo. See LICENSE file for full copyright and licensing details.
|
||||
from odoo.tests.common import BaseCase
|
||||
from odoo.addons.google_calendar.utils.google_calendar import GoogleEvent
|
||||
|
||||
|
||||
class TestGoogleEvent(BaseCase):
|
||||
def test_google_event_readonly(self):
|
||||
event = GoogleEvent()
|
||||
with self.assertRaises(TypeError):
|
||||
event._events['foo'] = 'bar'
|
||||
with self.assertRaises(AttributeError):
|
||||
event._events.update({'foo': 'bar'})
|
||||
with self.assertRaises(TypeError):
|
||||
dict.update(event._events, {'foo': 'bar'})
|
||||
@@ -1,7 +1,7 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
# Part of Odoo. See LICENSE file for full copyright and licensing details.
|
||||
|
||||
from odoo.tools import email_normalize
|
||||
from odoo.tools import email_normalize, ReadonlyDict
|
||||
import logging
|
||||
from typing import Iterator, Mapping
|
||||
from collections import abc
|
||||
@@ -24,14 +24,15 @@ class GoogleEvent(abc.Set):
|
||||
"""
|
||||
|
||||
def __init__(self, iterable=()):
|
||||
self._events = {}
|
||||
_events = {}
|
||||
for item in iterable:
|
||||
if isinstance(item, self.__class__):
|
||||
self._events[item.id] = item._events[item.id]
|
||||
_events[item.id] = item._events[item.id]
|
||||
elif isinstance(item, Mapping):
|
||||
self._events[item.get('id')] = item
|
||||
_events[item.get('id')] = item
|
||||
else:
|
||||
raise ValueError("Only %s or iterable of dict are supported" % self.__class__.__name__)
|
||||
self._events = ReadonlyDict(_events)
|
||||
|
||||
def __iter__(self) -> Iterator['GoogleEvent']:
|
||||
return iter(GoogleEvent([vals]) for vals in self._events.values())
|
||||
|
||||
@@ -16,6 +16,7 @@ TIMEOUT = 20
|
||||
|
||||
DEFAULT_MICROSOFT_AUTH_ENDPOINT = 'https://login.microsoftonline.com/common/oauth2/v2.0/authorize'
|
||||
DEFAULT_MICROSOFT_TOKEN_ENDPOINT = 'https://login.microsoftonline.com/common/oauth2/v2.0/token'
|
||||
DEFAULT_MICROSOFT_GRAPH_ENDPOINT = 'https://graph.microsoft.com'
|
||||
|
||||
RESOURCE_NOT_FOUND_STATUSES = (204, 404)
|
||||
|
||||
@@ -124,7 +125,7 @@ class MicrosoftService(models.AbstractModel):
|
||||
raise self.env['res.config.settings'].get_config_warning(error_msg)
|
||||
|
||||
@api.model
|
||||
def _do_request(self, uri, params=None, headers=None, method='POST', preuri="https://graph.microsoft.com", timeout=TIMEOUT):
|
||||
def _do_request(self, uri, params=None, headers=None, method='POST', preuri=DEFAULT_MICROSOFT_GRAPH_ENDPOINT, timeout=TIMEOUT):
|
||||
""" Execute the request to Microsoft API. Return a tuple ('HTTP_CODE', 'HTTP_RESPONSE')
|
||||
:param uri : the url to contact
|
||||
:param params : dict or already encoded parameters for the request to make
|
||||
@@ -137,6 +138,10 @@ class MicrosoftService(models.AbstractModel):
|
||||
if headers is None:
|
||||
headers = {}
|
||||
|
||||
assert urls.url_parse(preuri + uri).host in [
|
||||
urls.url_parse(url).host for url in (DEFAULT_MICROSOFT_TOKEN_ENDPOINT, DEFAULT_MICROSOFT_GRAPH_ENDPOINT)
|
||||
]
|
||||
|
||||
_logger.debug("Uri: %s - Type : %s - Headers: %s - Params : %s !" % (uri, method, headers, params))
|
||||
|
||||
ask_time = fields.Datetime.now()
|
||||
|
||||
@@ -302,3 +302,12 @@ class TestMicrosoftEvent(TestCommon):
|
||||
|
||||
self.assertNotIn(self.simple_event, synced_events)
|
||||
self.assertIn(self.simple_event, not_synced_events)
|
||||
|
||||
def test_microsoft_event_readonly(self):
|
||||
event = MicrosoftEvent()
|
||||
with self.assertRaises(TypeError):
|
||||
event._events['foo'] = 'bar'
|
||||
with self.assertRaises(AttributeError):
|
||||
event._events.update({'foo': 'bar'})
|
||||
with self.assertRaises(TypeError):
|
||||
dict.update(event._events, {'foo': 'bar'})
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from odoo.api import model
|
||||
from typing import Iterator, Mapping
|
||||
from collections import abc
|
||||
from odoo.tools import ReadonlyDict
|
||||
from odoo.addons.microsoft_calendar.utils.event_id_storage import combine_ids
|
||||
|
||||
|
||||
@@ -17,14 +18,15 @@ class MicrosoftEvent(abc.Set):
|
||||
"""
|
||||
|
||||
def __init__(self, iterable=()):
|
||||
self._events = {}
|
||||
_events = {}
|
||||
for item in iterable:
|
||||
if isinstance(item, self.__class__):
|
||||
self._events[item.id] = item._events[item.id]
|
||||
_events[item.id] = item._events[item.id]
|
||||
elif isinstance(item, Mapping):
|
||||
self._events[item.get('id')] = item
|
||||
_events[item.get('id')] = item
|
||||
else:
|
||||
raise ValueError("Only %s or iterable of dict are supported" % self.__class__.__name__)
|
||||
self._events = ReadonlyDict(_events)
|
||||
|
||||
def __iter__(self) -> Iterator['MicrosoftEvent']:
|
||||
return iter(MicrosoftEvent([vals]) for vals in self._events.values())
|
||||
|
||||
@@ -496,3 +496,14 @@ class TestAddonsFileAccess(BaseCase):
|
||||
self.assertCannotRead(__file__, ValueError, filter_ext=('.png',))
|
||||
# file doesnt exist but has wrong extension
|
||||
self.assertCannotRead(__file__.replace('.py', '.foo'), ValueError, filter_ext=('.png',))
|
||||
|
||||
|
||||
class TestDictTools(BaseCase):
|
||||
def test_readonly_dict(self):
|
||||
d = misc.ReadonlyDict({'foo': 'bar'})
|
||||
with self.assertRaises(TypeError):
|
||||
d['baz'] = 'xyz'
|
||||
with self.assertRaises(AttributeError):
|
||||
d.update({'baz': 'xyz'})
|
||||
with self.assertRaises(TypeError):
|
||||
dict.update(d, {'baz': 'xyz'})
|
||||
|
||||
@@ -1656,6 +1656,40 @@ pickle.dumps = pickle_.dumps
|
||||
pickle.HIGHEST_PROTOCOL = pickle_.HIGHEST_PROTOCOL
|
||||
|
||||
|
||||
class ReadonlyDict(Mapping):
|
||||
"""Helper for an unmodifiable dictionary, not even updatable using `dict.update`.
|
||||
|
||||
This is similar to a `frozendict`, with one drawback and one advantage:
|
||||
|
||||
- `dict.update` works for a `frozendict` but not for a `ReadonlyDict`.
|
||||
- `json.dumps` works for a `frozendict` by default but not for a `ReadonlyDict`.
|
||||
|
||||
This comes from the fact `frozendict` inherits from `dict`
|
||||
while `ReadonlyDict` inherits from `collections.abc.Mapping`.
|
||||
|
||||
So, depending on your needs,
|
||||
whether you absolutely must prevent the dictionary from being updated (e.g., for security reasons)
|
||||
or you require it to be supported by `json.dumps`, you can choose either option.
|
||||
|
||||
E.g.
|
||||
data = ReadonlyDict({'foo': 'bar'})
|
||||
data['baz'] = 'xyz' # raises exception
|
||||
data.update({'baz', 'xyz'}) # raises exception
|
||||
dict.update(data, {'baz': 'xyz'}) # raises exception
|
||||
"""
|
||||
def __init__(self, data):
|
||||
self.__data = dict(data)
|
||||
|
||||
def __getitem__(self, key):
|
||||
return self.__data[key]
|
||||
|
||||
def __len__(self):
|
||||
return len(self.__data)
|
||||
|
||||
def __iter__(self):
|
||||
return iter(self.__data)
|
||||
|
||||
|
||||
class DotDict(dict):
|
||||
"""Helper for dot.notation access to dictionary attributes
|
||||
|
||||
|
||||
Reference in New Issue
Block a user