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
137
138
|
"""
What `meshbay-node` prints and asks for, verb by verb, recorded once and held.
A characterisation test, like test_dispatch_golden.py: it states no rule of its
own, it pins the CLI as it is, so that `main()` can be taken apart underneath
it. Every verb test_cli_dispatch.py walks, the ones that ask for confirmation
without --yes, each verb with a subcommand it does not have, `--help` and a
verb that does not exist, is run with the daemon stubbed out, and what is
recorded is the exit code, stdout, stderr, each loopback call with its body,
each question asked and each `systemctl` run. The help text is in there whole:
the user guide quotes it.
The recording is `golden/cli.json`. When a change to this behaviour is
intended, regenerate it with MESHBAY_GOLDEN_WRITE=1 and read the diff.
"""
import builtins
import getpass
import json
import os
import subprocess
import sys
from pathlib import Path
import pytest
from meshbay_node import config as config_mod
from meshbay_node import daemon as daemon_mod
from test_cli_dispatch import VERBS
GOLDEN = Path(__file__).parent / "golden" / "cli.json"
# The verbs that ask first, asked and answered "no": test_cli_dispatch.py
# passes --yes to all of them.
UNCONFIRMED = [
["root", "remove", "media"],
["gek", "rotate"],
["file", "rm", "abc"],
["video", "rematch"],
["chat", "encrypt-history"],
["denylist", "clear"],
]
# Each verb's last resort, its own usage line.
UNKNOWN_SUB = [[verb, "no-such-sub"] for verb in (
"gek", "operator", "member", "group", "root", "file", "video", "chat",
"denylist", "stun", "transfers", "autostart", "service")]
CASES = [["--help"], ["no-such-verb"], *VERBS, *UNCONFIRMED, *UNKNOWN_SUB]
def _run(argv: list[str], tmp_path: Path, capsys) -> dict:
api: list = []
asked: list = []
systemctl: list = []
def fake_api(cfg, path, method="GET", timeout=30, body=None):
api.append([method, path, body])
return {
"groups": [], "files": [], "identities": [], "members": [],
"invites": [], "users": [], "jtis": [], "count": 0, "removed": 0,
"subject": "all", "code": "TEST-CODE", "expires_at": "",
"link": "https://example.invalid/#/invite?v=1", "invite_id": "ab" * 16,
"user_id": "u", "authorized_members": 0, "errors": [],
"name": "g", "group_id": "g", "shared_dir": str(tmp_path),
"config": str(tmp_path / "node.toml"),
"enabled": False, "epoch": 1, "converted": 0,
"backup": str(tmp_path / "chat.db.bak"),
"encrypted": False, "plaintext_messages": 0,
"encrypted_messages": 0, "max_age_days": 30,
}
def fake_input(prompt=""):
asked.append(prompt)
return "n"
def fake_run(args, **kw):
systemctl.append(args)
return subprocess.CompletedProcess(args, 0, stdout="", stderr="")
conf = tmp_path / "node.toml"
conf.write_text('[hub]\nurl = "https://example.invalid"\n')
code = 0
with pytest.MonkeyPatch.context() as mp:
mp.setattr(daemon_mod, "_daemon_api", fake_api)
mp.setattr(daemon_mod, "_resolve_group", lambda cfg, g: "g" * 32)
mp.setattr(daemon_mod, "DEFAULT_CONFIG_PATH", conf)
# The data directory and the keystore's default path come from these;
# left real, `status` would record the developer's own node.
mp.setenv("HOME", str(tmp_path / "home"))
mp.setattr(config_mod, "DEFAULT_CONFIG_PATH", conf)
mp.setattr(builtins, "input", fake_input)
mp.setattr(getpass, "getpass", lambda *a, **kw: "test-password")
mp.setattr(subprocess, "run", fake_run)
# argparse wraps the help to the terminal's width.
mp.setenv("COLUMNS", "100")
mp.setattr(sys, "argv", ["meshbay-node", *argv])
try:
daemon_mod.main()
except SystemExit as e:
code = e.code
out = capsys.readouterr()
def mask(text: str) -> str:
return text.replace(str(tmp_path), "<tmp>")
return json.loads(mask(json.dumps({
"exit": code, "stdout": out.out, "stderr": out.err,
"api": api, "asked": asked, "systemctl": systemctl})))
def _record(tmp_path: Path, capsys) -> dict:
return {" ".join(argv): _run(argv, tmp_path, capsys) for argv in CASES}
def test_every_verb_prints_as_recorded(tmp_path, capsys):
if os.environ.get("MESHBAY_GOLDEN_WRITE") == "1":
GOLDEN.write_text(json.dumps(_record(tmp_path, capsys), indent=1,
sort_keys=True, ensure_ascii=False) + "\n",
encoding="utf-8")
golden = json.loads(GOLDEN.read_text(encoding="utf-8"))
now = _record(tmp_path, capsys)
assert sorted(now) == sorted(golden), (
"the set of verbs changed; regenerate on purpose if it should")
differ = [k for k in golden if now[k] != golden[k]]
detail = "\n".join(
f" {k}\n" + "\n".join(f" {f}: was {golden[k][f]!r}\n {f}: now {now[k][f]!r}"
for f in golden[k] if golden[k][f] != now[k][f])
for k in differ[:5])
assert not differ, f"{len(differ)} verb(s) behave differently:\n{detail}"
def test_the_recording_exercises_the_cli():
"""A recording of nothing would hold nothing."""
golden = json.loads(GOLDEN.read_text(encoding="utf-8"))
assert "usage: meshbay-node" in golden["--help"]["stdout"]
assert golden["no-such-verb"]["exit"] == 2
assert sum(1 for c in golden.values() if c["api"]) > 30, "the daemon was never asked"
assert sum(1 for c in golden.values() if c["exit"] not in (0, None)) > 5, (
"no usage line was ever reached")
assert any(c["asked"] for c in golden.values()), "no confirmation was ever asked"
|