Extend MutableList and MutableDict functionality
Added next methods to MutableList: - append - extend - insert - pop - __setslice__ - __delslice__ Added next methods to MutableDict: - pop - popitem Fixed doc strings. Fixed copy and deepcopy. Covered by unittests. Change-Id: Ia35628f09f25f7008afa6fcf48c15f03bcbca51a Partial-Bug: 1482658
This commit is contained in:
Valentin Kaplov
parent
d26b966e67
commit
984f3bfc6a
|
|
@ -17,6 +17,34 @@
|
|||
import copy
|
||||
|
||||
from sqlalchemy.ext.mutable import Mutable
|
||||
from sqlalchemy.ext.mutable import MutableDict as MutableDictBase
|
||||
|
||||
|
||||
class MutableDict(MutableDictBase):
|
||||
# TODO(vkaplov): delete this class after methods pop and popitem
|
||||
# will be implemented in sqlalchemy lib.
|
||||
# https://bitbucket.org/zzzeek/sqlalchemy/issues/3605
|
||||
|
||||
def pop(self, *args):
|
||||
"""Pop element and emit change events if item removed.
|
||||
|
||||
:param args: (key, default) 'default' is optional
|
||||
:return: element
|
||||
:raises KeyError: if dict is empty or element not found.
|
||||
"""
|
||||
result = dict.pop(self, *args)
|
||||
self.changed()
|
||||
return result
|
||||
|
||||
def popitem(self):
|
||||
"""Pop arbitrary element and emit change events if item removed.
|
||||
|
||||
:return: (key, value)
|
||||
:raises KeyError: if dict is empty.
|
||||
"""
|
||||
result = dict.popitem(self)
|
||||
self.changed()
|
||||
return result
|
||||
|
||||
|
||||
class MutableList(Mutable, list):
|
||||
|
|
@ -26,7 +54,13 @@ class MutableList(Mutable, list):
|
|||
|
||||
@classmethod
|
||||
def coerce(cls, key, value):
|
||||
"""Convert plain lists to MutableList."""
|
||||
"""Convert plain lists to MutableList.
|
||||
|
||||
:param key: string name of the ORM-mapped attribute being set.
|
||||
:param value: the incoming value.
|
||||
:return: the method should return the coerced value
|
||||
:raises ValueError: if the coercion cannot be completed.
|
||||
"""
|
||||
|
||||
if not isinstance(value, MutableList):
|
||||
if isinstance(value, list):
|
||||
|
|
@ -37,31 +71,126 @@ class MutableList(Mutable, list):
|
|||
else:
|
||||
return value
|
||||
|
||||
def append(self, value):
|
||||
"""Append value to end and emit change events.
|
||||
|
||||
:param value: element value
|
||||
"""
|
||||
|
||||
list.append(self, value)
|
||||
self.changed()
|
||||
|
||||
def extend(self, iterable):
|
||||
"""Extend list and emit change events.
|
||||
|
||||
:param iterable: iterable on elements
|
||||
"""
|
||||
|
||||
list.extend(self, iterable)
|
||||
self.changed()
|
||||
|
||||
def insert(self, index, value):
|
||||
"""Insert value before index and emit change events.
|
||||
|
||||
:param index: index of next element
|
||||
:param value: value of inserting element
|
||||
"""
|
||||
|
||||
list.insert(self, index, value)
|
||||
self.changed()
|
||||
|
||||
def pop(self, index=-1):
|
||||
"""Pop element and emit change events if item removed.
|
||||
|
||||
:param index: index of element (default last)
|
||||
:return: element
|
||||
:raises IndexError: if list is empty or index is out of range.
|
||||
"""
|
||||
|
||||
result = list.pop(self, index)
|
||||
self.changed()
|
||||
return result
|
||||
|
||||
def remove(self, value):
|
||||
"""Remove first occurrence of value.
|
||||
|
||||
If occurrence removed, then emit change events.
|
||||
|
||||
:param value: value of element
|
||||
:raises ValueError: if the value is not present.
|
||||
"""
|
||||
|
||||
list.remove(self, value)
|
||||
self.changed()
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
"""Detect list set events and emit change events."""
|
||||
"""Detect list set events and emit change events.
|
||||
|
||||
:param key: index of element
|
||||
:param value: new value of element
|
||||
"""
|
||||
|
||||
list.__setitem__(self, key, value)
|
||||
self.changed()
|
||||
|
||||
def __delitem__(self, key):
|
||||
"""Detect list del events and emit change events."""
|
||||
"""Detect list del events and emit change events.
|
||||
|
||||
:param key: index of element
|
||||
"""
|
||||
|
||||
list.__delitem__(self, key)
|
||||
self.changed()
|
||||
|
||||
def __getstate__(self):
|
||||
"""Get state as builtin list
|
||||
|
||||
:return: current state
|
||||
"""
|
||||
|
||||
return list(self)
|
||||
|
||||
def __setstate__(self, state):
|
||||
"""Detect setstate event and emit change events.
|
||||
|
||||
:param state: new object state
|
||||
"""
|
||||
|
||||
self[:] = state
|
||||
|
||||
def __setslice__(self, first, last, sequence):
|
||||
"""Envoke setslice and emit change events.
|
||||
|
||||
:param first: first element index
|
||||
:param last: last (not included) element index
|
||||
:param sequence: elements that to be inserted
|
||||
"""
|
||||
|
||||
list.__setslice__(self, first, last, sequence)
|
||||
self.changed()
|
||||
|
||||
def __delslice__(self, first, last):
|
||||
"""Envoke delslice and emit change events.
|
||||
|
||||
:param first: first element index
|
||||
:param last: last (not included) element index
|
||||
"""
|
||||
|
||||
list.__delslice__(self, first, last)
|
||||
self.changed()
|
||||
|
||||
@classmethod
|
||||
def __copy__(cls, value):
|
||||
"""use copy via constructor."""
|
||||
"""Create and return copy of value.
|
||||
|
||||
return MutableList(value)
|
||||
:param value: MutableList object
|
||||
"""
|
||||
clone = MutableList()
|
||||
clone.__setstate__(value)
|
||||
return clone
|
||||
|
||||
def __deepcopy__(self, memo, _deepcopy=copy.deepcopy):
|
||||
"""recursive copy each element."""
|
||||
|
||||
return MutableList(_deepcopy(x, memo) for x in self)
|
||||
"""Recursive copy each element."""
|
||||
clone = MutableList()
|
||||
clone.__setstate__((_deepcopy(x, memo) for x in self))
|
||||
return clone
|
||||
|
|
|
|||
|
|
@ -21,8 +21,6 @@ from nailgun.db.sqlalchemy.models import NodeBondInterface
|
|||
from nailgun.db.sqlalchemy.models import NodeNICInterface
|
||||
from nailgun.db.sqlalchemy.models import Release
|
||||
|
||||
from nailgun.db.sqlalchemy.models.mutable import MutableList
|
||||
|
||||
from nailgun.test.base import BaseTestCase
|
||||
from oslo_serialization import jsonutils
|
||||
|
||||
|
|
@ -247,13 +245,3 @@ class TestNodeInterfacesDbModels(BaseTestCase):
|
|||
{'test_property': 'test_value'}) # str type cause ValueError
|
||||
}
|
||||
self.assertRaises(ValueError, NodeBondInterface, **sample_bond_data)
|
||||
|
||||
def test_copy_mutable_list(self):
|
||||
mlist = MutableList([{"a": 1, "b": 2}])
|
||||
shallow_copy = copy.copy(mlist)
|
||||
self.assertEqual(mlist, shallow_copy)
|
||||
self.assertIs(mlist[0], shallow_copy[0])
|
||||
|
||||
deep_copy = copy.deepcopy(mlist)
|
||||
self.assertEqual(mlist, deep_copy)
|
||||
self.assertIsNot(mlist[0], deep_copy[0])
|
||||
|
|
|
|||
|
|
@ -0,0 +1,246 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright 2015 Mirantis, Inc.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from copy import copy
|
||||
from copy import deepcopy
|
||||
from mock import patch
|
||||
|
||||
from nailgun.db.sqlalchemy.models.mutable import MutableDict
|
||||
from nailgun.db.sqlalchemy.models.mutable import MutableList
|
||||
from nailgun.test.base import BaseUnitTest
|
||||
|
||||
|
||||
@patch('nailgun.db.sqlalchemy.models.mutable.MutableDict.changed')
|
||||
class TestMutableDict(BaseUnitTest):
|
||||
def setUp(self):
|
||||
self.standard = {'1': 1}
|
||||
self.mutable_dict = MutableDict()
|
||||
self.mutable_dict.update(self.standard)
|
||||
|
||||
def test_pop_existing(self, m_changed):
|
||||
self.assertEqual(self.mutable_dict.pop('1'), self.standard.pop('1'))
|
||||
m_changed.assert_called_once_with()
|
||||
|
||||
def test_pop_existing_with_default(self, m_changed):
|
||||
self.assertEqual(self.mutable_dict.pop('1', None),
|
||||
self.standard.pop('1', None))
|
||||
m_changed.assert_called_once_with()
|
||||
|
||||
def test_pop_not_existing(self, m_changed):
|
||||
self.assertRaises(KeyError, self.mutable_dict.pop, '2')
|
||||
self.assertFalse(m_changed.called)
|
||||
|
||||
def test_pop_not_existing_with_default(self, m_changed):
|
||||
self.assertEqual(self.mutable_dict.pop('2', {}),
|
||||
self.standard.pop('2', {}))
|
||||
m_changed.assert_called_once_with()
|
||||
|
||||
def test_popitem(self, m_changed):
|
||||
self.assertItemsEqual(self.mutable_dict.popitem(),
|
||||
self.standard.popitem())
|
||||
m_changed.assert_called_once_with()
|
||||
|
||||
m_changed.reset_mock()
|
||||
self.assertRaises(KeyError, self.mutable_dict.popitem)
|
||||
self.assertFalse(m_changed.called)
|
||||
|
||||
|
||||
class TestMutableListBase(BaseUnitTest):
|
||||
def setUp(self):
|
||||
self.mutable_list = MutableList()
|
||||
|
||||
|
||||
@patch('sqlalchemy.ext.mutable.Mutable.coerce')
|
||||
class TestMutableListCoerce(TestMutableListBase):
|
||||
def setUp(self):
|
||||
super(TestMutableListCoerce, self).setUp()
|
||||
|
||||
def test_coerce_mutable_list(self, m_coerce):
|
||||
lst = MutableList()
|
||||
self.assertIsInstance(
|
||||
self.mutable_list.coerce('key', lst), MutableList)
|
||||
self.assertFalse(m_coerce.called)
|
||||
|
||||
def test_coerce_list(self, m_coerce):
|
||||
lst = list()
|
||||
self.assertIsInstance(
|
||||
self.mutable_list.coerce('key', lst), MutableList)
|
||||
self.assertFalse(m_coerce.called)
|
||||
|
||||
def test_coerce_not_acceptable_object(self, m_coerce):
|
||||
m_coerce.return_value = None
|
||||
obj = dict()
|
||||
self.mutable_list.coerce('key', obj)
|
||||
m_coerce.assert_called_once_with('key', obj)
|
||||
|
||||
|
||||
@patch('nailgun.db.sqlalchemy.models.mutable.MutableList.changed')
|
||||
class TestMutableList(TestMutableListBase):
|
||||
def setUp(self):
|
||||
super(TestMutableList, self).setUp()
|
||||
self.standard = ['element1', 'element2']
|
||||
self.mutable_list.extend(self.standard)
|
||||
|
||||
def _check(self, m_changed, method, *args, **kwargs):
|
||||
getattr(self.mutable_list, method)(*args, **kwargs)
|
||||
getattr(self.standard, method)(*args, **kwargs)
|
||||
self._assert_call_list_changed_once(m_changed)
|
||||
|
||||
def _assert_list_not_changed(self, m_changed):
|
||||
self.assertFalse(m_changed.called)
|
||||
self.assertItemsEqual(self.standard, self.mutable_list)
|
||||
|
||||
def _assert_call_list_changed_once(self, m_changed):
|
||||
m_changed.assert_called_once_with()
|
||||
self.assertItemsEqual(self.standard, self.mutable_list)
|
||||
|
||||
def test_append(self, m_changed):
|
||||
self._check(m_changed, 'append', 'element')
|
||||
|
||||
def test_extend(self, m_changed):
|
||||
self._check(m_changed, 'extend',
|
||||
('element3', 'element4', 'element5'))
|
||||
|
||||
def test_extend_failure(self, m_changed):
|
||||
self.assertRaises(TypeError, self.mutable_list.extend, None)
|
||||
self._assert_list_not_changed(m_changed)
|
||||
|
||||
def test_insert(self, m_changed):
|
||||
self._check(m_changed, 'insert', 1, 'new_element')
|
||||
|
||||
def test_insert_failure(self, m_changed):
|
||||
self.assertRaises(TypeError, self.mutable_list.insert, None, 'element')
|
||||
self._assert_list_not_changed(m_changed)
|
||||
|
||||
def test_pop_default(self, m_changed):
|
||||
self.assertEqual(self.standard.pop(), self.mutable_list.pop())
|
||||
self._assert_call_list_changed_once(m_changed)
|
||||
|
||||
def test_pop_of_specified_element(self, m_changed):
|
||||
self.assertEqual(self.standard.pop(0), self.mutable_list.pop(0))
|
||||
self._assert_call_list_changed_once(m_changed)
|
||||
|
||||
def test_pop_wrong_index_type(self, m_changed):
|
||||
self.assertRaises(TypeError, self.mutable_list.pop, 'str')
|
||||
self._assert_list_not_changed(m_changed)
|
||||
|
||||
def test_pop_out_of_range(self, m_changed):
|
||||
self.assertRaises(
|
||||
IndexError, self.mutable_list.pop, len(self.mutable_list))
|
||||
self._assert_list_not_changed(m_changed)
|
||||
|
||||
def test_remove(self, m_changed):
|
||||
self._check(m_changed, 'remove', 'element1')
|
||||
|
||||
def test_remove_failure(self, m_changed):
|
||||
self.assertRaises(ValueError, self.mutable_list.remove, None)
|
||||
self._assert_list_not_changed(m_changed)
|
||||
|
||||
def test_set_item(self, m_changed):
|
||||
self.mutable_list[0] = 'new_element'
|
||||
self.standard[0] = 'new_element'
|
||||
self._assert_call_list_changed_once(m_changed)
|
||||
|
||||
def test_set_item_failure(self, m_changed):
|
||||
self.assertRaises(
|
||||
IndexError, self.mutable_list.__setitem__,
|
||||
len(self.mutable_list), 'element')
|
||||
self._assert_list_not_changed(m_changed)
|
||||
|
||||
def test_del_item(self, m_changed):
|
||||
del self.mutable_list[0]
|
||||
del self.standard[0]
|
||||
self._assert_call_list_changed_once(m_changed)
|
||||
|
||||
def test_del_item_failure(self, m_changed):
|
||||
self.assertRaises(
|
||||
IndexError, self.mutable_list.__delitem__, len(self.mutable_list))
|
||||
self._assert_list_not_changed(m_changed)
|
||||
|
||||
def test_get_state(self, m_changed):
|
||||
self.assertIsInstance(self.mutable_list.__getstate__(), list)
|
||||
self._assert_list_not_changed(m_changed)
|
||||
|
||||
def test_set_state(self, m_changed):
|
||||
self.mutable_list.__setstate__([])
|
||||
self.standard = []
|
||||
self._assert_call_list_changed_once(m_changed)
|
||||
|
||||
def test_set_state_failure(self, m_changed):
|
||||
self.assertRaises(TypeError, self.mutable_list.__setstate__, None)
|
||||
self._assert_list_not_changed(m_changed)
|
||||
|
||||
def test_set_slice(self, m_changed):
|
||||
self._check(m_changed, '__setslice__', 0, 2, ('element',))
|
||||
|
||||
def test_set_slice_failure(self, m_changed):
|
||||
self.assertRaises(
|
||||
TypeError, self.mutable_list.__setslice__, 0, 5, None)
|
||||
self._assert_list_not_changed(m_changed)
|
||||
|
||||
def test_set_slice_integration(self, m_changed):
|
||||
self.mutable_list[0:2] = ['new_element']
|
||||
self.standard[0:2] = ['new_element']
|
||||
self._assert_call_list_changed_once(m_changed)
|
||||
|
||||
m_changed.reset_mock()
|
||||
self.mutable_list[:] = ["again_new_element"]
|
||||
self.standard[:] = ["again_new_element"]
|
||||
self._assert_call_list_changed_once(m_changed)
|
||||
|
||||
def test_del_slice(self, m_changed):
|
||||
self._check(m_changed, '__delslice__', 0, 2)
|
||||
|
||||
def test_del_slice_failure(self, m_changed):
|
||||
self.assertRaises(
|
||||
TypeError, self.mutable_list.__delslice__, 0, None)
|
||||
self._assert_list_not_changed(m_changed)
|
||||
|
||||
def test_del_slice_integration(self, m_changed):
|
||||
del self.mutable_list[0:2]
|
||||
del self.standard[0:2]
|
||||
self._assert_call_list_changed_once(m_changed)
|
||||
|
||||
def test_copy(self, m_changed):
|
||||
clone = copy(self.mutable_list)
|
||||
self.assertEqual(clone, self.mutable_list)
|
||||
m_changed.assert_called_once_with()
|
||||
|
||||
def test_deep_copy(self, m_changed):
|
||||
lst = MutableList(('element1', 'element2'))
|
||||
self.mutable_list.insert(0, lst)
|
||||
|
||||
m_changed.reset_mock()
|
||||
clone = deepcopy(self.mutable_list)
|
||||
# changed should calls two times
|
||||
# - root cloned list (clone)
|
||||
# - mutable list element in root list (cloned lst)
|
||||
self.assertEqual(m_changed.call_count, 2)
|
||||
|
||||
lst[0] = 'new_element'
|
||||
self.assertEqual(clone[0][0], 'element1')
|
||||
self.assertEqual(self.mutable_list[0][0], 'new_element')
|
||||
|
||||
|
||||
@patch('nailgun.db.sqlalchemy.models.mutable.MutableList.changed')
|
||||
class TestEmptyMutableList(TestMutableListBase):
|
||||
def setUp(self):
|
||||
super(TestEmptyMutableList, self).setUp()
|
||||
|
||||
def test_pop_default_from_empty_list(self, m_changed):
|
||||
self.assertRaises(IndexError, self.mutable_list.pop)
|
||||
self.assertFalse(m_changed.called)
|
||||
self.assertEqual([], self.mutable_list)
|
||||
Loading…
Reference in New Issue