diff --git a/src/pyDataInterface/data_service/data_service.py b/src/pyDataInterface/data_service/data_service.py index 7833eaf..a89f28d 100644 --- a/src/pyDataInterface/data_service/data_service.py +++ b/src/pyDataInterface/data_service/data_service.py @@ -3,7 +3,7 @@ import inspect import threading from collections.abc import Callable from concurrent.futures import Future -from typing import Any, cast +from typing import Any import rpyc from loguru import logger @@ -12,30 +12,6 @@ from .data_service_list import DataServiceList class DataService(rpyc.Service): - _full_access_path: set[str] - """ TODO: improve this docstring - A set of strings, each representing a unique path to access the attribute from an - exposed class instance. Each path starts with the name of the exposed class. It's - dynamically updated to accurately represent the current attribute structure. - This attribute is used to emit notifications to a web server whenever the attribute - changes, allowing for real-time tracking and updates of class instance - modifications. - - Example: - -------- - >>> class SubClass(DataService): - >>> pass - - >>> class ExposedClass(DataService): - >>> attr = SubClass() - - >>> service = ExposedClass() - >>> # ... expose class - - >>> print(service.attr._full_access_path) # {"ServiceClass.attr"} - - Have a look at tests/test_full_access_path.py to see more examples. - """ _list_mapping: dict[int, DataServiceList] = {} """ A dictionary mapping the id of the original lists to the corresponding @@ -47,19 +23,24 @@ class DataService(rpyc.Service): """ def __init__(self) -> None: + # Keep track of the root object. This helps to filter the emission of + # notifications + self._root: "DataService" = self + # dictionary to keep track of running tasks self.__tasks: dict[str, Future[None]] = {} self._autostart_tasks: dict[str, tuple[Any]] if "_autostart_tasks" not in self.__dict__: self._autostart_tasks = {} + self._callbacks: set[Callable[[str, Any], None]] = set() self._set_start_and_stop_for_async_methods() self._start_async_loop_in_thread() self._start_autostart_tasks() - self._update_full_access_path(self, f"{self.__class__.__name__}") - self._turn_lists_into_notify_lists() + self._register_callbacks(self, f"{self.__class__.__name__}") + self._turn_lists_into_notify_lists(self, f"{self.__class__.__name__}") self._do_something_with_properties() self._initialised = True @@ -69,43 +50,47 @@ class DataService(rpyc.Service): if isinstance(attr_value, property): # If attribute is a property logger.debug(attr_value.fget.__code__.co_names) - def _turn_lists_into_notify_lists(self) -> None: - def create_callback(attr_name: str) -> Callable: - """TODO: explain what this is used for... - Create a callback with current attr_name captured in the default argument. - - Default arguments solve the late binding problem by capturing the value at - the time the lambda is defined, not when it is called, thus preventing - attr_name from being overwritten in another loop iteratianother - """ - - return lambda index, value, attr_name=attr_name: self._emit( - access_path=self._full_access_path, - name=f"{attr_name}[{index}]", - value=value, - ) - + def _turn_lists_into_notify_lists( + self, obj: "DataService", parent_path: str + ) -> None: # Convert all list attributes (both class and instance) to DataServiceList - for attr_name in set(dir(self)) - set(dir(object)): - attr_value = getattr(self, attr_name) + for attr_name in set(dir(obj)) - set(dir(object)) - {"_root"}: + attr_value = getattr(obj, attr_name) - if isinstance(attr_value, list): + if isinstance(attr_value, DataService): + new_path = f"{parent_path}.{attr_name}" + self._turn_lists_into_notify_lists(attr_value, new_path) + elif isinstance(attr_value, list): # Create callback for current attr_name - callback = create_callback(attr_name) + # Default arguments solve the late binding problem by capturing the + # value at the time the lambda is defined, not when it is called. This + # prevents attr_name from being overwritten in the next loop iteration. + callback = ( + lambda index, value, attr_name=attr_name: self._emit_notification( + parent_path=parent_path, + name=f"{attr_name}[{index}]", + value=value, + ) + if self == self._root + else None + ) # Check if attr_value is already a DataServiceList or in the mapping if isinstance(attr_value, DataServiceList): attr_value.add_callback(callback) continue - - if id(attr_value) in self._list_mapping: + elif id(attr_value) in self._list_mapping: notifying_list = self._list_mapping[id(attr_value)] notifying_list.add_callback(callback) else: notifying_list = DataServiceList(attr_value, callback=[callback]) self._list_mapping[id(attr_value)] = notifying_list - setattr(self, attr_name, notifying_list) + setattr(obj, attr_name, notifying_list) + for i, item in enumerate(attr_value): + if isinstance(item, DataService): + new_path = f"{parent_path}.{attr_name}[{i}]" + self._turn_lists_into_notify_lists(item, new_path) def _start_autostart_tasks(self) -> None: if self._autostart_tasks is not None: @@ -151,37 +136,57 @@ class DataService(rpyc.Service): setattr(self, f"start_{name}", start_task) setattr(self, f"stop_{name}", stop_task) - def _update_full_access_path(self, obj: "DataService", parent_path: str) -> None: + def _register_callbacks(self, obj: "DataService", parent_path: str) -> None: """ - Recursive helper function to update '_full_access_path' for the object and all + Recursive helper function to register callbacks for the object and all its nested attributes """ - parent_class_name = parent_path.split(".")[0] if parent_path else None + # Create and register a callback for the object + # only emit the notification when the call was registered by the root object + callback: Callable[[str, Any], None] = ( + lambda name, value: obj._emit_notification( + parent_path=parent_path, name=name, value=value + ) + if self == self._root + else None + ) - # Remove all access paths that don't start with the parent class name. As the - # exposed class is instantiated last, this ensures that all access paths start - # with the root class - access_path: set[str] = { - p - for p in cast(list[str], getattr(obj, "_full_access_path", set())) - if not parent_class_name or p.startswith(parent_class_name) - } - # add the new access path - access_path.add(parent_path) - setattr(obj, "_full_access_path", access_path) + obj._callbacks.add(callback) - # Recursively update access paths for all nested attributes of the object - for nested_attr_name in set(dir(obj)) - set(dir(object)): + # Recursively register callbacks for all nested attributes of the object + attribute_set = set(dir(obj)) - set(dir(object)) - {"_root"} + for nested_attr_name in attribute_set: nested_attr = getattr(obj, nested_attr_name) if isinstance(nested_attr, list): - for i, list_item in enumerate(nested_attr): - if isinstance(list_item, DataService): - new_path = f"{parent_path}.{nested_attr_name}[{i}]" - self._update_full_access_path(list_item, new_path) + self._register_list_callbacks( + nested_attr, parent_path, nested_attr_name + ) elif isinstance(nested_attr, DataService): - new_path = f"{parent_path}.{nested_attr_name}" - self._update_full_access_path(nested_attr, new_path) + self._register_service_callbacks( + nested_attr, parent_path, nested_attr_name + ) + + def _register_list_callbacks( + self, nested_attr: list[Any], parent_path: str, attr_name: str + ) -> None: + """Handles registration of callbacks for list attributes""" + for i, list_item in enumerate(nested_attr): + if isinstance(list_item, DataService): + self._register_service_callbacks( + list_item, parent_path, f"{attr_name}[{i}]" + ) + + def _register_service_callbacks( + self, nested_attr: "DataService", parent_path: str, attr_name: str + ) -> None: + """Handles registration of callbacks for DataService attributes""" + + # as the DataService is an attribute of self, change the root object + nested_attr._root = self._root + + new_path = f"{parent_path}.{attr_name}" + self._register_callbacks(nested_attr, new_path) def _start_loop(self) -> None: asyncio.set_event_loop(self.__loop) @@ -195,17 +200,15 @@ class DataService(rpyc.Service): self.__thread.join() def __setattr__(self, __name: str, __value: Any) -> None: - if self.__dict__.get("_initialised"): - access_path: set[str] = getattr(self, "_full_access_path", set()) - if access_path: - self._emit(access_path, __name, __value) + super().__setattr__(__name, __value) + if self.__dict__.get("_initialised") and not __name == "_initialised": + for callback in self._callbacks: + callback(__name, __value) # TODO: add emits for properties -> can use co_names, which is a tuple # containing the names used by the bytecode - super().__setattr__(__name, __value) - def _emit(self, access_path: set[str], name: str, value: Any) -> None: - for path in access_path: - logger.debug(f"{path}.{name} changed to {value}!") + def _emit_notification(self, parent_path: str, name: str, value: Any) -> None: + logger.debug(f"{parent_path}.{name} changed to {value}!") def _rpyc_getattr(self, name: str) -> Any: if name.startswith("_"): diff --git a/tests/__init__.py b/tests/__init__.py index e69de29..8aea465 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -0,0 +1,13 @@ +from typing import Any + +from pyDataInterface import DataService + + +def emit(self: Any, parent_path: str, name: str, value: Any) -> None: + if isinstance(value, DataService): + value = value.serialize() + + print(f"{parent_path}.{name} = {value}") + + +DataService._emit_notification = emit # type: ignore diff --git a/tests/test_emit_on_change.py b/tests/test_DataServiceList.py similarity index 66% rename from tests/test_emit_on_change.py rename to tests/test_DataServiceList.py index 965063c..dd8799e 100644 --- a/tests/test_emit_on_change.py +++ b/tests/test_DataServiceList.py @@ -1,45 +1,8 @@ -from typing import Any - from pytest import CaptureFixture from pyDataInterface import DataService -def emit(self: Any, access_path: set[str], name: str, value: Any) -> None: - if isinstance(value, DataService): - value = value.serialize() - - for path in access_path: - print(f"{path}.{name} = {value}") - - -DataService._emit = emit # type: ignore - - -def test_class_attribute(capsys: CaptureFixture) -> None: - class ServiceClass(DataService): - attr = 0 - - service_instance = ServiceClass() - - service_instance.attr = 1 - captured = capsys.readouterr() - assert captured.out == "ServiceClass.attr = 1\n" - - -def test_instance_attribute(capsys: CaptureFixture) -> None: - class ServiceClass(DataService): - def __init__(self) -> None: - self.attr = "Hello World" - super().__init__() - - service_instance = ServiceClass() - - service_instance.attr = "Hello" - captured = capsys.readouterr() - assert captured.out == "ServiceClass.attr = Hello\n" - - def test_class_list_attribute(capsys: CaptureFixture) -> None: class ServiceClass(DataService): attr = [0, 1] @@ -89,15 +52,15 @@ def test_reused_instance_list_attribute(capsys: CaptureFixture) -> None: service_instance = ServiceClass() - service_instance.attr[0] = "Hello" + service_instance.attr[0] = 20 captured = capsys.readouterr() assert service_instance.attr == service_instance.attr_2 assert service_instance.attr != service_instance.attr_3 expected_output = sorted( [ - "ServiceClass.attr[0] = Hello", - "ServiceClass.attr_2[0] = Hello", + "ServiceClass.attr[0] = 20", + "ServiceClass.attr_2[0] = 20", ] ) actual_output = sorted(captured.out.strip().split("\n")) @@ -123,15 +86,15 @@ def test_nested_reused_instance_list_attribute(capsys: CaptureFixture) -> None: service_instance = ServiceClass() _ = capsys.readouterr() - service_instance.attr[0] = "Hello" + service_instance.attr[0] = 20 captured = capsys.readouterr() assert service_instance.attr == service_instance.subclass.attr_list expected_output = sorted( [ - "ServiceClass.subclass.attr_list_2[0] = Hello", - "ServiceClass.subclass.attr_list[0] = Hello", - "ServiceClass.attr[0] = Hello", + "ServiceClass.subclass.attr_list_2[0] = 20", + "ServiceClass.subclass.attr_list[0] = 20", + "ServiceClass.attr[0] = 20", ] ) actual_output = sorted(captured.out.strip().split("\n")) diff --git a/tests/test_DataService_attribute_callbacks.py b/tests/test_DataService_attribute_callbacks.py new file mode 100644 index 0000000..e19d95a --- /dev/null +++ b/tests/test_DataService_attribute_callbacks.py @@ -0,0 +1,568 @@ +from pytest import CaptureFixture + +from pyDataInterface import DataService + + +def test_class_attributes(capsys: CaptureFixture) -> None: + class SubClass(DataService): + name = "Hello" + + class ServiceClass(DataService): + attr_1 = SubClass() + + service_instance = ServiceClass() + _ = capsys.readouterr() + service_instance.attr_1.name = "Hi" + + captured = capsys.readouterr() + assert captured.out.strip() == "ServiceClass.attr_1.name = Hi" + + +def test_instance_attributes(capsys: CaptureFixture) -> None: + class SubClass(DataService): + name = "Hello" + + class ServiceClass(DataService): + def __init__(self) -> None: + self.attr_1 = SubClass() + super().__init__() + + service_instance = ServiceClass() + _ = capsys.readouterr() + service_instance.attr_1.name = "Hi" + + captured = capsys.readouterr() + assert captured.out.strip() == "ServiceClass.attr_1.name = Hi" + + +def test_class_attribute(capsys: CaptureFixture) -> None: + class ServiceClass(DataService): + attr = 0 + + service_instance = ServiceClass() + + service_instance.attr = 1 + captured = capsys.readouterr() + assert captured.out == "ServiceClass.attr = 1\n" + + +def test_instance_attribute(capsys: CaptureFixture) -> None: + class ServiceClass(DataService): + def __init__(self) -> None: + self.attr = "Hello World" + super().__init__() + + service_instance = ServiceClass() + + service_instance.attr = "Hello" + captured = capsys.readouterr() + assert captured.out == "ServiceClass.attr = Hello\n" + + +def test_reused_instance_attributes(capsys: CaptureFixture) -> None: + class SubClass(DataService): + name = "Hello" + + subclass_instance = SubClass() + + class ServiceClass(DataService): + def __init__(self) -> None: + self.attr_1 = subclass_instance + self.attr_2 = subclass_instance + super().__init__() + + service_instance = ServiceClass() + _ = capsys.readouterr() + service_instance.attr_1.name = "Hi" + + captured = capsys.readouterr() + assert service_instance.attr_1 == service_instance.attr_2 + expected_output = sorted( + [ + "ServiceClass.attr_1.name = Hi", + "ServiceClass.attr_2.name = Hi", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + +def test_reused_attributes_mixed(capsys: CaptureFixture) -> None: + class SubClass(DataService): + pass + + subclass_instance = SubClass() + + class ServiceClass(DataService): + attr_1 = subclass_instance + + def __init__(self) -> None: + self.attr_2 = subclass_instance + super().__init__() + + service_instance = ServiceClass() + _ = capsys.readouterr() + service_instance.attr_1.name = "Hi" + + captured = capsys.readouterr() + assert service_instance.attr_1 == service_instance.attr_2 + expected_output = sorted( + [ + "ServiceClass.attr_1.name = Hi", + "ServiceClass.attr_2.name = Hi", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + +def test_nested_class_attributes(capsys: CaptureFixture) -> None: + class SubSubSubClass(DataService): + name = "Hello" + + class SubSubClass(DataService): + name = "Hello" + attr = SubSubSubClass() + + class SubClass(DataService): + name = "Hello" + attr = SubSubClass() + + class ServiceClass(DataService): + name = "Hello" + attr = SubClass() + + service_instance = ServiceClass() + _ = capsys.readouterr() + service_instance.attr.attr.attr.name = "Hi" + service_instance.attr.attr.name = "Hou" + service_instance.attr.name = "foo" + service_instance.name = "bar" + + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr.attr.attr.name = Hi", + "ServiceClass.attr.attr.name = Hou", + "ServiceClass.attr.name = foo", + "ServiceClass.name = bar", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + +def test_nested_instance_attributes(capsys: CaptureFixture) -> None: + class SubSubSubClass(DataService): + name = "Hello" + + class SubSubClass(DataService): + def __init__(self) -> None: + self.attr = SubSubSubClass() + self.name = "Hello" + super().__init__() + + class SubClass(DataService): + def __init__(self) -> None: + self.attr = SubSubClass() + self.name = "Hello" + super().__init__() + + class ServiceClass(DataService): + def __init__(self) -> None: + self.attr = SubClass() + self.name = "Hello" + super().__init__() + + service_instance = ServiceClass() + _ = capsys.readouterr() + service_instance.attr.attr.attr.name = "Hi" + service_instance.attr.attr.name = "Hou" + service_instance.attr.name = "foo" + service_instance.name = "bar" + + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr.attr.attr.name = Hi", + "ServiceClass.attr.attr.name = Hou", + "ServiceClass.attr.name = foo", + "ServiceClass.name = bar", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + +def test_advanced_nested_class_attributes(capsys: CaptureFixture) -> None: + class SubSubSubClass(DataService): + name = "Hello" + + class SubSubClass(DataService): + attr = SubSubSubClass() + + class SubClass(DataService): + attr = SubSubClass() + + class ServiceClass(DataService): + attr = SubClass() + subattr = SubSubClass() + + service_instance = ServiceClass() + _ = capsys.readouterr() + service_instance.attr.attr.attr.name = "Hi" + + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr.attr.attr.name = Hi", + "ServiceClass.subattr.attr.name = Hi", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + service_instance.subattr.attr.name = "Ho" + + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr.attr.attr.name = Ho", + "ServiceClass.subattr.attr.name = Ho", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + +def test_advanced_nested_instance_attributes(capsys: CaptureFixture) -> None: + class SubSubSubClass(DataService): + name = "Hello" + + class SubSubClass(DataService): + def __init__(self) -> None: + self.attr = SubSubSubClass() + super().__init__() + + subsubclass_instance = SubSubClass() + + class SubClass(DataService): + def __init__(self) -> None: + self.attr = subsubclass_instance + super().__init__() + + class ServiceClass(DataService): + def __init__(self) -> None: + self.attr = SubClass() + self.subattr = subsubclass_instance + super().__init__() + + service_instance = ServiceClass() + _ = capsys.readouterr() + service_instance.attr.attr.attr.name = "Hi" + + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr.attr.attr.name = Hi", + "ServiceClass.subattr.attr.name = Hi", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + service_instance.subattr.attr.name = "Ho" + + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr.attr.attr.name = Ho", + "ServiceClass.subattr.attr.name = Ho", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + +def test_advanced_nested_attributes_mixed(capsys: CaptureFixture) -> None: + class SubSubClass(DataService): + name = "Hello" + + class SubClass(DataService): + class_attr = SubSubClass() + + def __init__(self) -> None: + self.attr_1 = SubSubClass() + super().__init__() + + class ServiceClass(DataService): + class_attr = SubClass() + + def __init__(self) -> None: + self.attr = SubClass() + super().__init__() + + service_instance = ServiceClass() + # Subclass.attr is the same for all instances + assert service_instance.attr.class_attr == service_instance.class_attr.class_attr + + # attr_1 is different for all instances of SubClass + assert service_instance.attr.attr_1 != service_instance.class_attr.attr_1 + + # instances of SubSubClass are unequal + assert service_instance.attr.attr_1 != service_instance.class_attr.class_attr + + _ = capsys.readouterr() + + service_instance.class_attr.class_attr.name = "Ho" + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.class_attr.class_attr.name = Ho", + "ServiceClass.attr.class_attr.name = Ho", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + service_instance.class_attr.attr_1.name = "Ho" + captured = capsys.readouterr() + expected_output = sorted(["ServiceClass.class_attr.attr_1.name = Ho"]) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + service_instance.attr.class_attr.name = "Ho" + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr.class_attr.name = Ho", + "ServiceClass.class_attr.class_attr.name = Ho", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + service_instance.attr.attr_1.name = "Ho" + captured = capsys.readouterr() + expected_output = sorted(["ServiceClass.attr.attr_1.name = Ho"]) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + +def test_class_list_attributes(capsys: CaptureFixture) -> None: + class SubClass(DataService): + name = "Hello" + + subclass_instance = SubClass() + + class ServiceClass(DataService): + attr_list = [SubClass() for _ in range(2)] + attr_list_2 = [subclass_instance, subclass_instance] + attr = subclass_instance + + service_instance = ServiceClass() + _ = capsys.readouterr() + + assert service_instance.attr_list[0] != service_instance.attr_list[1] + + service_instance.attr_list[0].name = "Ho" + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr_list[0].name = Ho", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + service_instance.attr_list[1].name = "Ho" + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr_list[1].name = Ho", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + assert service_instance.attr_list_2[0] == service_instance.attr + assert service_instance.attr_list_2[0] == service_instance.attr_list_2[1] + + service_instance.attr_list_2[0].name = "Ho" + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr_list_2[0].name = Ho", + "ServiceClass.attr_list_2[1].name = Ho", + "ServiceClass.attr.name = Ho", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + service_instance.attr_list_2[1].name = "Ho" + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr_list_2[0].name = Ho", + "ServiceClass.attr_list_2[1].name = Ho", + "ServiceClass.attr.name = Ho", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + +def test_nested_class_list_attributes(capsys: CaptureFixture) -> None: + class SubSubClass(DataService): + name = "Hello" + + subsubclass_instance = SubSubClass() + + class SubClass(DataService): + attr_list = [subsubclass_instance] + + class ServiceClass(DataService): + attr = [SubClass()] + subattr = subsubclass_instance + + service_instance = ServiceClass() + _ = capsys.readouterr() + + assert service_instance.attr[0].attr_list[0] == service_instance.subattr + + service_instance.attr[0].attr_list[0].name = "Ho" + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr[0].attr_list[0].name = Ho", + "ServiceClass.subattr.name = Ho", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + service_instance.subattr.name = "Ho" + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr[0].attr_list[0].name = Ho", + "ServiceClass.subattr.name = Ho", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + +def test_instance_list_attributes(capsys: CaptureFixture) -> None: + class SubClass(DataService): + name = "Hello" + + subclass_instance = SubClass() + + class ServiceClass(DataService): + def __init__(self) -> None: + self.attr_list = [SubClass() for _ in range(2)] + self.attr_list_2 = [subclass_instance, subclass_instance] + self.attr = subclass_instance + super().__init__() + + service_instance = ServiceClass() + _ = capsys.readouterr() + + assert service_instance.attr_list[0] != service_instance.attr_list[1] + + service_instance.attr_list[0].name = "Ho" + captured = capsys.readouterr() + expected_output = sorted(["ServiceClass.attr_list[0].name = Ho"]) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + service_instance.attr_list[1].name = "Ho" + captured = capsys.readouterr() + expected_output = sorted(["ServiceClass.attr_list[1].name = Ho"]) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + assert service_instance.attr_list_2[0] == service_instance.attr + assert service_instance.attr_list_2[0] == service_instance.attr_list_2[1] + + service_instance.attr_list_2[0].name = "Ho" + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr.name = Ho", + "ServiceClass.attr_list_2[0].name = Ho", + "ServiceClass.attr_list_2[1].name = Ho", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + service_instance.attr_list_2[1].name = "Ho" + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr.name = Ho", + "ServiceClass.attr_list_2[0].name = Ho", + "ServiceClass.attr_list_2[1].name = Ho", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + service_instance.attr.name = "Ho" + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr.name = Ho", + "ServiceClass.attr_list_2[0].name = Ho", + "ServiceClass.attr_list_2[1].name = Ho", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + +def test_nested_instance_list_attributes(capsys: CaptureFixture) -> None: + class SubSubClass(DataService): + name = "Hello" + + subsubclass_instance = SubSubClass() + + class SubClass(DataService): + def __init__(self) -> None: + self.attr_list = [subsubclass_instance] + super().__init__() + + class ServiceClass(DataService): + class_attr = subsubclass_instance + + def __init__(self) -> None: + self.attr = [SubClass()] + super().__init__() + + service_instance = ServiceClass() + _ = capsys.readouterr() + + assert service_instance.attr[0].attr_list[0] == service_instance.class_attr + + service_instance.attr[0].attr_list[0].name = "Ho" + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr[0].attr_list[0].name = Ho", + "ServiceClass.class_attr.name = Ho", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output + + service_instance.class_attr.name = "Ho" + captured = capsys.readouterr() + expected_output = sorted( + [ + "ServiceClass.attr[0].attr_list[0].name = Ho", + "ServiceClass.class_attr.name = Ho", + ] + ) + actual_output = sorted(captured.out.strip().split("\n")) + assert actual_output == expected_output diff --git a/tests/test_full_access_path.py b/tests/test_full_access_path.py deleted file mode 100644 index fc22247..0000000 --- a/tests/test_full_access_path.py +++ /dev/null @@ -1,364 +0,0 @@ -from pyDataInterface import DataService - - -def test_class_attributes() -> None: - class SubClass(DataService): - pass - - class ServiceClass(DataService): - attr_1 = SubClass() - - test_service = ServiceClass() - assert test_service.attr_1._full_access_path == {"ServiceClass.attr_1"} - - -def test_instance_attributes() -> None: - class SubClass(DataService): - pass - - class ServiceClass(DataService): - def __init__(self): - self.attr_1 = SubClass() - super().__init__() - - test_service = ServiceClass() - assert test_service.attr_1._full_access_path == {"ServiceClass.attr_1"} - - -def test_reused_instance_attributes() -> None: - class SubClass(DataService): - pass - - subclass_instance = SubClass() - - class ServiceClass(DataService): - def __init__(self): - self.attr_1 = subclass_instance - self.attr_2 = subclass_instance - super().__init__() - - test_service = ServiceClass() - assert test_service.attr_1._full_access_path == { - "ServiceClass.attr_1", - "ServiceClass.attr_2", - } - assert test_service.attr_2._full_access_path == { - "ServiceClass.attr_1", - "ServiceClass.attr_2", - } - - assert test_service.attr_1._full_access_path == { - "ServiceClass.attr_1", - "ServiceClass.attr_2", - } - - -def test_reused_attributes_mixed() -> None: - class SubClass(DataService): - pass - - subclass_instance = SubClass() - - class ServiceClass(DataService): - attr_1 = subclass_instance - - def __init__(self): - self.attr_2 = subclass_instance - super().__init__() - - test_service = ServiceClass() - assert test_service.attr_1._full_access_path == { - "ServiceClass.attr_1", - "ServiceClass.attr_2", - } - assert test_service.attr_2._full_access_path == { - "ServiceClass.attr_1", - "ServiceClass.attr_2", - } - - -def test_nested_class_attributes() -> None: - class SubSubSubClass(DataService): - pass - - class SubSubClass(DataService): - attr = SubSubSubClass() - - class SubClass(DataService): - attr = SubSubClass() - - class ServiceClass(DataService): - attr = SubClass() - - test_service = ServiceClass() - assert test_service.attr._full_access_path == { - "ServiceClass.attr", - } - assert test_service.attr.attr._full_access_path == { - "ServiceClass.attr.attr", - } - assert test_service.attr.attr.attr._full_access_path == { - "ServiceClass.attr.attr.attr", - } - - -def test_nested_instance_attributes() -> None: - class SubSubSubClass(DataService): - pass - - class SubSubClass(DataService): - def __init__(self): - self.attr = SubSubSubClass() - super().__init__() - - class SubClass(DataService): - def __init__(self): - self.attr = SubSubClass() - super().__init__() - - class ServiceClass(DataService): - def __init__(self): - self.attr = SubClass() - super().__init__() - - test_service = ServiceClass() - assert test_service.attr._full_access_path == { - "ServiceClass.attr", - } - assert test_service.attr.attr._full_access_path == { - "ServiceClass.attr.attr", - } - assert test_service.attr.attr.attr._full_access_path == { - "ServiceClass.attr.attr.attr", - } - - -def test_advanced_nested_instance_attributes() -> None: - class SubSubSubClass(DataService): - pass - - class SubSubClass(DataService): - def __init__(self): - self.attr = SubSubSubClass() - super().__init__() - - subsubclass_instance = SubSubClass() - - class SubClass(DataService): - def __init__(self): - self.attr = subsubclass_instance - super().__init__() - - class ServiceClass(DataService): - def __init__(self): - self.attr = SubClass() - self.subattr = subsubclass_instance - super().__init__() - - test_service = ServiceClass() - assert test_service.attr._full_access_path == { - "ServiceClass.attr", - } - assert test_service.attr.attr._full_access_path == { - "ServiceClass.attr.attr", - "ServiceClass.subattr", - } - assert test_service.attr.attr.attr._full_access_path == { - "ServiceClass.attr.attr.attr", - "ServiceClass.subattr.attr", # as the SubSubSubClass does not implement anything, both subattr.attr and attr.attr.attr refer to the same instance - } - - -def test_advanced_nested_class_attributes() -> None: - class SubSubSubClass(DataService): - pass - - class SubSubClass(DataService): - attr = SubSubSubClass() - - class SubClass(DataService): - attr = SubSubClass() - - class ServiceClass(DataService): - attr = SubClass() - subattr = SubSubClass() - - test_service = ServiceClass() - assert test_service.attr._full_access_path == { - "ServiceClass.attr", - } - assert test_service.subattr._full_access_path == { - "ServiceClass.subattr", - } - assert test_service.attr.attr._full_access_path == { - "ServiceClass.attr.attr", - } - assert test_service.attr.attr.attr._full_access_path == { - "ServiceClass.attr.attr.attr", - "ServiceClass.subattr.attr", # as the SubSubSubClass does not implement anything, both subattr.attr and attr.attr.attr refer to the same instance - } - - -def test_advanced_nested_attributes_mixed() -> None: - class SubSubClass(DataService): - pass - - class SubClass(DataService): - attr = SubSubClass() - - def __init__(self): - self.attr_1 = SubSubClass() - super().__init__() - - class ServiceClass(DataService): - subattr = SubClass() - - def __init__(self): - self.attr = SubClass() - super().__init__() - - test_service = ServiceClass() - assert test_service.attr._full_access_path == { - "ServiceClass.attr", - } - assert test_service.subattr._full_access_path == { - "ServiceClass.subattr", - } - - # Subclass.attr is the same for all instances - assert test_service.attr.attr == test_service.subattr.attr - assert test_service.attr.attr._full_access_path == { - "ServiceClass.attr.attr", - "ServiceClass.subattr.attr", - } - assert test_service.subattr.attr._full_access_path == { - "ServiceClass.subattr.attr", - "ServiceClass.attr.attr", - } - - # attr_1 is different for all instances of SubClass - assert test_service.attr.attr_1 != test_service.subattr.attr - assert test_service.attr.attr_1 != test_service.subattr.attr_1 - assert test_service.subattr.attr_1._full_access_path == { - "ServiceClass.subattr.attr_1", - } - assert test_service.attr.attr_1._full_access_path == { - "ServiceClass.attr.attr_1", - } - - -def test_class_list_attributes() -> None: - class SubClass(DataService): - pass - - subclass_instance = SubClass() - - class ServiceClass(DataService): - attr_list = [SubClass() for _ in range(2)] - attr_list_2 = [subclass_instance, subclass_instance] - attr = subclass_instance - - test_service = ServiceClass() - assert test_service.attr_list[0] != test_service.attr_list[1] - assert test_service.attr_list[0]._full_access_path == { - "ServiceClass.attr_list[0]", - } - assert test_service.attr_list[1]._full_access_path == { - "ServiceClass.attr_list[1]", - } - - assert test_service.attr_list_2[0] == test_service.attr - assert test_service.attr_list_2[0] == test_service.attr_list_2[1] - assert test_service.attr_list_2[0]._full_access_path == { - "ServiceClass.attr", - "ServiceClass.attr_list_2[0]", - "ServiceClass.attr_list_2[1]", - } - assert test_service.attr_list_2[1]._full_access_path == { - "ServiceClass.attr", - "ServiceClass.attr_list_2[0]", - "ServiceClass.attr_list_2[1]", - } - - -def test_nested_class_list_attributes() -> None: - class SubSubClass(DataService): - pass - - subsubclass_instance = SubSubClass() - - class SubClass(DataService): - attr_list = [subsubclass_instance] - - class ServiceClass(DataService): - attr = [SubClass()] - subattr = subsubclass_instance - - test_service = ServiceClass() - assert test_service.attr[0].attr_list[0] == test_service.subattr - assert test_service.attr[0].attr_list[0]._full_access_path == { - "ServiceClass.attr[0].attr_list[0]", - "ServiceClass.subattr", - } - - -def test_instance_list_attributes() -> None: - class SubClass(DataService): - pass - - subclass_instance = SubClass() - - class ServiceClass(DataService): - def __init__(self): - self.attr_list = [SubClass() for _ in range(2)] - self.attr_list_2 = [subclass_instance, subclass_instance] - self.attr = subclass_instance - super().__init__() - - test_service = ServiceClass() - assert test_service.attr_list[0] != test_service.attr_list[1] - assert test_service.attr_list[0]._full_access_path == { - "ServiceClass.attr_list[0]", - } - assert test_service.attr_list[1]._full_access_path == { - "ServiceClass.attr_list[1]", - } - - assert test_service.attr_list_2[0] == test_service.attr - assert test_service.attr_list_2[0] == test_service.attr_list_2[1] - assert test_service.attr_list_2[0]._full_access_path == { - "ServiceClass.attr", - "ServiceClass.attr_list_2[0]", - "ServiceClass.attr_list_2[1]", - } - assert test_service.attr_list_2[1]._full_access_path == { - "ServiceClass.attr", - "ServiceClass.attr_list_2[0]", - "ServiceClass.attr_list_2[1]", - } - - -def test_nested_instance_list_attributes() -> None: - class SubSubClass(DataService): - pass - - subsubclass_instance = SubSubClass() - - class SubClass(DataService): - def __init__(self): - self.attr_list = [subsubclass_instance] - super().__init__() - - class ServiceClass(DataService): - subattr = subsubclass_instance - - def __init__(self): - self.attr = [SubClass()] - super().__init__() - - test_service = ServiceClass() - assert test_service.attr[0].attr_list[0] == test_service.subattr - assert test_service.attr[0].attr_list[0]._full_access_path == { - "ServiceClass.attr[0].attr_list[0]", - "ServiceClass.subattr", - }