mirror of
https://github.com/tiqi-group/pydase.git
synced 2025-06-03 20:30:40 +02:00
adds test for starting and stopping tasks
This commit is contained in:
parent
ca41e12014
commit
2461f85ef0
@ -1,3 +1,4 @@
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
import pydase
|
||||
@ -95,3 +96,30 @@ async def test_DataService_subclass_list_autostart_task_callback(
|
||||
assert "'sub_services_list[0].my_other_task' changed to '{}'" in caplog.text
|
||||
assert "'sub_services_list[1].my_task' changed to '{}'" in caplog.text
|
||||
assert "'sub_services_list[1].my_other_task' changed to '{}'" in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_start_and_stop_task_methods(caplog: LogCaptureFixture) -> None:
|
||||
class MyService(pydase.DataService):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
|
||||
async def my_task(self, param: str) -> None:
|
||||
while True:
|
||||
logger.debug("Logging param: %s", param)
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
# Your test code here
|
||||
service_instance = MyService()
|
||||
state_manager = StateManager(service_instance)
|
||||
DataServiceObserver(state_manager)
|
||||
service_instance.start_my_task("Hello")
|
||||
await asyncio.sleep(0.01)
|
||||
|
||||
assert "'my_task' changed to '{'param': 'Hello'}'" in caplog.text
|
||||
assert "Logging param: Hello" in caplog.text
|
||||
caplog.clear()
|
||||
|
||||
service_instance.stop_my_task()
|
||||
await asyncio.sleep(0.01)
|
||||
assert "Task 'my_task' was cancelled" in caplog.text
|
||||
|
Loading…
x
Reference in New Issue
Block a user