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

tasks: return result of task after finishing gracefully

Browse Source 
Tasks that finished gracefully were restarted again. This fixes that.
This commit is contained in:
Mose Müller
2025-01-20 06:10:48 +01:00
parent aacc69ae94
commit 053050a62c
2 changed files with 22 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
tests/task/test_task.py
View File

@ -436,3 +436,24 @@ async def test_exit_on_failure_exceeding_rate_limit(
await asyncio.sleep(0.5)
assert "os.kill called with signal=" in caplog.text
assert "Task 'my_task' encountered an exception" in caplog.text
@pytest.mark.asyncio(scope="function")
async def test_gracefully_finishing_task(
monkeypatch: pytest.MonkeyPatch, caplog: LogCaptureFixture
) -> None:
class MyService(pydase.DataService):
@task()
async def my_task(self) -> None:
print("Hello")
await asyncio.sleep(0.1)
service_instance = MyService()
state_manager = StateManager(service_instance)
DataServiceObserver(state_manager)
service_instance.my_task.start()
await asyncio.sleep(0.05)
assert service_instance.my_task.status == TaskStatus.RUNNING
await asyncio.sleep(0.1)
assert service_instance.my_task.status == TaskStatus.NOT_RUNNING