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
|
"""
The interactive mode, driven by Textual's pilot against the in-memory hub and
node: what a person does with the keys, and what lands on disk and on the node.
"""
import asyncio
import time
import pytest
from fake_hub import PASSPHRASE, USERNAME, FakeHub
from fake_node import FakeNode
from meshbay_cli import __main__ as cli
from meshbay_cli.client import pool
from meshbay_cli.commands import context
from meshbay_cli.tui.app import MeshBayApp
from meshbay_cli.tui.panes import LocalPane, RemotePane
from meshbay_cli.tui.screens import MainScreen, SignInScreen
from meshbay_cli.tui.widgets import Choice
@pytest.fixture
def world(monkeypatch, tmp_path):
hub, node = FakeHub(), FakeNode()
monkeypatch.setattr(context, "Hub", lambda url: hub)
async def open_group(account, group):
return node.session()
monkeypatch.setattr(pool, "open_group", open_group)
work = tmp_path / "work"
work.mkdir()
monkeypatch.chdir(work)
return hub, node, work
def _app() -> MeshBayApp:
args = cli.build_parser().parse_args(["--hub", "https://hub.test"])
return MeshBayApp(context.Context(args))
async def _until(pilot, condition, timeout: float = 10) -> None:
deadline = time.monotonic() + timeout
while not condition():
assert time.monotonic() < deadline, "timed out"
await pilot.pause(0.05)
def _names(pane) -> list[str]:
"""Row names as shown, without the mark or a note such as "(read-only)"."""
return [pane.table.get_row_at(i)[0].plain.split(" (")[0].strip()
for i in range(pane.table.row_count)]
async def _open(pilot, pane, name: str) -> None:
await _until(pilot, lambda: name in _names(pane))
pane.table.move_cursor(row=_names(pane).index(name))
pane.table.focus()
await pilot.press("enter")
async def test_signing_in_then_downloading_and_uploading(world):
hub, node, work = world
node.add("share", "notes.txt", b"hello")
node.add("share/2024", "a.jpg", b"A" * 3000)
node.add("share", "[bold]x[/bold]\x1b.txt", b"tricky")
(work / "mine.txt").write_bytes(b"from here")
app = _app()
async with app.run_test(size=(140, 40)) as pilot:
# Nothing stored: the sign-in screen, then the panes.
await _until(pilot, lambda: isinstance(app.screen, SignInScreen))
app.screen.query_one("#user").value = USERNAME
app.screen.query_one("#pass").value = PASSPHRASE
app.screen.query_one("#pass").focus()
await pilot.press("enter")
await _until(pilot, lambda: isinstance(app.screen, MainScreen))
assert len(hub.devices) == 1, "signing in here remembers the machine"
remote = app.screen.query_one(RemotePane)
local = app.screen.query_one(LocalPane)
await _open(pilot, remote, f"Family@{USERNAME}/")
await _open(pilot, remote, "share/")
await _until(pilot, lambda: "notes.txt" in _names(remote))
# A name somebody else chose is shown as text, escapes and all.
assert "[bold]x[/bold]\\x1b.txt" in _names(remote)
# Mark two, F5: both land in the local folder.
for name in ("notes.txt", "2024/"):
remote.table.move_cursor(row=_names(remote).index(name))
await pilot.press("space")
await pilot.press("f5")
await _until(pilot, lambda: (work / "2024" / "a.jpg").exists()
and (work / "notes.txt").exists())
assert (work / "notes.txt").read_bytes() == b"hello"
assert (work / "2024" / "a.jpg").read_bytes() == b"A" * 3000
# Again: already there, asked once for the batch.
remote.table.move_cursor(row=_names(remote).index("notes.txt"))
remote.table.focus()
(work / "notes.txt").write_bytes(b"changed here")
await pilot.press("f5")
await _until(pilot, lambda: isinstance(app.screen, Choice))
await pilot.click("#choice-skip")
await pilot.pause(0.2)
assert (work / "notes.txt").read_bytes() == b"changed here"
# Upload from the left into the open folder.
await _until(pilot, lambda: isinstance(app.screen, MainScreen))
local.action_refresh()
await _until(pilot, lambda: "mine.txt" in _names(local))
local.table.move_cursor(row=_names(local).index("mine.txt"))
local.table.focus()
await pilot.press("f5")
await _until(pilot, lambda: node.stored("share/mine.txt") == b"from here")
# The node's index changed: the right pane shows it.
await _until(pilot, lambda: "mine.txt" in _names(remote))
# A name already there is asked about, never replaced.
local.table.focus()
await pilot.press("f5")
await _until(pilot, lambda: isinstance(app.screen, Choice))
await pilot.click("#choice-cancel")
await pilot.pause(0.2)
assert node.stored("share/mine.txt") == b"from here"
async def test_a_remembered_session_opens_on_the_panes(world, capsys):
hub, node, _ = world
assert await asyncio.to_thread(cli.main, ["--hub", "https://hub.test", "-u", USERNAME,
"--password", PASSPHRASE, "login"]) == 0
app = _app()
async with app.run_test(size=(120, 30)) as pilot:
await _until(pilot, lambda: isinstance(app.screen, MainScreen))
remote = app.screen.query_one(RemotePane)
await _until(pilot, lambda: f"Family@{USERNAME}/" in _names(remote))
async def test_a_read_only_folder_takes_no_upload(world):
hub, node, work = world
node.roots[0]["writable"] = False
node.add("share", "notes.txt", b"hello")
(work / "mine.txt").write_bytes(b"x")
assert await asyncio.to_thread(cli.main, ["--hub", "https://hub.test", "-u", USERNAME,
"--password", PASSPHRASE, "login"]) == 0
app = _app()
async with app.run_test(size=(120, 30)) as pilot:
await _until(pilot, lambda: isinstance(app.screen, MainScreen))
remote = app.screen.query_one(RemotePane)
local = app.screen.query_one(LocalPane)
await _open(pilot, remote, f"Family@{USERNAME}/")
await _open(pilot, remote, "share/")
await _until(pilot, lambda: "notes.txt" in _names(remote))
await _until(pilot, lambda: "mine.txt" in _names(local))
local.table.move_cursor(row=_names(local).index("mine.txt"))
local.table.focus()
await pilot.press("f5")
await pilot.pause(0.3)
assert node.stored("share/mine.txt") is None
assert any("does not accept uploads" in n.message for n in app._notifications)
|