taskiq/tests/formatters/test_proxy_formatter.py at a6894fd7215ee60d7ed646ecbd1d32b282726884 · taskiq-python/taskiq · GitHub

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
from taskiq.brokers.inmemory_broker import InMemoryBroker
from taskiq.message import BrokerMessage, TaskiqMessage


async def test_proxy_dumps() -> None:
    # uses json serializer by default
    broker = InMemoryBroker()
    msg = TaskiqMessage(
        task_id="task-id",
        task_name="task.name",
        labels={"label1": 1, "label2": "text"},
        args=[1, "a"],
        kwargs={"p1": "v1"},
    )
    expected = BrokerMessage(
        task_id="task-id",
        task_name="task.name",
        message=(
            b'{"task_id": "task-id", "task_name": "task.name", '
            b'"labels": {"label1": 1, "label2": "text"}, '
            b'"labels_types": null, '
            b'"args": [1, "a"], "kwargs": {"p1": "v1"}}'
        ),
        labels={"label1": 1, "label2": "text"},
    )
    assert broker.formatter.dumps(msg) == expected


async def test_proxy_loads() -> None:
    # uses json serializer by default
    broker = InMemoryBroker()
    msg = (
        b'{"task_id":"task-id","task_name":"task.name",'
        b'"labels":{"label1":1,"label2":"text"},'
        b'"args":[1,"a"],"kwargs":{"p1":"v1"}}'
    )
    expected = TaskiqMessage(
        task_id="task-id",
        task_name="task.name",
        labels={"label1": 1, "label2": "text"},
        args=[1, "a"],
        kwargs={"p1": "v1"},
    )
    assert broker.formatter.loads(msg) == expected