-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathtest_subscriptions.py
More file actions
136 lines (106 loc) · 4.48 KB
/
Copy pathtest_subscriptions.py
File metadata and controls
136 lines (106 loc) · 4.48 KB
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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
import asyncio
import uuid
import pytest
from natsrpy import CallbackSubscription, IteratorSubscription, Message, Nats
async def test_subscribe_returns_iterator(nats: Nats) -> None:
subj = uuid.uuid4().hex
async with nats.subscribe(subject=subj) as sub:
assert isinstance(sub, IteratorSubscription)
async def test_subscribe_with_callback(nats: Nats) -> None:
subj = uuid.uuid4().hex
received: list[bytes] = []
event = asyncio.Event()
async def callback(msg: Message) -> None:
received.append(msg.payload)
event.set()
async with nats.subscribe(subject=subj, callback=callback) as sub:
assert isinstance(sub, CallbackSubscription)
await nats.publish(subj, b"callback-test")
await asyncio.wait_for(event.wait(), timeout=5.0)
assert received == [b"callback-test"]
async def test_iterator_next_with_timeout(nats: Nats) -> None:
subj = uuid.uuid4().hex
async with nats.subscribe(subject=subj) as sub:
await nats.publish(subj, b"timeout-test")
message = await asyncio.wait_for(anext(sub), timeout=5.0)
assert message.payload == b"timeout-test"
async def test_iterator_aiter_protocol(nats: Nats) -> None:
subj = uuid.uuid4().hex
async with nats.subscribe(subject=subj) as sub:
payloads = [f"msg-{i}".encode() for i in range(3)]
for p in payloads:
await nats.publish(subj, p)
received = []
async for msg in sub:
received.append(msg.payload)
if len(received) == 3:
break
assert received == payloads
async def test_iterator_unsubscribe(nats: Nats) -> None:
subj = uuid.uuid4().hex
async with nats.subscribe(subject=subj) as sub:
await sub.unsubscribe()
async def test_iterator_unsubscribe_with_limit(nats: Nats) -> None:
subj = uuid.uuid4().hex
async with nats.subscribe(subject=subj) as sub:
await sub.unsubscribe(limit=2)
await nats.publish(subj, b"msg-1")
await nats.publish(subj, b"msg-2")
msg1 = await asyncio.wait_for(anext(sub), timeout=5.0)
msg2 = await asyncio.wait_for(anext(sub), timeout=5.0)
assert msg1.payload == b"msg-1"
assert msg2.payload == b"msg-2"
with pytest.raises(StopAsyncIteration):
await asyncio.wait_for(anext(sub), timeout=5.0)
async def test_iterator_drain(nats_url: str) -> None:
client = Nats(addrs=[nats_url])
await client.startup()
subj = uuid.uuid4().hex
async with client.subscribe(subject=subj) as sub:
await sub.drain()
await client.shutdown()
async def test_callback_receives_message(nats: Nats) -> None:
subj = uuid.uuid4().hex
event = asyncio.Event()
async def callback(_: Message) -> None:
event.set()
async with nats.subscribe(subject=subj, callback=callback) as sub:
assert isinstance(sub, CallbackSubscription)
await nats.publish(subj, b"trigger")
await asyncio.wait_for(event.wait(), timeout=5.0)
async def test_multiple_subscribers(nats: Nats) -> None:
subj = uuid.uuid4().hex
async with (
nats.subscribe(subject=subj) as sub1,
nats.subscribe(subject=subj) as sub2,
):
await nats.publish(subj, b"multi-sub")
msg1 = await anext(sub1)
msg2 = await anext(sub2)
assert msg1.payload == b"multi-sub"
assert msg2.payload == b"multi-sub"
async def test_wildcard_subscription(nats: Nats) -> None:
prefix = uuid.uuid4().hex
async with nats.subscribe(subject=f"{prefix}.*") as sub:
await nats.publish(f"{prefix}.one", b"wildcard-1")
await nats.publish(f"{prefix}.two", b"wildcard-2")
msg1 = await anext(sub)
msg2 = await anext(sub)
assert msg1.payload == b"wildcard-1"
assert msg2.payload == b"wildcard-2"
async def test_fullwild_subscription(nats: Nats) -> None:
prefix = uuid.uuid4().hex
async with nats.subscribe(subject=f"{prefix}.>") as sub:
await nats.publish(f"{prefix}.a.b.c", b"full-wild")
msg = await anext(sub)
assert msg.payload == b"full-wild"
assert msg.subject == f"{prefix}.a.b.c"
async def test_subscription_ctx_manager_detatch(nats: Nats) -> None:
subj = uuid.uuid4().hex
ctx = nats.subscribe(subject=subj)
sub = await ctx.detatch()
assert isinstance(sub, IteratorSubscription)
await nats.publish(subj, b"detatch-test")
msg = await asyncio.wait_for(anext(sub), timeout=5.0)
assert msg.payload == b"detatch-test"
await sub.unsubscribe()