implement active list transformation

owrx/active/list/__init__.py

@@ -0,0 +1,104 @@
+from abc import ABC, abstractmethod
+
+import logging
+logger = logging.getLogger(__name__)
+
+
+class ActiveListChange(ABC):
+    pass
+
+
+class ActiveListIndexUpdated(ActiveListChange):
+    def __init__(self, index: int, oldValue, newValue):
+        self.index = index
+        self.oldValue = oldValue
+        self.newValue = newValue
+
+
+class ActiveListIndexAppended(ActiveListChange):
+    def __init__(self, index: int, newValue):
+        self.index = index
+        self.newValue = newValue
+
+
+class ActiveListIndexDeleted(ActiveListChange):
+    def __init__(self, index: int, oldValue):
+        self.index = index
+        self.oldValue = oldValue
+
+
+class ActiveListListener(ABC):
+    @abstractmethod
+    def onListChange(self, changes: list[ActiveListChange]):
+        pass
+
+
+class ActiveListTransformationListener(ActiveListListener):
+    def __init__(self, transformation: callable, target: "ActiveList"):
+        self.transformation = transformation
+        self.target = target
+
+    def onListChange(self, changes: list[ActiveListChange]):
+        for change in changes:
+            if isinstance(change, ActiveListIndexUpdated):
+                self.target[change.index] = self.transformation(change.newValue)
+            elif isinstance(change, ActiveListIndexAppended):
+                self.target.append(self.transformation(change.newValue))
+            elif isinstance(change, ActiveListIndexDeleted):
+                del self.target[change.index]
+
+
+class ActiveList:
+    def __init__(self, elements: list = None):
+        self.delegate = elements.copy() if elements is not None else []
+        self.listeners = []
+
+    def addListener(self, listener: ActiveListListener):
+        if listener in self.listeners:
+            return
+        self.listeners.append(listener)
+
+    def removeListener(self, listener: ActiveListListener):
+        if listener not in self.listeners:
+            return
+        self.listeners.remove(listener)
+
+    def append(self, value):
+        self.delegate.append(value)
+        self.__fireChanges([ActiveListIndexAppended(len(self) - 1, value)])
+
+    def __fireChanges(self, changes: list[ActiveListChange]):
+        for listener in self.listeners:
+            try:
+                listener.onListChange(changes)
+            except Exception:
+                logger.exception("Exception during onListChange notification")
+
+    def remove(self, value):
+        self.__delitem__(self.delegate.index(value))
+
+    def map(self, transform: callable):
+        res = ActiveList([transform(v) for v in self])
+        self.addListener(ActiveListTransformationListener(transform, res))
+        return res
+
+    def __setitem__(self, key, value):
+        if self.delegate[key] == value:
+            return
+        oldValue = self.delegate[key]
+        self.delegate[key] = value
+        self.__fireChanges([ActiveListIndexUpdated(key, oldValue, value)])
+
+    def __delitem__(self, key):
+        oldValue = self.delegate[key]
+        del self.delegate[key]
+        self.__fireChanges([ActiveListIndexDeleted(key, oldValue)])
+
+    def __getitem__(self, key):
+        return self.delegate[key]
+
+    def __len__(self):
+        return len(self.delegate)
+
+    def __iter__(self):
+        return self.delegate.__iter__()

test/owrx/active/list/test_active_list.py

@@ -0,0 +1,123 @@
+from owrx.active.list import ActiveList, ActiveListIndexUpdated, ActiveListIndexAppended, ActiveListIndexDeleted
+from unittest import TestCase
+from unittest.mock import Mock
+
+
+class ActiveListTest(TestCase):
+    def testListIndexReadAccess(self):
+        list = ActiveList(["testvalue"])
+        self.assertEqual(list[0], "testvalue")
+
+    def testListIndexWriteAccess(self):
+        list = ActiveList(["initialvalue"])
+        list[0] = "testvalue"
+        self.assertEqual(list[0], "testvalue")
+
+    def testListLength(self):
+        list = ActiveList(["somevalue"])
+        self.assertEqual(len(list), 1)
+
+    def testListIndexChangeNotification(self):
+        list = ActiveList(["initialvalue"])
+        listenerMock = Mock()
+        list.addListener(listenerMock)
+        list[0] = "testvalue"
+        listenerMock.onListChange.assert_called_once()
+        changes, = listenerMock.onListChange.call_args.args
+        self.assertEqual(len(changes), 1)
+        self.assertIsInstance(changes[0], ActiveListIndexUpdated)
+        self.assertEqual(changes[0].index, 0)
+        self.assertEqual(changes[0].oldValue, "initialvalue")
+        self.assertEqual(changes[0].newValue, "testvalue")
+
+    def testListIndexChangeNotficationNotDisturbedByException(self):
+        list = ActiveList(["initialvalue"])
+        throwingMock = Mock()
+        throwingMock.onListChange.side_effect = RuntimeError("this is a drill")
+        list.addListener(throwingMock)
+        listenerMock = Mock()
+        list.addListener(listenerMock)
+        list[0] = "testvalue"
+        listenerMock.onListChange.assert_called_once()
+
+    def testListAppend(self):
+        list = ActiveList()
+        list.append("testvalue")
+        self.assertEqual(len(list), 1)
+        self.assertEqual(list[0], "testvalue")
+
+    def testListAppendNotification(self):
+        list = ActiveList()
+        listenerMock = Mock()
+        list.addListener(listenerMock)
+        list.append("testvalue")
+        listenerMock.onListChange.assert_called_once()
+        changes, = listenerMock.onListChange.call_args.args
+        self.assertEqual(len(changes), 1)
+        self.assertIsInstance(changes[0], ActiveListIndexAppended)
+        self.assertEqual(changes[0].index, 0)
+        self.assertEqual(changes[0].newValue, "testvalue")
+
+    def testListDelete(self):
+        list = ActiveList(["value1", "value2"])
+        del list[0]
+        self.assertEqual(len(list), 1)
+        self.assertEqual(list[0], "value2")
+
+    def testListDeleteNotification(self):
+        list = ActiveList(["value1", "value2"])
+        listenerMock = Mock()
+        list.addListener(listenerMock)
+        del list[0]
+        listenerMock.onListChange.assert_called_once()
+        changes, = listenerMock.onListChange.call_args.args
+        self.assertEqual(len(changes), 1)
+        self.assertIsInstance(changes[0], ActiveListIndexDeleted)
+        self.assertEqual(changes[0].index, 0)
+        self.assertEqual(changes[0].oldValue, 'value1')
+
+    def testListDeleteByValue(self):
+        list = ActiveList(["value1", "value2"])
+        list.remove("value1")
+        self.assertEqual(len(list), 1)
+        self.assertEqual(list[0], "value2")
+
+    def testListComprehension(self):
+        list = ActiveList(["initialvalue"])
+        x = [m for m in list]
+        self.assertEqual(len(x), 1)
+        self.assertEqual(x[0], "initialvalue")
+
+    def testListenerRemoval(self):
+        list = ActiveList(["initialvalue"])
+        listenerMock = Mock()
+        list.addListener(listenerMock)
+        list[0] = "testvalue"
+        listenerMock.onListChange.assert_called_once()
+        listenerMock.reset_mock()
+        list.removeListener(listenerMock)
+        list[0] = "someothervalue"
+        listenerMock.onListChange.assert_not_called()
+
+    def testListMapTransformation(self):
+        list = ActiveList(["somevalue"])
+        transformedList = list.map(lambda x: "prefix-{}".format(x))
+        self.assertEqual(transformedList[0], "prefix-somevalue")
+
+    def testActiveTransformationUpdate(self):
+        list = ActiveList(["initialvalue"])
+        transformedList = list.map(lambda x: "prefix-{}".format(x))
+        list[0] = "testvalue"
+        self.assertEqual(transformedList[0], "prefix-testvalue")
+
+    def testActiveTransformationAppend(self):
+        list = ActiveList(["initialvalue"])
+        transformedList = list.map(lambda x: "prefix-{}".format(x))
+        list.append("newvalue")
+        self.assertEqual(transformedList[1], "prefix-newvalue")
+
+    def testActiveTransformationDelete(self):
+        list = ActiveList(["value1", "value2"])
+        transformedList = list.map(lambda x: "prefix-{}".format(x))
+        del list[0]
+        self.assertEqual(transformedList[0], "prefix-value2")