Tribler/py-ipv8

Add a helper function to `TestBase` to await registered tasks

Closed this issue · 1 comments

While writing tests, I often use the following pattern to await a registered task:

async def test_something(self) -> None:
    task_manager.user_logic_registers_tasks()
    
    await sleep(0)
    for task in task_manager.get_tasks():
        if task.get_name() == "task name":
            await task

    # assert something ...

Ideally, I would not like to repeat myself this much. Instead, I'd like TestBase to offer this functionality. Something like this:

async def test_something(self) -> None:
    task_manager.user_logic_registers_tasks()

    await self.tm_task("task name", task_manager)

    # assert something ...

I'll update this issue with a superior suggestion by @egbertbouman. The TaskManager can support a get_task(task_name: str) -> Future.

Perhaps we can also add get_anonymous_tasks(base_name: str) -> list[Future].

This would change the envisioned functionality to be the following:

async def test_something(self) -> None:
    task_manager.user_logic_registers_tasks()
    await sleep(0)

    await task_manager.get_task("task name")

    # assert something ...