aboutsummaryrefslogtreecommitdiffstats
path: root/tests/test_task_runner.py
blob: 2aaa9f914d67b01be00560917b1c2b64baf5d6c8 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
from collections import defaultdict

from slack.shared import shared
from slack.task import Future, create_task, weechat_task_cb


def test_run_single_task():
    shared.active_tasks = defaultdict(list)
    shared.active_futures = {}
    future = Future[str]()

    async def awaitable():
        result = await future
        return "awaitable", result

    task = create_task(awaitable())
    weechat_task_cb(future.id, "data")

    assert not shared.active_tasks
    assert not shared.active_futures
    assert task.result() == ("awaitable", ("data",))


def test_run_nested_task():
    shared.active_tasks = defaultdict(list)
    shared.active_futures = {}
    future = Future[str]()

    async def awaitable1():
        result = await future
        return "awaitable1", result

    async def awaitable2():
        result = await create_task(awaitable1())
        return "awaitable2", result

    task = create_task(awaitable2())
    weechat_task_cb(future.id, "data")

    assert not shared.active_tasks
    assert not shared.active_futures
    assert task.result() == ("awaitable2", ("awaitable1", ("data",)))


def test_run_two_tasks_concurrently():
    shared.active_tasks = defaultdict(list)
    shared.active_futures = {}
    future1 = Future[str]()
    future2 = Future[str]()

    async def awaitable(future: Future[str]):
        result = await future
        return "awaitable", result

    task1 = create_task(awaitable(future1))
    task2 = create_task(awaitable(future2))
    weechat_task_cb(future1.id, "data1")
    weechat_task_cb(future2.id, "data2")

    assert not shared.active_tasks
    assert not shared.active_futures
    assert task1.result() == ("awaitable", ("data1",))
    assert task2.result() == ("awaitable", ("data2",))


def test_task_without_await():
    shared.active_tasks = defaultdict(list)
    shared.active_futures = {}

    async def fun_without_await():
        pass

    async def run():
        await create_task(fun_without_await())

    create_task(run())

    assert not shared.active_tasks
    assert not shared.active_futures