add first shot at active list implementation

owrx/active/list/__init__.py

@@ -0,0 +1,70 @@
+from abc import ABC, abstractmethod
+
+import logging
+logger = logging.getLogger(__name__)
+
+
+class ActiveListListener(ABC):
+    @abstractmethod
+    def onIndexChanged(self, index, newValue):
+        pass
+
+    @abstractmethod
+    def onAppend(self, newValue):
+        pass
+
+    @abstractmethod
+    def onDelete(self, index):
+        pass
+
+
+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)
+        for listener in self.listeners:
+            try:
+                listener.onAppend(value)
+            except Exception:
+                logger.exception("Exception during onAppend notification")
+
+    def remove(self, value):
+        self.__delitem__(self.delegate.index(value))
+
+    def __setitem__(self, key, value):
+        self.delegate[key] = value
+        for listener in self.listeners:
+            try:
+                listener.onIndexChanged(key, value)
+            except Exception:
+                logger.exception("Exception during onKeyChanged notification")
+
+    def __delitem__(self, key):
+        del self.delegate[key]
+        for listener in self.listeners:
+            try:
+                listener.onDelete(key)
+            except Exception:
+                logger.exception("Exception during onDelete notification")
+
+    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,73 @@
+from owrx.active.list import ActiveList
+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.onIndexChanged.assert_called_once_with(0, "testvalue")
+
+    def testListIndexChangeNotficationNotDisturbedByException(self):
+        list = ActiveList(["initialvalue"])
+        throwingMock = Mock()
+        throwingMock.onIndexChanged.side_effect = RuntimeError("this is a drill")
+        list.addListener(throwingMock)
+        listenerMock = Mock()
+        list.addListener(listenerMock)
+        list[0] = "testvalue"
+        listenerMock.onIndexChanged.assert_called_once_with(0, "testvalue")
+
+    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.onAppend.assert_called_once_with("testvalue")
+
+    def testListDelete(self):
+        list = ActiveList(["value1", "value2"])
+        del list[0]
+        self.assertEqual(len(list), 1)
+        self.assertEqual(list[0], "value2")
+
+    def testListDelteNotification(self):
+        list = ActiveList(["value1", "value2"])
+        listenerMock = Mock()
+        list.addListener(listenerMock)
+        del list[0]
+        listenerMock.onDelete.assert_called_once_with(0)
+
+    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")