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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
|
import time
from copy import copy
from datetime import datetime, timezone
from typing import List, Tuple
import shortuuid
from aio_pika import DeliveryMode, Message
from aio_pika.abc import FieldValue, HeadersType, MessageInfo
def test_message_copy():
msg1 = Message(
bytes(shortuuid.uuid(), "utf-8"),
content_type="application/json",
content_encoding="text",
timestamp=datetime(2000, 1, 1),
headers={"h1": "v1", "h2": "v2"},
)
msg2 = copy(msg1)
msg1.lock()
assert not msg2.locked
def test_message_info():
body = bytes(shortuuid.uuid(), "utf-8")
info = MessageInfo(
app_id="test",
body_size=len(body),
cluster_id=None,
consumer_tag=None,
content_encoding="text",
content_type="application/json",
correlation_id="1",
delivery_mode=DeliveryMode.PERSISTENT,
delivery_tag=None,
exchange=None,
expiration=1.5,
headers={"foo": "bar"},
message_id=shortuuid.uuid(),
priority=0,
redelivered=None,
reply_to="test",
routing_key=None,
timestamp=datetime.fromtimestamp(int(time.time()), tz=timezone.utc),
type="0",
user_id="guest",
)
msg = Message(
body=body,
headers={"foo": "bar"},
content_type="application/json",
content_encoding="text",
delivery_mode=DeliveryMode.PERSISTENT,
priority=0,
correlation_id="1",
reply_to="test",
expiration=1.5,
message_id=info["message_id"],
timestamp=info["timestamp"],
type="0",
user_id="guest",
app_id="test",
)
assert info == msg.info()
def test_headers_setter():
data: HeadersType = {"foo": "bar"}
data_expected = {"foo": "bar"}
msg = Message(b"", headers={"bar": "baz"})
msg.headers = data
assert msg.headers == data_expected
def test_headers_content():
data: Tuple[List[FieldValue], ...] = (
[42, 42],
[b"foo", b"foo"],
[b"\00", b"\00"],
)
for src, value in data:
msg = Message(b"", headers={"value": src})
assert msg.headers["value"] == value
def test_headers_set():
msg = Message(b"", headers={"header": "value"})
data = (
["header-1", 42, 42],
["header-2", b"foo", b"foo"],
["header-3", b"\00", b"\00"],
["header-4", {"foo": "bar"}, {"foo": "bar"}],
)
for name, src, value in data: # type: ignore
msg.headers[name] = value # type: ignore
assert msg.headers[name] == value # type: ignore
assert msg.headers["header"] == "value"
|
