diff options
Diffstat (limited to 'packages/meshbay-cli/tests/test_interactive.py')
| -rw-r--r-- | packages/meshbay-cli/tests/test_interactive.py | 159 |
1 files changed, 159 insertions, 0 deletions
diff --git a/packages/meshbay-cli/tests/test_interactive.py b/packages/meshbay-cli/tests/test_interactive.py new file mode 100644 index 0000000..680b9e0 --- /dev/null +++ b/packages/meshbay-cli/tests/test_interactive.py @@ -0,0 +1,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) |