diff --git a/owrx/active/list/__init__.py b/owrx/active/list/__init__.py new file mode 100644 index 0000000..5ab5f0f --- /dev/null +++ b/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__() diff --git a/test/owrx/active/__init__.py b/test/owrx/active/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/test/owrx/active/list/__init__.py b/test/owrx/active/list/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/test/owrx/active/list/test_active_list.py b/test/owrx/active/list/test_active_list.py new file mode 100644 index 0000000..5cb7772 --- /dev/null +++ b/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")