ITQC/pydase
24
0
Fork 0
mirror of https://github.com/tiqi-group/pydase.git synced 2025-04-21 16:50:02 +02:00
Code Issues Packages Projects Releases Wiki Activity

implements clear, insert, remove, extend and pop for observable lists

Browse Source
This commit is contained in:
Mose Müller 2023-12-06 18:02:26 +01:00
parent f6bf229c8c
commit 8e641c1b84
2 changed files with 237 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

69
src/pydase/observer_pattern/observable/observable_object.py
View File

@ -153,39 +153,60 @@ class _ObservableList(ObservableObject, list[Any]):
self._notify_changed("", self)
def clear(self) -> None:
logger.warning(
"'clear' has not been overridden yet. This might lead to unexpected "
"behaviour."
)
self._remove_self_from_observables()
super().clear()
self._notify_changed("", self)
def extend(self, __iterable: Iterable[Any]) -> None:
logger.warning(
"'extend' has not been overridden yet. This might lead to unexpected "
"behaviour."
)
return super().extend(__iterable)
self._remove_self_from_observables()
try:
super().extend(__iterable)
finally:
for i, item in enumerate(self):
super().__setitem__(i, self._initialise_new_objects(f"[{i}]", item))
self._notify_changed("", self)
def insert(self, __index: SupportsIndex, __object: Any) -> None:
logger.warning(
"'insert' has not been overridden yet. This might lead to unexpected "
"behaviour."
)
super().insert(__index, __object)
self._remove_self_from_observables()
try:
super().insert(__index, __object)
finally:
for i, item in enumerate(self):
super().__setitem__(i, self._initialise_new_objects(f"[{i}]", item))
self._notify_changed("", self)
def pop(self, __index: SupportsIndex = -1) -> Any:
logger.warning(
"'pop' has not been overridden yet. This might lead to unexpected "
"behaviour."
)
return super().pop(__index)
self._remove_self_from_observables()
try:
popped_item = super().pop(__index)
finally:
for i, item in enumerate(self):
super().__setitem__(i, self._initialise_new_objects(f"[{i}]", item))
self._notify_changed("", self)
return popped_item
def remove(self, __value: Any) -> None:
logger.warning(
"'remove' has not been overridden yet. This might lead to unexpected "
"behaviour."
)
super().remove(__value)
self._remove_self_from_observables()
try:
super().remove(__value)
finally:
for i, item in enumerate(self):
super().__setitem__(i, self._initialise_new_objects(f"[{i}]", item))
self._notify_changed("", self)
def _remove_self_from_observables(self) -> None:
for i in range(len(self)):
self._remove_observer_if_observable(f"[{i}]")
def _remove_observer_if_observable(self, name: str) -> None:
key = int(name[1:-1])

233
tests/observer_pattern/observable/test_observable_object.py
View File

@ -20,7 +20,7 @@ def test_simple_instance_list_attribute(caplog: pytest.LogCaptureFixture) -> Non
self.list_attr = [1, 2]
instance = MyObservable()
observer = MyObserver(instance)
MyObserver(instance)
instance.list_attr[0] = 12
assert "'list_attr[0]' changed to '12'" in caplog.text
@ -38,7 +38,7 @@ def test_instance_object_list_attribute(caplog: pytest.LogCaptureFixture) -> Non
self.list_attr = [NestedObservable()]
instance = MyObservable()
observer = MyObserver(instance)
MyObserver(instance)
instance.list_attr[0].name = "Ciao"
assert "'list_attr[0].name' changed to 'Ciao'" in caplog.text
@ -49,7 +49,7 @@ def test_simple_class_list_attribute(caplog: pytest.LogCaptureFixture) -> None:
list_attr = [1, 2]
instance = MyObservable()
observer = MyObserver(instance)
MyObserver(instance)
instance.list_attr[0] = 12
assert "'list_attr[0]' changed to '12'" in caplog.text
@ -63,7 +63,7 @@ def test_class_object_list_attribute(caplog: pytest.LogCaptureFixture) -> None:
list_attr = [NestedObservable()]
instance = MyObservable()
observer = MyObserver(instance)
MyObserver(instance)
instance.list_attr[0].name = "Ciao"
assert "'list_attr[0].name' changed to 'Ciao'" in caplog.text
@ -76,7 +76,7 @@ def test_simple_instance_dict_attribute(caplog: pytest.LogCaptureFixture) -> Non
self.dict_attr = {"first": "Hello"}
instance = MyObservable()
observer = MyObserver(instance)
MyObserver(instance)
instance.dict_attr["first"] = "Ciao"
instance.dict_attr["second"] = "World"
@ -89,7 +89,7 @@ def test_simple_class_dict_attribute(caplog: pytest.LogCaptureFixture) -> None:
dict_attr = {"first": "Hello"}
instance = MyObservable()
observer = MyObserver(instance)
MyObserver(instance)
instance.dict_attr["first"] = "Ciao"
instance.dict_attr["second"] = "World"
@ -109,7 +109,7 @@ def test_instance_dict_attribute(caplog: pytest.LogCaptureFixture) -> None:
self.dict_attr = {"first": NestedObservable()}
instance = MyObservable()
observer = MyObserver(instance)
MyObserver(instance)
instance.dict_attr["first"].name = "Ciao"
assert "'dict_attr['first'].name' changed to 'Ciao'" in caplog.text
@ -123,7 +123,7 @@ def test_class_dict_attribute(caplog: pytest.LogCaptureFixture) -> None:
dict_attr = {"first": NestedObservable()}
instance = MyObservable()
observer = MyObserver(instance)
MyObserver(instance)
instance.dict_attr["first"].name = "Ciao"
assert "'dict_attr['first'].name' changed to 'Ciao'" in caplog.text
@ -140,7 +140,7 @@ def test_removed_observer_on_class_list_attr(caplog: pytest.LogCaptureFixture) -
changed_list_attr = [nested_instance]
instance = MyObservable()
observer = MyObserver(instance)
MyObserver(instance)
instance.changed_list_attr[0] = "Ciao"
assert "'changed_list_attr[0]' changed to 'Ciao'" in caplog.text
@ -169,7 +169,7 @@ def test_removed_observer_on_instance_dict_attr(
self.changed_dict_attr = {"nested": nested_instance}
instance = MyObservable()
observer = MyObserver(instance)
MyObserver(instance)
instance.changed_dict_attr["nested"] = "Ciao"
assert "'changed_dict_attr['nested']' changed to 'Ciao'" in caplog.text
@ -198,7 +198,7 @@ def test_removed_observer_on_instance_list_attr(
self.changed_list_attr = [nested_instance]
instance = MyObservable()
observer = MyObserver(instance)
MyObserver(instance)
instance.changed_list_attr[0] = "Ciao"
assert "'changed_list_attr[0]' changed to 'Ciao'" in caplog.text
@ -225,7 +225,7 @@ def test_removed_observer_on_class_dict_attr(caplog: pytest.LogCaptureFixture) -
self.changed_dict_attr = {"nested": nested_instance}
instance = MyObservable()
observer = MyObserver(instance)
MyObserver(instance)
instance.changed_dict_attr["nested"] = "Ciao"
assert "'changed_dict_attr['nested']' changed to 'Ciao'" in caplog.text
@ -246,7 +246,7 @@ def test_nested_dict_instances(caplog: pytest.LogCaptureFixture) -> None:
self.nested_dict_attr = {"nested": dict_instance}
instance = MyObservable()
observer = MyObserver(instance)
MyObserver(instance)
instance.nested_dict_attr["nested"]["first"] = "Ciao"
assert "'nested_dict_attr['nested']['first']' changed to 'Ciao'" in caplog.text
@ -261,14 +261,14 @@ def test_dict_in_list_instance(caplog: pytest.LogCaptureFixture) -> None:
self.dict_in_list = [dict_instance]
instance = MyObservable()
observer = MyObserver(instance)
MyObserver(instance)
instance.dict_in_list[0]["first"] = "Ciao"
assert "'dict_in_list[0]['first']' changed to 'Ciao'" in caplog.text
def test_list_in_dict_instance(caplog: pytest.LogCaptureFixture) -> None:
list_instance = [1, 2, 3]
list_instance: list[Any] = [1, 2, 3]
class MyObservable(Observable):
def __init__(self) -> None:
@ -276,48 +276,199 @@ def test_list_in_dict_instance(caplog: pytest.LogCaptureFixture) -> None:
self.list_in_dict = {"some_list": list_instance}
instance = MyObservable()
observer = MyObserver(instance)
MyObserver(instance)
instance.list_in_dict["some_list"][0] = "Ciao"
assert "'list_in_dict['some_list'][0]' changed to 'Ciao'" in caplog.text
def test_list_warnings(caplog: pytest.LogCaptureFixture) -> None:
def test_list_append(caplog: pytest.LogCaptureFixture) -> None:
class OtherObservable(Observable):
def __init__(self) -> None:
super().__init__()
self.greeting = "Other Observable"
class MyObservable(Observable):
def __init__(self) -> None:
super().__init__()
self.my_list = [1, 2, 3]
self.my_list = []
observable_instance = MyObservable()
MyObserver(observable_instance)
observable_instance.my_list.insert(1, -1)
assert (
"'insert' has not been overridden yet. This might lead to unexpected "
"behaviour."
) in caplog.text
observable_instance.my_list.append(OtherObservable())
assert f"'my_list' changed to '{observable_instance.my_list}'" in caplog.text
caplog.clear()
observable_instance.my_list.extend([1])
assert (
"'extend' has not been overridden yet. This might lead to unexpected "
"behaviour."
) in caplog.text
observable_instance.my_list.append(OtherObservable())
assert f"'my_list' changed to '{observable_instance.my_list}'" in caplog.text
caplog.clear()
observable_instance.my_list.remove(1)
assert (
"'remove' has not been overridden yet. This might lead to unexpected "
"behaviour."
) in caplog.text
observable_instance.my_list[0].greeting = "Hi"
observable_instance.my_list[1].greeting = "Hello"
assert observable_instance.my_list[0].greeting == "Hi"
assert observable_instance.my_list[1].greeting == "Hello"
assert "'my_list[0].greeting' changed to 'Hi'" in caplog.text
assert "'my_list[1].greeting' changed to 'Hello'" in caplog.text
def test_list_pop(caplog: pytest.LogCaptureFixture) -> None:
class OtherObservable(Observable):
def __init__(self) -> None:
super().__init__()
self.greeting = "Hello there!"
class MyObservable(Observable):
def __init__(self) -> None:
super().__init__()
self.my_list = [OtherObservable() for _ in range(2)]
observable_instance = MyObservable()
MyObserver(observable_instance)
popped_instance = observable_instance.my_list.pop(0)
assert len(observable_instance.my_list) == 1
assert f"'my_list' changed to '{observable_instance.my_list}'" in caplog.text
# checks if observer is removed
popped_instance.greeting = "Ciao"
assert "'my_list[0].greeting' changed to 'Ciao'" not in caplog.text
caplog.clear()
observable_instance.my_list.pop()
assert (
"'pop' has not been overridden yet. This might lead to unexpected behaviour."
) in caplog.text
caplog.clear()
# checks if observer keys have been updated (index 1 moved to 0)
observable_instance.my_list[0].greeting = "Hi"
assert "'my_list[0].greeting' changed to 'Hi'" in caplog.text
def test_list_clear(caplog: pytest.LogCaptureFixture) -> None:
class OtherObservable(Observable):
def __init__(self) -> None:
super().__init__()
self.greeting = "Hello there!"
other_observable_instance = OtherObservable()
class MyObservable(Observable):
def __init__(self) -> None:
super().__init__()
self.my_list = [other_observable_instance]
observable_instance = MyObservable()
MyObserver(observable_instance)
other_observable_instance.greeting = "Hello"
assert "'my_list[0].greeting' changed to 'Hello'" in caplog.text
observable_instance.my_list.clear()
assert (
"'clear' has not been overridden yet. This might lead to unexpected behaviour."
) in caplog.text
assert len(observable_instance.my_list) == 0
assert "'my_list' changed to '[]'" in caplog.text
# checks if observer has been removed
other_observable_instance.greeting = "Hi"
assert "'my_list[0].greeting' changed to 'Hi'" not in caplog.text
def test_list_extend(caplog: pytest.LogCaptureFixture) -> None:
class OtherObservable(Observable):
def __init__(self) -> None:
super().__init__()
self.greeting = "Hello there!"
other_observable_instance = OtherObservable()
class MyObservable(Observable):
def __init__(self) -> None:
super().__init__()
self.my_list = []
observable_instance = MyObservable()
MyObserver(observable_instance)
other_observable_instance.greeting = "Hello"
assert "'my_list[0].greeting' changed to 'Hello'" not in caplog.text
observable_instance.my_list.extend([other_observable_instance, OtherObservable()])
assert len(observable_instance.my_list) == 2
assert f"'my_list' changed to '{observable_instance.my_list}'" in caplog.text
# checks if observer has been removed
other_observable_instance.greeting = "Hi"
assert "'my_list[0].greeting' changed to 'Hi'" in caplog.text
observable_instance.my_list[1].greeting = "Ciao"
assert "'my_list[1].greeting' changed to 'Ciao'" in caplog.text
def test_list_insert(caplog: pytest.LogCaptureFixture) -> None:
class OtherObservable(Observable):
def __init__(self) -> None:
super().__init__()
self.greeting = "Hello there!"
other_observable_instance_1 = OtherObservable()
other_observable_instance_2 = OtherObservable()
class MyObservable(Observable):
def __init__(self) -> None:
super().__init__()
self.my_list = [other_observable_instance_1, OtherObservable()]
observable_instance = MyObservable()
MyObserver(observable_instance)
other_observable_instance_1.greeting = "Hello"
assert "'my_list[0].greeting' changed to 'Hello'" in caplog.text
observable_instance.my_list.insert(0, other_observable_instance_2)
assert len(observable_instance.my_list) == 3
assert f"'my_list' changed to '{observable_instance.my_list}'" in caplog.text
# checks if observer keys have been updated
other_observable_instance_2.greeting = "Hey"
other_observable_instance_1.greeting = "Hi"
observable_instance.my_list[2].greeting = "Ciao"
assert "'my_list[0].greeting' changed to 'Hey'" in caplog.text
assert "'my_list[1].greeting' changed to 'Hi'" in caplog.text
assert "'my_list[2].greeting' changed to 'Ciao'" in caplog.text
def test_list_remove(caplog: pytest.LogCaptureFixture) -> None:
class OtherObservable(Observable):
def __init__(self) -> None:
super().__init__()
self.greeting = "Hello there!"
other_observable_instance_1 = OtherObservable()
other_observable_instance_2 = OtherObservable()
class MyObservable(Observable):
def __init__(self) -> None:
super().__init__()
self.my_list = [other_observable_instance_1, other_observable_instance_2]
observable_instance = MyObservable()
MyObserver(observable_instance)
other_observable_instance_1.greeting = "Hello"
other_observable_instance_2.greeting = "Hi"
caplog.clear()
observable_instance.my_list.remove(other_observable_instance_1)
assert len(observable_instance.my_list) == 1
assert f"'my_list' changed to '{observable_instance.my_list}'" in caplog.text
caplog.clear()
# checks if observer has been removed
other_observable_instance_1.greeting = "Hi"
assert "'my_list[0].greeting' changed to 'Hi'" not in caplog.text
caplog.clear()
# checks if observer key was updated correctly (was index 1)
other_observable_instance_2.greeting = "Ciao"
assert "'my_list[0].greeting' changed to 'Ciao'" in caplog.text