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
|
"""
`chat --follow`: new messages as the node pushes them, a dead connection opened
again, and what was said meanwhile read from the history — each message once.
"""
import asyncio
import json
import pytest
from fake_hub import PASSPHRASE, USERNAME, FakeHub
from fake_node import FakeNode
from meshbay_cli import __main__ as cli
from meshbay_cli.commands import chat, context
@pytest.fixture
def follow(monkeypatch, capsys):
hub, node = FakeHub(), FakeNode()
monkeypatch.setattr(context, "Hub", lambda url: hub)
sessions = []
async def open_group(account, group):
sessions.append(node.session())
return sessions[-1]
monkeypatch.setattr(chat, "open_group", open_group)
monkeypatch.setattr(chat, "PING_EVERY", 0.05)
monkeypatch.setattr(chat, "RECONNECT_WAITS", (0.01,))
assert cli.main(["--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE,
"login"]) == 0
capsys.readouterr()
def start(*argv):
args = cli.build_parser().parse_args(cli.chat.rewrite_count(list(argv), cli.ACTIONS))
return asyncio.ensure_future(chat.run(context.Context(args)))
return node, sessions, start
def _push(session, node):
session.ch._emit({**node.chat[-1], "type": "chat_msg"})
async def _until(cond, timeout=5.0):
deadline = asyncio.get_running_loop().time() + timeout
while not cond():
assert asyncio.get_running_loop().time() < deadline, "timed out"
await asyncio.sleep(0.02)
async def test_follow_prints_pushes_reconnects_and_never_repeats(follow, capsys):
node, sessions, start = follow
node.say("before")
task = start("--json", "chat", "-1", "Family", "--follow")
printed = []
def lines():
printed.extend(json.loads(x)["text"] for x in capsys.readouterr().out.splitlines())
return printed
try:
await _until(lambda: lines() == ["before"])
node.say("live one")
_push(sessions[-1], node)
await _until(lambda: lines()[-1:] == ["live one"])
# The connection drops; meanwhile two messages are stored and only the
# last of them is pushed, once the new connection is up.
sessions[-1].ch.closed.set()
node.say("while away")
await _until(lambda: len(sessions) == 2)
await _until(lambda: "while away" in lines())
node.say("after")
_push(sessions[-1], node)
_push(sessions[-1], node) # the node said it twice
await _until(lambda: lines()[-1:] == ["after"])
# A connection that stops answering its ping is replaced too.
node.unresponsive = True
await _until(lambda: len(sessions) == 3)
node.unresponsive = False
node.say("on the third connection")
_push(sessions[-1], node)
await _until(lambda: lines()[-1:] == ["on the third connection"])
await asyncio.sleep(0.2)
assert lines() == ["before", "live one", "while away", "after",
"on the third connection"]
finally:
task.cancel()
with pytest.raises(asyncio.CancelledError):
await task
def test_follow_and_post_are_not_asked_at_once(follow, capsys):
assert cli.main(["chat", "Family", "-f", "-m", "x"]) == 1
assert cli.main(["chat", "-f"]) == 1
|