ITQC/pydase
24
0
Fork 0
mirror of https://github.com/tiqi-group/pydase.git synced 2025-06-12 15:57:12 +02:00
Code Issues Packages Projects Releases Wiki Activity

updates tests

Browse Source
This commit is contained in:
Mose Müller
2023-12-04 17:23:42 +01:00
parent 0aa1595da4
commit 89b5a9cc9e
5 changed files with 321 additions and 324 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
tests/data_service/test_task_manager.py
View File

@ -1,6 +1,8 @@
import logging
import pydase
from pydase.data_service.data_service_observer import DataServiceObserver
from pydase.data_service.state_manager import StateManager
from pytest import LogCaptureFixture
logger = logging.getLogger()
@ -21,8 +23,10 @@ def test_autostart_task_callback(caplog: LogCaptureFixture) -> None:
async def my_other_task(self) -> None:
logger.info("Triggered other task.")
service = MyService()
service._task_manager.start_autostart_tasks()
service_instance = MyService()
state_manager = StateManager(service_instance)
DataServiceObserver(state_manager)
service_instance._task_manager.start_autostart_tasks()
assert "'my_task' changed to '{}'" in caplog.text
assert "'my_other_task' changed to '{}'" in caplog.text
@ -48,14 +52,16 @@ def test_DataService_subclass_autostart_task_callback(
class MyService(pydase.DataService):
sub_service = MySubService()
service = MyService()
service._task_manager.start_autostart_tasks()
service_instance = MyService()
state_manager = StateManager(service_instance)
DataServiceObserver(state_manager)
service_instance._task_manager.start_autostart_tasks()
assert "'sub_service.my_task' changed to '{}'" in caplog.text
assert "'sub_service.my_other_task' changed to '{}'" in caplog.text
def test_DataServiceList_subclass_autostart_task_callback(
def test_DataService_subclass_list_autostart_task_callback(
caplog: LogCaptureFixture,
) -> None:
class MySubService(pydase.DataService):
@ -75,8 +81,10 @@ def test_DataServiceList_subclass_autostart_task_callback(
class MyService(pydase.DataService):
sub_services_list = [MySubService() for i in range(2)]
service = MyService()
service._task_manager.start_autostart_tasks()
service_instance = MyService()
state_manager = StateManager(service_instance)
DataServiceObserver(state_manager)
service_instance._task_manager.start_autostart_tasks()
assert "'sub_services_list[0].my_task' changed to '{}'" in caplog.text
assert "'sub_services_list[0].my_other_task' changed to '{}'" in caplog.text