69 lines
1.5 KiB
Python
69 lines
1.5 KiB
Python
|
"""Tests for background task queue."""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from fusionagi.api.task_queue import BackgroundTaskQueue, TaskStatus
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def queue():
|
||
|
|
return BackgroundTaskQueue(max_concurrent=3)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_submit_and_complete(queue):
|
||
|
|
async def work():
|
||
|
|
await asyncio.sleep(0.01)
|
||
|
|
return 42
|
||
|
|
|
||
|
|
tid = queue.submit(work)
|
||
|
|
await asyncio.sleep(0.05)
|
||
|
|
result = queue.get_status(tid)
|
||
|
|
assert result is not None
|
||
|
|
assert result.status == TaskStatus.COMPLETED
|
||
|
|
assert result.result == 42
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_failed_task(queue):
|
||
|
|
async def fail():
|
||
|
|
raise ValueError("boom")
|
||
|
|
|
||
|
|
tid = queue.submit(fail)
|
||
|
|
await asyncio.sleep(0.05)
|
||
|
|
result = queue.get_status(tid)
|
||
|
|
assert result is not None
|
||
|
|
assert result.status == TaskStatus.FAILED
|
||
|
|
assert "boom" in (result.error or "")
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_list_tasks(queue):
|
||
|
|
async def noop():
|
||
|
|
pass
|
||
|
|
|
||
|
|
queue.submit(noop)
|
||
|
|
queue.submit(noop)
|
||
|
|
await asyncio.sleep(0.05)
|
||
|
|
tasks = queue.list_tasks()
|
||
|
|
assert len(tasks) == 2
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_list_tasks_filtered(queue):
|
||
|
|
async def noop():
|
||
|
|
pass
|
||
|
|
|
||
|
|
queue.submit(noop)
|
||
|
|
await asyncio.sleep(0.05)
|
||
|
|
completed = queue.list_tasks(status=TaskStatus.COMPLETED)
|
||
|
|
assert len(completed) == 1
|
||
|
|
pending = queue.list_tasks(status=TaskStatus.PENDING)
|
||
|
|
assert len(pending) == 0
|
||
|
|
|
||
|
|
|
||
|
|
def test_nonexistent_task(queue):
|
||
|
|
assert queue.get_status("nonexistent") is None
|