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

moving from _full_access_name to callbacks

Browse Source
This commit is contained in:
Mose Müller 2023-08-02 12:06:19 +02:00
parent b67c0f9da3
commit 460be17ecb
5 changed files with 671 additions and 488 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

163
src/pyDataInterface/data_service/data_service.py
View File

@ -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("_"):

13
tests/__init__.py
View File

@ -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

51
tests/test_emit_on_change.py → tests/test_DataServiceList.py
View File

@ -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"))

568
tests/test_DataService_attribute_callbacks.py Normal file
View File

@ -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

364
tests/test_full_access_path.py
View File

@ -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",
}