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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
|
"""
Every action, run through `main()` with the hub and the node stubbed.
A branch nobody has run is covered by nothing — the node's `reload` once shipped
with a missing import. `test_every_action_is_run_here` refuses an action added
to the parser without a run below.
"""
import io
import json
import pytest
from fake_hub import PASSPHRASE, USERNAME, FakeHub
from meshbay_cli import __main__ as cli
from meshbay_cli.client import pool
from meshbay_cli.commands import context
RUN: set[str] = set()
@pytest.fixture
def hub(monkeypatch):
fake = FakeHub()
monkeypatch.setattr(context, "Hub", lambda url: fake)
return fake
def main(*argv: str) -> int:
RUN.add(next(a for a in argv if a in cli.ACTIONS))
return cli.main(list(argv))
class FakeSession:
async def index(self):
return {"roots": [{"name": "share", "available": True, "writable": True}],
"dirs": ["share"],
"entries": [{"id": "a" * 64, "name": "notes.txt", "path": "share",
"size": 6, "type": "document", "added_at": 0}]}
async def close(self):
pass
def test_login_ls_logout(hub, monkeypatch, capsys):
monkeypatch.setattr(pool, "open_group", lambda account, group: _session())
assert main("--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE, "login") == 0
assert "Signed in: alice-test" in capsys.readouterr().out
assert main("ls") == 0
assert capsys.readouterr().out.strip() == f"Family@{USERNAME}"
assert main("ls", "--json", "Family/share") == 0
[item] = json.loads(capsys.readouterr().out)
assert item["path"] == f"Family@{USERNAME}/share/notes.txt" and item["kind"] == "file"
assert main("logout") == 0
assert "Signed out." in capsys.readouterr().out
assert not hub.devices
assert main("ls") == 2
assert "Not signed in" in capsys.readouterr().err
def test_login_without_remembering_stores_nothing(hub, isolated_home, capsys):
assert main("--hub", hub.url, "-u", USERNAME, "login", "--no-remember",
"--password", PASSPHRASE) == 0
assert "Credentials accepted" in capsys.readouterr().out
assert not hub.devices
assert not any(isolated_home.rglob("secrets.enc"))
def test_a_wrong_passphrase_exits_2(hub, capsys):
assert main("--hub", hub.url, "-u", USERNAME, "--password", "nope", "login") == 2
assert "wrong username or passphrase" in capsys.readouterr().err
def test_no_passphrase_and_no_terminal_exits_2(hub, monkeypatch, capsys):
monkeypatch.setattr("sys.stdin.isatty", lambda: False)
assert main("--hub", hub.url, "-u", USERNAME, "login") == 2
assert "MESHBAY_PASSWORD" in capsys.readouterr().err
async def _session():
return FakeSession()
@pytest.fixture
def node(hub, monkeypatch, capsys):
from fake_node import FakeNode
fake = FakeNode()
async def open_group(account, group):
return fake.session()
monkeypatch.setattr(pool, "open_group", open_group)
assert main("--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE, "login") == 0
capsys.readouterr()
return fake
def test_get_files_folders_and_patterns(node, tmp_path, monkeypatch, capsys):
node.add("share", "notes.txt", b"hello")
node.add("share/2024", "a.jpg", b"A" * 3000)
node.add("share/2024", "b.jpg", b"B" * 10)
node.add("share/2024/deep", "c.txt", b"C")
node.dirs.add("share/2024/empty")
monkeypatch.chdir(tmp_path)
assert main("get", "Family/share/notes.txt") == 0
assert (tmp_path / "notes.txt").read_bytes() == b"hello"
assert "Downloaded 1 file" in capsys.readouterr().err
# Already there: refused by default, the run fails at the end; -f overwrites.
(tmp_path / "notes.txt").write_bytes(b"mine")
assert main("get", "Family/share/notes.txt") == 4
assert "already exists" in capsys.readouterr().err
assert (tmp_path / "notes.txt").read_bytes() == b"mine"
assert main("get", "-f", "Family/share/notes.txt") == 0
assert (tmp_path / "notes.txt").read_bytes() == b"hello"
assert main("get", "Family/share/2024/*.jpg", "-o", "pics/") == 0
assert sorted(p.name for p in (tmp_path / "pics").iterdir()) == ["a.jpg", "b.jpg"]
assert main("get", "Family/share/2024") == 1 # a folder needs -R
assert "use -R" in capsys.readouterr().err
assert main("get", "-R", "Family/share/2024", "-o", "tree/") == 0
assert (tmp_path / "tree/2024/deep/c.txt").read_bytes() == b"C"
assert (tmp_path / "tree/2024/empty").is_dir()
assert main("get", "Family/share/nope*") == 3
assert main("get", "Family/share/notes.txt", "-o", "-") == 0
assert capsys.readouterr().out == "hello"
def test_put_files_folders_and_stdin(node, tmp_path, monkeypatch, capsys):
node.add("share", "taken.txt", b"theirs")
(tmp_path / "a.txt").write_bytes(b"one")
(tmp_path / "Taken.TXT").write_bytes(b"mine")
scans = tmp_path / "Scans"
(scans / "2024").mkdir(parents=True)
(scans / "2024" / "p1.pdf").write_bytes(b"%PDF" + b"x" * 100_000)
(scans / "empty").mkdir()
assert main("put", str(tmp_path / "a.txt"), "Family/share") == 0
assert node.stored("share/a.txt") == b"one"
# A name already there, whatever its case: nothing is sent at all.
assert main("put", str(tmp_path / "a.txt"), str(tmp_path / "Taken.TXT"),
"Family/share") == 8
assert "already there" in capsys.readouterr().err
assert node.stored("share/Taken.TXT") is None
# -n skips it and sends the rest.
assert main("put", "-n", str(tmp_path / "Taken.TXT"), "Family/share") == 0
assert main("put", str(scans), "Family/share") == 1 # a folder needs -R
assert main("put", "-R", str(scans), "Family/share") == 0
assert node.stored("share/Scans/2024/p1.pdf").startswith(b"%PDF")
assert "share/Scans/empty" in node.dirs
bad = tmp_path / ".hidden"
bad.write_bytes(b"x")
assert main("put", str(bad), "Family/share") == 8
assert node.stored("share/.hidden") is None
monkeypatch.setattr("sys.stdin", io.TextIOWrapper(io.BytesIO(b"piped bytes")))
assert main("put", "-", "Family/share/from-stdin.txt") == 0
assert node.stored("share/from-stdin.txt") == b"piped bytes"
def test_put_into_a_read_only_folder_sends_nothing(node, tmp_path, capsys):
node.roots[0]["writable"] = False
(tmp_path / "a.txt").write_bytes(b"one")
assert main("put", str(tmp_path / "a.txt"), "Family/share") == 8
assert "read-only" in capsys.readouterr().err
def test_trust_lists_and_forgets_pins(hub, capsys):
assert main("trust", "list") == 0 # nothing stored: nothing pinned
assert main("--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE,
"--json", "login") == 0
assert json.loads(capsys.readouterr().out)["groups"] == 1
from meshbay_cli.client.store import Store
store = Store()
store.load()
store.state["pins"] = {"node-1": "pk-1"}
store.save()
assert main("--json", "trust", "list") == 0
assert json.loads(capsys.readouterr().out) == {"node-1": "pk-1"}
assert main("trust", "reset") == 1
assert main("trust", "reset", "node-2") == 3
assert main("trust", "reset", "node-1") == 0
assert main("--json", "trust", "list") == 0
capsys.readouterr()
assert main("--json", "logout") == 0
assert json.loads(capsys.readouterr().out) == {"signed_out": True}
def test_a_usage_error_is_exit_1_not_the_sign_in_code(capsys):
"""argparse's own 2 would read as a refused sign-in to a script."""
assert cli.main(["ls", "--no-such-option"]) == 1
assert cli.main(["no-such-action"]) == 1
def test_groups_lists_each_group_with_its_state(hub, capsys):
hub.groups.append({"id": "g2", "name": "Club/Photos", "owner_username": "bob-test",
"node_online": False, "visibility": "public", "is_admin": False,
"hosted": True, "last_activity_at": "2026-10-01T12:00:00+00:00"})
hub.groups.append({"id": "g3", "name": "Draft\x1b[31m", "owner_username": USERNAME,
"visibility": "private", "is_admin": True, "hosted": False})
assert main("groups") == 2 # signed in nowhere
capsys.readouterr()
assert main("--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE, "login") == 0
capsys.readouterr()
assert main("groups") == 0
header, *lines = capsys.readouterr().out.splitlines()
assert header.split() == ["NODE", "ROLE", "VISIBILITY", "LAST", "ACTIVITY", "GROUP"]
# In name order; a "/" in a name is encoded so the path can be typed back.
assert lines[0].split()[:3] == ["down", "member", "public"]
assert lines[0].endswith("Club%2FPhotos@bob-test")
assert "2026-10-0" in lines[0]
assert lines[1].split()[:3] == ["none", "owner", "private"]
assert "\x1b" not in lines[1] and lines[1].endswith(f"Draft\\x1b[31m@{USERNAME}")
assert lines[2].split()[:3] == ["up", "owner", "private"]
assert main("--json", "groups") == 0
rows = json.loads(capsys.readouterr().out)
assert [(r["group"], r["node"], r["role"]) for r in rows] == [
("Club/Photos", "down", "member"), ("Draft\x1b[31m", "none", "owner"),
("Family", "up", "owner")]
assert rows[0]["path"] == "Club%2FPhotos@bob-test"
def test_every_action_is_run_here():
"""Runs last in this module: every action the parser knows was run above."""
assert RUN == set(cli.ACTIONS), f"never run: {set(cli.ACTIONS) - RUN}"
|