taskiq/tests/formatters/test_json_formatter.py at 8f97d38c0f4568292fbc039f409ad01664640343 · 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
45
46
47
48
49
50
51
52
53
54
55
import json

import pytest

from taskiq.formatters.json_formatter import JSONFormatter
from taskiq.message import BrokerMessage, TaskiqMessage


@pytest.mark.anyio
async def test_json_dumps() -> None:
    fmt = JSONFormatter()
    msg = TaskiqMessage(
        task_id="task-id",
        task_name="task.name",
        queue="taksiq",
        labels={"label1": 1, "label2": "text"},
        args=[1, "a"],
        kwargs={"p1": "v1"},
    )
    expected = BrokerMessage(
        task_id="task-id",
        task_name="task.name",
        queue="taksiq",
        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"},
    )
    dumped = fmt.dumps(msg)
    assert dumped.task_id == expected.task_id
    assert dumped.task_name == expected.task_name
    assert dumped.labels == expected.labels
    assert json.loads(dumped.message) == json.loads(expected.message)


@pytest.mark.anyio
async def test_json_loads() -> None:
    fmt = JSONFormatter()
    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"},
        queue="taksiq",
        args=[1, "a"],
        kwargs={"p1": "v1"},
    )
    assert fmt.loads(msg) == expected