mirror of
https://github.com/tiqi-group/pydase.git
synced 2025-04-20 00:10:03 +02:00
149 lines
4.2 KiB
Python
149 lines
4.2 KiB
Python
import logging
|
|
from typing import Any
|
|
|
|
import pydase
|
|
import pytest
|
|
from pydase.data_service.data_service_observer import DataServiceObserver
|
|
from pydase.data_service.state_manager import StateManager
|
|
from pydase.utils.serialization.serializer import SerializationError
|
|
|
|
logger = logging.getLogger()
|
|
|
|
|
|
def test_static_property_dependencies() -> None:
|
|
class SubClass(pydase.DataService):
|
|
_name = "SubClass"
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self._name
|
|
|
|
@name.setter
|
|
def name(self, value: str) -> None:
|
|
self._name = value
|
|
|
|
class ServiceClass(pydase.DataService):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.list_attr = [SubClass()]
|
|
self._name = "ServiceClass"
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self._name
|
|
|
|
@name.setter
|
|
def name(self, value: str) -> None:
|
|
self._name = value
|
|
|
|
service_instance = ServiceClass()
|
|
state_manager = StateManager(service_instance)
|
|
observer = DataServiceObserver(state_manager)
|
|
logger.debug(observer.property_deps_dict)
|
|
assert observer.property_deps_dict == {
|
|
"list_attr[0]._name": ["list_attr[0].name"],
|
|
"_name": ["name"],
|
|
}
|
|
|
|
|
|
def test_dynamic_list_property_dependencies() -> None:
|
|
class SubClass(pydase.DataService):
|
|
_name = "SubClass"
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self._name
|
|
|
|
@name.setter
|
|
def name(self, value: str) -> None:
|
|
self._name = value
|
|
|
|
class ServiceClass(pydase.DataService):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.list_attr = [SubClass()]
|
|
|
|
service_instance = ServiceClass()
|
|
state_manager = StateManager(service_instance)
|
|
observer = DataServiceObserver(state_manager)
|
|
|
|
assert observer.property_deps_dict == {
|
|
"list_attr[0]._name": ["list_attr[0].name"],
|
|
}
|
|
|
|
service_instance.list_attr.append(SubClass())
|
|
|
|
assert observer.property_deps_dict == {
|
|
"list_attr[0]._name": ["list_attr[0].name"],
|
|
"list_attr[1]._name": ["list_attr[1].name"],
|
|
}
|
|
|
|
|
|
def test_protected_or_private_change_logs(caplog: pytest.LogCaptureFixture) -> None:
|
|
class OtherService(pydase.DataService):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self._name = "Hi"
|
|
|
|
class MyService(pydase.DataService):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.subclass = OtherService()
|
|
|
|
service = MyService()
|
|
state_manager = StateManager(service)
|
|
DataServiceObserver(state_manager)
|
|
|
|
service.subclass._name = "Hello"
|
|
assert "'subclass._name' changed to 'Hello'" not in caplog.text
|
|
|
|
|
|
def test_dynamic_list_entry_with_property(caplog: pytest.LogCaptureFixture) -> None:
|
|
class PropertyClass(pydase.DataService):
|
|
_name = "Hello"
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
"""The name property."""
|
|
return self._name
|
|
|
|
class MyService(pydase.DataService):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.list_attr = []
|
|
|
|
def toggle_high_voltage(self) -> None:
|
|
self.list_attr = []
|
|
self.list_attr.append(PropertyClass())
|
|
self.list_attr[0]._name = "Hoooo"
|
|
|
|
service = MyService()
|
|
state_manager = StateManager(service)
|
|
DataServiceObserver(state_manager)
|
|
service.toggle_high_voltage()
|
|
|
|
assert "'list_attr[0].name' changed to 'Hello'" not in caplog.text
|
|
assert "'list_attr[0].name' changed to 'Hoooo'" in caplog.text
|
|
|
|
|
|
def test_private_attribute_does_not_have_to_be_serializable() -> None:
|
|
class MyService(pydase.DataService):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.publ_attr: Any = 1
|
|
self.__priv_attr = (1,)
|
|
|
|
def change_publ_attr(self) -> None:
|
|
self.publ_attr = (2,) # cannot be serialized
|
|
|
|
def change_priv_attr(self) -> None:
|
|
self.__priv_attr = (2,)
|
|
|
|
service_instance = MyService()
|
|
pydase.Server(service_instance)
|
|
|
|
with pytest.raises(SerializationError):
|
|
service_instance.change_publ_attr()
|
|
|
|
service_instance.change_priv_attr()
|