tests/test_background_tasks.py

import asyncio
import unittest

from api import background_tasks


class BackgroundTaskSupervisorTests(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        await background_tasks.shutdown_background_tasks()

    async def asyncTearDown(self):
        await background_tasks.shutdown_background_tasks()

    async def test_duplicate_name_reuses_task_and_closes_duplicate_coroutine(self):
        started = asyncio.Event()
        release = asyncio.Event()

        async def worker():
            started.set()
            await release.wait()

        first = background_tasks.spawn_background_task(worker(), name="duplicate")
        await started.wait()
        duplicate = background_tasks.spawn_background_task(worker(), name="duplicate")
        self.assertIs(first, duplicate)
        release.set()
        await background_tasks.shutdown_background_tasks()
        self.assertTrue(first.done())

    async def test_shutdown_cancels_and_awaits_running_task(self):
        cancelled = asyncio.Event()

        async def worker():
            try:
                await asyncio.Event().wait()
            except asyncio.CancelledError:
                cancelled.set()
                raise

        background_tasks.spawn_background_task(worker(), name="cancellable")
        await asyncio.sleep(0)
        await background_tasks.shutdown_background_tasks()
        self.assertTrue(cancelled.is_set())
        self.assertFalse(background_tasks._tasks)

    async def test_completed_task_is_not_left_in_registry_after_shutdown(self):
        async def worker():
            return "ok"

        task = background_tasks.spawn_background_task(worker(), name="completed")
        await task
        await background_tasks.shutdown_background_tasks()
        self.assertFalse(background_tasks._tasks)


if __name__ == "__main__":
    unittest.main()