""" 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) async def test_delete_removes_what_is_yours_and_asks_first(world): from fake_hub import USER_ID hub, node, work = world mine = node.add("share", "mine.txt", b"1", uploader_id=USER_ID) theirs = node.add("share", "theirs.txt", b"2", uploader_id="someone-else") 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 _open(pilot, remote, f"Family@{USERNAME}/") await _open(pilot, remote, "share/") await _until(pilot, lambda: "theirs.txt" in _names(remote)) # Somebody else's file, marked with one's own: nothing is deleted. for name in ("mine.txt", "theirs.txt"): remote.table.move_cursor(row=_names(remote).index(name)) await pilot.press("space") await pilot.press("delete") await _until(pilot, lambda: isinstance(app.screen, Choice)) assert "theirs.txt" in str(app.screen.message) await pilot.click("#choice-ok") assert mine["id"] in node.files and theirs["id"] in node.files # One's own: asked, cancelled, asked again, deleted, and gone from the pane. await _until(pilot, lambda: isinstance(app.screen, MainScreen)) remote.marks.clear() remote.redraw() remote.table.move_cursor(row=_names(remote).index("mine.txt")) await pilot.press("delete") await _until(pilot, lambda: isinstance(app.screen, Choice)) await pilot.click("#choice-cancel") assert mine["id"] in node.files await _until(pilot, lambda: isinstance(app.screen, MainScreen)) await pilot.press("delete") await _until(pilot, lambda: isinstance(app.screen, Choice)) await pilot.click("#choice-delete") await _until(pilot, lambda: mine["id"] not in node.files) await _until(pilot, lambda: "mine.txt" not in _names(remote)) assert theirs["id"] in node.files # On the left, Delete removes nothing of this machine's. app.screen.query_one(LocalPane).table.focus() await pilot.press("delete") await pilot.pause(0.2) assert any("select them on the right" in n.message for n in app._notifications)