diff options
Diffstat (limited to 'packages/meshbay-cli/tests/test_e2e_local.py')
| -rw-r--r-- | packages/meshbay-cli/tests/test_e2e_local.py | 570 |
1 files changed, 570 insertions, 0 deletions
diff --git a/packages/meshbay-cli/tests/test_e2e_local.py b/packages/meshbay-cli/tests/test_e2e_local.py new file mode 100644 index 0000000..4283819 --- /dev/null +++ b/packages/meshbay-cli/tests/test_e2e_local.py @@ -0,0 +1,570 @@ +""" +The real `meshbay-cli` against a real hub and a real node, both on loopback. + +Opt-in (MESHBAY_CLI_E2E=1): it starts a throwaway hub (SQLite, a fresh key) and +a node in a home of its own, which takes a while and runs Argon2 at 128 MB +several times. Nothing reaches a real hub or the developer's node. + +What it sets up, as an operator and a browser would: +- an account, activated directly in the throwaway hub's database (no mail); +- a group whose name holds a `/` and an `@`, which the hub allows and the path + syntax has to carry; +- a node linked to the account, hosting the group over a folder whose names + are the awkward ones real data has: spaces, accents, an apostrophe, a long + name without spaces, and a name with an escape character in it; +- the account's first contact with the node — a member invitation code, then a + keypair bundle left on the node — done here the way a browser does it, + because the CLI does not join groups. + +Then the CLI: login, ls of groups, of a root, recursive, JSON, a second run +from the identity it kept, and logout. + + MESHBAY_CLI_E2E=1 pytest packages/meshbay-cli/tests/test_e2e_local.py -v + +MESHBAY_CLI_E2E_BIN names another meshbay-cli to run in place of the one in +this venv: a packaged build, through a wrapper that mounts it where it installs. +""" + +import asyncio +import base64 +import json +import os +import re +import secrets +import socket +import sqlite3 +import subprocess +import sys +import time +from pathlib import Path + +import httpx +import pytest + +pytestmark = [ + pytest.mark.skipif(os.environ.get("MESHBAY_CLI_E2E") != "1", + reason="opt-in: set MESHBAY_CLI_E2E=1 (see the module docstring)"), + pytest.mark.timeout(600), +] + +USERNAME = "alice-e2e-cli" +PASSPHRASE = "a long enough test passphrase, nobody's" +GROUP_NAME = "Some Saga/Box @Set" +BIN = Path(sys.executable).parent + + +def _free_port() -> int: + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _wait_http(url: str, timeout: float = 60) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + try: + httpx.get(url, timeout=2) + return + except httpx.HTTPError: + time.sleep(0.3) + raise AssertionError(f"{url} did not come up") + + +@pytest.fixture(scope="module") +def world(tmp_path_factory): + root = tmp_path_factory.mktemp("e2e") + procs: list[subprocess.Popen] = [] + try: + yield _build(root, procs) + finally: + for p in procs: + p.terminate() + for p in procs: + try: + p.wait(10) + except subprocess.TimeoutExpired: + p.kill() + + +def _build(root: Path, procs: list) -> dict: + from cryptography.hazmat.primitives import serialization + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + from meshbay_cli.client.keys import auth_key + + # ── The hub ────────────────────────────────────────────────────────────── + hub_home = root / "hub-home" + hub_home.mkdir() + key = root / "hub_private.pem" + key.write_bytes(Ed25519PrivateKey.generate().private_bytes( + serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, + serialization.NoEncryption())) + db = root / "hub.db" + port = _free_port() + hub_url = f"http://127.0.0.1:{port}" + hub_env = {**os.environ, "HOME": str(hub_home), + "MESHBAY_DATABASE_URL": f"sqlite+aiosqlite:///{db}", + "MESHBAY_HUB_KEY": str(key), "MESHBAY_HUB_ID": "e2e-hub", + "MESHBAY_CAPTCHA_ENABLED": "false"} + hub_log = open(root / "hub.log", "w") + procs.append(subprocess.Popen( + [sys.executable, "-m", "uvicorn", "meshbay_hub.app:create_app", "--factory", + "--host", "127.0.0.1", "--port", str(port)], + env=hub_env, stdout=hub_log, stderr=subprocess.STDOUT)) + _wait_http(f"{hub_url}/v1/hub/version") + + ak = auth_key(USERNAME, PASSPHRASE) + with httpx.Client(base_url=hub_url, timeout=60) as hub: + r = hub.post("/v1/users/register", json={ + "username": USERNAME, "email": "alice@example.invalid", "auth_key": ak}) + assert r.status_code == 201, r.text + with sqlite3.connect(db) as conn: + conn.execute("UPDATE users SET status='active' WHERE username=?", (USERNAME,)) + r = hub.post("/v1/users/login", json={"username": USERNAME, "auth_key": ak}) + assert r.status_code == 200, r.text + auth = {"Authorization": f"Bearer {r.json()['access_token']}"} + r = hub.post("/v1/groups", headers=auth, json={ + "name": GROUP_NAME, "visibility": "private", "join_policy": "invite"}) + assert r.status_code == 201, r.text + group_id = r.json()["group_id"] + + # ── The node ───────────────────────────────────────────────────────── + node_home = root / "node-home" + node_home.mkdir() + share = root / "share" + (share / "2024" / "été à la mer").mkdir(parents=True) + (share / "2024" / "été à la mer" / "plage de l'ouest.jpg").write_bytes( + secrets.token_bytes(3000)) + (share / "2024" / "A_Very_Long_Name_Without_Any_Spaces_In_It_1992-2000.flac")\ + .write_bytes(secrets.token_bytes(2_500_000)) + (share / "notes.txt").write_text("hello\n") + (share / "evil\x1b[31mred.txt").write_text("escape\n") + node_env = {**os.environ, "HOME": str(node_home), + "XDG_CONFIG_HOME": str(node_home / ".config"), + "XDG_DATA_HOME": str(node_home / ".local/share"), + "XDG_STATE_HOME": str(node_home / ".local/state")} + node = [str(BIN / "meshbay-node")] + + def run_node(*args: str) -> str: + p = subprocess.run(node + list(args), env=node_env, capture_output=True, + text=True, timeout=120) + assert p.returncode == 0, f"meshbay-node {args}: {p.stdout}\n{p.stderr}" + return p.stdout + + out = run_node("init", "--hub-url", hub_url, "--username", USERNAME) + node_pk = re.search(r"Node key: (\S+)", out).group(1) + cfg = next(node_home.rglob("node.toml")) + ui_port = _free_port() + cfg.write_text(cfg.read_text().replace("ui_port = 18000", + f"ui_port = {ui_port}")) + r = hub.put("/v1/users/me/node_key", headers=auth, + json={"pk_node_ed25519": node_pk}) + assert r.status_code == 200, r.text + + node_log = open(root / "node.log", "w") + procs.append(subprocess.Popen(node, env=node_env, stdout=node_log, + stderr=subprocess.STDOUT)) + _wait_http(f"http://127.0.0.1:{ui_port}/", timeout=90) + + # Attaching reloads the node by itself; the key needs the group loaded. + run_node("group", "add", group_id, "--dir", str(share)) + deadline = time.monotonic() + 90 + while "GEK ready" not in (out := _try(run_node, "gek", "init", "--group", group_id)): + assert time.monotonic() < deadline, f"gek init never succeeded: {out}" + time.sleep(2) + code = re.search(r"INVITATION CODE\s+(\S+)", + run_node("member", "invite", USERNAME, "--group", group_id)).group(1) + + # The node is online at the hub once it lists itself for the group. + deadline = time.monotonic() + 90 + while not hub.get(f"/v1/groups/{group_id}/nodes", headers=auth).json()["nodes"]: + assert time.monotonic() < deadline, "the node never reached the hub" + time.sleep(1) + + app = App(hub_url, group_id) + asyncio.run(app.first_contact(code)) + return {"root": root, "hub_url": hub_url, "group_id": group_id, "db": db, + "share": share, "app": app} + + +def _try(fn, *args) -> str: + try: + return fn(*args) + except AssertionError as e: + return str(e) + + +class App: + """The application's side of the test: a device already linked to the node.""" + + def __init__(self, hub_url: str, group_id: str): + self.hub_url, self.group_id = hub_url, group_id + self.ident = None + + async def session(self, code: str = ""): + """Signed in, handshake done, joined: (channel, close, node_pk, nonce_s, user_id).""" + import os as _os + + from aiortc import ( + RTCConfiguration, + RTCIceServer, + RTCPeerConnection, + RTCSessionDescription, + ) + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey + from meshbay_cli.client.account import Account + from meshbay_cli.client.hub import Hub + from meshbay_cli.client.keys import Identity, b64, unb64 + from meshbay_cli.client.session import STUN_SERVERS, Channel, _dtls_fingerprint + from meshbay_common import MNP_VERSION + from meshbay_common.crypto import unwrap_gek_aes + from meshbay_common.handshake import MNP_MIN_SUPPORTED, make_proof, webrtc_binding + from meshbay_common.join import join_transcript + + hub = Hub(self.hub_url) + account = Account(hub, USERNAME, lambda: PASSPHRASE) + await account.sign_in(remember=False) + token = await account.token() + node = (await hub.group_nodes(token, self.group_id))[0] + mnp = await hub.mnp_token(token, node["pk_node"], self.group_id) + pc = RTCPeerConnection(RTCConfiguration( + iceServers=[RTCIceServer(urls=u) for u in STUN_SERVERS])) + dc = pc.createDataChannel("mnp", ordered=True) + ch = Channel(dc) + opened = asyncio.Event() + dc.on("open", opened.set) + await pc.setLocalDescription(await pc.createOffer()) + offer = pc.localDescription.sdp + answer = await hub.offer(token, node["node_id"], offer) + await pc.setRemoteDescription(RTCSessionDescription(sdp=answer, type="answer")) + await asyncio.wait_for(opened.wait(), 30) + binding = webrtc_binding(_dtls_fingerprint(offer), _dtls_fingerprint(answer)) + nonce_c = _os.urandom(32) + chal = await ch.request({"type": "handshake", "v": MNP_VERSION, + "v_min": MNP_MIN_SUPPORTED, "token": mnp, + "group_id": self.group_id, "nonce": b64(nonce_c)}, + "handshake_challenge") + nonce_s, node_pk = unb64(chal["nonce"]), chal["node_pk"] + if self.ident is None: + self.ident = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()) + ident, ts = self.ident, int(time.time()) + joined = await ch.request({ + "type": "join_request", "v": MNP_VERSION, "group_id": self.group_id, + "pk_ed25519": ident.pk_ed, "pk_x25519": ident.pk_x, "code": code, "ts": ts, + "sig": b64(ident.sk_ed.sign(join_transcript( + node_pk_b64=node_pk, group_id=self.group_id, user_id=account.user_id, + pk_ed25519_b64=ident.pk_ed, pk_x25519_b64=ident.pk_x, + nonce_node=nonce_s, ts=ts)))}, "join_result") + assert joined.get("ok"), joined + gek = unwrap_gek_aes(joined, ident.sk_x.private_bytes_raw(), + ident.sk_x.public_key().public_bytes_raw()) + await ch.request({"type": "handshake_response", "v": MNP_VERSION, + "proof": b64(make_proof(gek, "client", self.group_id, nonce_c, + nonce_s, binding))}, "handshake_ack") + + async def close(): + await pc.close() + await account.close() + await hub.aclose() + return ch, close, node_pk, nonce_s, account + + async def first_contact(self, code: str) -> None: + """What a browser does the first time: join with the code, leave a bundle.""" + import os as _os + + from cryptography.hazmat.primitives.ciphers.aead import AESGCM + from meshbay_cli.client.keys import node_bundle_key + from meshbay_common import MNP_VERSION + + ch, close, node_pk, _, account = await self.session(code) + # MBK3, as keyderive.js encryptBundle writes it. + nonce = _os.urandom(12) + aad = f"meshbay:bundle:v3|{account.user_id}|{node_pk}".encode() + ct = AESGCM(node_bundle_key(account.master, node_pk)).encrypt( + nonce, json.dumps(self.ident.to_pkcs8()).encode(), aad) + bundle = base64.b64encode(b"MBK3" + bytes([1]) + nonce + ct).decode() + stored = await ch.request({"type": "keypair_bundle_store", "v": MNP_VERSION, + "bundle_enc": bundle}, "ack") + assert stored.get("detail") == "keypair_bundle_stored", stored + await close() + + async def withdraw_bundle(self) -> None: + """Browser access turned off: the application withdraws the node's copy.""" + from meshbay_common import MNP_VERSION + ch, close, *_ = await self.session() + r = await ch.request({"type": "keypair_bundle_delete", "v": MNP_VERSION}, "ack") + assert r.get("detail") == "keypair_bundle_deleted", r + await close() + + async def approve(self, code: str) -> None: + """Approve a device by the code it shows, as the group's Settings tab does.""" + from meshbay_cli.client.keys import b64 + from meshbay_common import MNP_VERSION + from meshbay_common.device import device_add_transcript, device_code_hash + ch, close, node_pk, nonce_s, account = await self.session() + listed = await ch.request({"type": "device_lookup", "v": MNP_VERSION}, + "device_lookup_result") + raw = code.replace("-", "").upper() + match = next(r for r in listed["requests"] + if device_code_hash(raw, r["pk_ed25519"], r["pk_x25519"]) + == r["code_hash"]) + ts = int(time.time()) + sig = self.ident.sk_ed.sign(device_add_transcript( + node_pk_b64=node_pk, user_id=account.user_id, + pk_ed25519_b64=match["pk_ed25519"], pk_x25519_b64=match["pk_x25519"], + nonce_node=nonce_s, ts=ts)) + await ch.request({"type": "device_add", "v": MNP_VERSION, + "pk_ed25519": match["pk_ed25519"], + "pk_x25519": match["pk_x25519"], + "code_hash": match["code_hash"], "ts": ts, "sig": b64(sig)}, + "device_add_ack") + await close() + + +def cli(world: dict, *args: str, stdin: str | None = None, + home: str = "cli") -> subprocess.CompletedProcess: + env = {**os.environ, "MESHBAY_CLI_HOME": str(world["root"] / home), + "PYTHON_KEYRING_BACKEND": "keyring.backends.fail.Keyring"} + env.pop("MESHBAY_PASSWORD", None) + exe = os.environ.get("MESHBAY_CLI_E2E_BIN") or str(BIN / "meshbay-cli") + return subprocess.run([exe, *args], env=env, input=stdin, + capture_output=True, text=True, timeout=180) + + +def _handle() -> str: + return GROUP_NAME.replace("%", "%25").replace("/", "%2F") + "@" + USERNAME + + +def test_the_cli_against_a_real_hub_and_node(world): + hub = world["hub_url"] + + r = cli(world, "--hub", hub, "-u", USERNAME, "login", "--password-file", "-", + stdin=PASSPHRASE + "\n") + assert r.returncode == 0, r.stderr + assert "Signed in" in r.stdout and "1 group" in r.stdout + # No key storage in this environment: the file fallback, said once. + assert "No OS key storage here" in r.stderr + + # The groups, from the hub alone; the stored session needs no passphrase. + r = cli(world, "ls") + assert r.returncode == 0, r.stderr + assert r.stdout.strip() == _handle() + r = cli(world, "groups") + assert r.returncode == 0, r.stderr + print(r.stdout) + [line] = r.stdout.splitlines()[1:] + assert line.split()[:3] == ["up", "owner", "private"] and line.endswith(_handle()) + + # One group, by its handle: its root. + r = cli(world, "-vv", "ls", "-l", _handle()) + assert r.returncode == 0, r.stderr + assert "keypair_bundle_fetch" in r.stderr # first contact: the node's copy + assert re.search(r"^root\s.*share/\s+\(writable\)$", r.stdout, re.M), r.stdout + + # Recursive: the awkward names arrive, the escape character is escaped. + r = cli(world, "-vv", "ls", "-R", "-l", f"{_handle()}/share") + assert r.returncode == 0, r.stderr + assert "keypair_bundle_fetch" not in r.stderr # the identity was kept + out = r.stdout + assert "plage de l'ouest.jpg" in out + assert "A_Very_Long_Name_Without_Any_Spaces_In_It_1992-2000.flac" in out + assert "\x1b" not in out and "evil\\x1b[31mred.txt" in out + assert "été à la mer/" in out + + # By name alone, case-insensitively, and as JSON. + r = cli(world, "--json", "ls", "some saga%2Fbox @set/share/2024") + assert r.returncode == 0, r.stderr + items = json.loads(r.stdout) + names = {i["name"]: i for i in items} + assert names["été à la mer"]["kind"] == "dir" + flac = names["A_Very_Long_Name_Without_Any_Spaces_In_It_1992-2000.flac"] + assert flac["kind"] == "file" and flac["type"] == "audio" + assert flac["size"] == 2_500_000 and len(flac["id"]) == 64 + + # Not found is exit 3; an unknown group too. + assert cli(world, "ls", f"{_handle()}/share/nope").returncode == 3 + assert cli(world, "ls", "No Such Group").returncode == 3 + + # ── get ────────────────────────────────────────────────────────────────── + out_dir = world["root"] / "downloads" + flac = "A_Very_Long_Name_Without_Any_Spaces_In_It_1992-2000.flac" + r = cli(world, "get", f"{_handle()}/share/2024/{flac}", "-o", f"{out_dir}/") + assert r.returncode == 0, r.stderr + assert (out_dir / flac).read_bytes() == (world["share"] / "2024" / flac).read_bytes() + r = cli(world, "get", "-R", f"{_handle()}/share", "-o", f"{out_dir}/tree/") + assert r.returncode == 0, r.stderr + got = out_dir / "tree" / "share" + assert (got / "2024" / "été à la mer" / "plage de l'ouest.jpg").read_bytes() == \ + (world["share"] / "2024" / "été à la mer" / "plage de l'ouest.jpg").read_bytes() + # The escape character could not be written as it is: renamed, and said so. + assert "evil\\x1b[31mred.txt is saved as evil_[31mred.txt" in r.stderr + assert (got / "evil_[31mred.txt").exists() + r = cli(world, "get", f"{_handle()}/share/notes.txt", "-o", "-") + assert r.returncode == 0 and r.stdout == "hello\n" + + # ── put ────────────────────────────────────────────────────────────────── + local = world["root"] / "to-send" + (local / "Scans" / "2024").mkdir(parents=True) + scan = secrets.token_bytes(300_000) + (local / "Scans" / "2024" / "page 1.pdf").write_bytes(scan) + (local / "Scans" / "empty").mkdir() + (local / "Notes.TXT").write_bytes(b"clash") + r = cli(world, "put", "-R", str(local / "Scans"), f"{_handle()}/share/2024") + assert r.returncode == 0, r.stderr + assert (world["share"] / "2024" / "Scans" / "2024" / "page 1.pdf").read_bytes() == scan + assert (world["share"] / "2024" / "Scans" / "empty").is_dir() + # notes.txt is there already, whatever the case: nothing sent. + r = cli(world, "put", str(local / "Notes.TXT"), f"{_handle()}/share") + assert r.returncode == 8 and "already there" in r.stderr + assert not (world["share"] / "Notes.TXT").exists() + r = cli(world, "put", "-", f"{_handle()}/share/piped.txt", stdin="through stdin") + assert r.returncode == 0, r.stderr + assert (world["share"] / "piped.txt").read_text() == "through stdin" + # And what was sent can be fetched back, once the node has indexed it. + deadline = time.monotonic() + 60 + while True: + r = cli(world, "get", f"{_handle()}/share/2024/Scans/2024/page 1.pdf", "-o", + str(out_dir / "back.pdf")) + if r.returncode == 0 or time.monotonic() > deadline: + break + time.sleep(1) + assert r.returncode == 0, r.stderr + assert (out_dir / "back.pdf").read_bytes() == scan + + # ── ls -R over every group ─────────────────────────────────────────────── + deadline = time.monotonic() + 60 + while True: + r = cli(world, "ls", "-R", "/") + assert r.returncode == 0, r.stderr + if "piped.txt" in r.stdout or time.monotonic() > deadline: + break + time.sleep(1) # the node indexes what it received on its own schedule + assert "page 1.pdf" in r.stdout and "piped.txt" in r.stdout, r.stdout + + # The device key was registered, and logout retires it. + with sqlite3.connect(world["db"]) as conn: + assert conn.execute("SELECT count(*) FROM user_devices").fetchone()[0] == 1 + r = cli(world, "logout") + assert r.returncode == 0, r.stderr + assert "Signed out." in r.stdout + with sqlite3.connect(world["db"]) as conn: + assert conn.execute("SELECT count(*) FROM user_devices").fetchone()[0] == 0 + assert not (world["root"] / "cli" / "secrets.enc").exists() + + # Signed out: a run needs credentials again, and says so. + r = cli(world, "ls") + assert r.returncode == 2 and "Not signed in" in r.stderr + + +def test_a_changed_node_key_is_refused_until_reset(world): + def run(*args, stdin=None): + return cli(world, *args, stdin=stdin, home="cli-trust") + + r = run("--hub", world["hub_url"], "-u", USERNAME, "login", "--password-file", "-", + stdin=PASSPHRASE + "\n") + assert r.returncode == 0, r.stderr + assert run("ls", _handle()).returncode == 0 + + r = run("--json", "trust", "list") + [(node_id, pk)] = json.loads(r.stdout).items() + state_path = world["root"] / "cli-trust" / "state.json" + state = json.loads(state_path.read_text()) + state["pins"][node_id] = base64.b64encode(b"\x01" * 32).decode() + state_path.write_text(json.dumps(state)) + r = run("ls", _handle()) + assert r.returncode == 6 and "trust reset" in r.stderr, r.stderr + assert run("trust", "reset", "no-such-node").returncode == 3 + r = run("trust", "reset", node_id) + assert r.returncode == 0, r.stderr + assert run("ls", _handle()).returncode == 0 + assert json.loads(run("--json", "trust", "list").stdout) == {node_id: pk} + assert run("logout").returncode == 0 + + +def test_a_machine_with_no_identity_there_is_linked_as_a_device(world): + """Browser access off: the node keeps no copy, so a new machine asks to be + approved, shows a code, and joins with its own keys once approved.""" + app = world["app"] + asyncio.run(app.withdraw_bundle()) + + def run(*args, stdin=None): + return cli(world, *args, stdin=stdin, home="cli2") + + r = run("--hub", world["hub_url"], "-u", USERNAME, "login", "--password-file", "-", + stdin=PASSPHRASE + "\n") + assert r.returncode == 0, r.stderr + + # No terminal here: the code is shown, and the run ends rather than waits. + r = run("ls", _handle()) + assert r.returncode == 5, r.stderr + found = re.search(r"enter\s+([0-9A-Z]{4}-[0-9A-Z]{4})", r.stderr) + assert found, r.stderr + code = found.group(1) + # A second run shows the same code: the request it filed is still pending. + r = run("ls", _handle()) + assert r.returncode == 5 and code in r.stderr, r.stderr + + asyncio.run(app.approve(code)) + r = run("ls", "-l", _handle()) + assert r.returncode == 0, r.stderr + assert "share/" in r.stdout + + assert run("logout").returncode == 0 + + +async def test_the_interactive_mode_against_the_real_node(world, monkeypatch, tmp_path): + """The panes, opened on a real group, and F5 bringing a file back.""" + from meshbay_cli import __main__ as main_module + from meshbay_cli.commands.context import Context + from meshbay_cli.tui.app import MeshBayApp + from meshbay_cli.tui.panes import RemotePane + from meshbay_cli.tui.screens import MainScreen + + # Browser access off, whichever test ran before: this machine must be linked. + await world["app"].withdraw_bundle() + r = await asyncio.to_thread(cli, world, "--hub", world["hub_url"], "-u", USERNAME, + "login", "--password-file", "-", + stdin=PASSPHRASE + "\n", home="cli-tui") + assert r.returncode == 0, r.stderr + monkeypatch.setenv("MESHBAY_CLI_HOME", str(world["root"] / "cli-tui")) + monkeypatch.chdir(tmp_path) + args = main_module.build_parser().parse_args(["--hub", world["hub_url"]]) + app = MeshBayApp(Context(args)) + + def names(pane): + return [pane.table.get_row_at(i)[0].plain.split(" (")[0].strip() + for i in range(pane.table.row_count)] + + async def until(pilot, cond, timeout=60): + deadline = time.monotonic() + timeout + while not cond(): + shown = (names(app.screen.query_one(RemotePane)) + if isinstance(app.screen, MainScreen) else None) + assert time.monotonic() < deadline, ( + f"timed out on {type(app.screen).__name__}; " + f"notified: {[n.message for n in app._notifications]}; remote: {shown}") + await pilot.pause(0.1) + + async with app.run_test(size=(140, 40)) as pilot: + await until(pilot, lambda: isinstance(app.screen, MainScreen)) + remote = app.screen.query_one(RemotePane) + name = _handle() + "/" + await until(pilot, lambda: name in names(remote)) + remote.table.move_cursor(row=names(remote).index(name)) + remote.table.focus() + await pilot.press("enter") + # The node keeps no copy of the identity: this machine is a new device, + # shows its code, and waits while it is approved elsewhere — then the + # group opens by itself. + await until(pilot, lambda: any("enter " in n.message for n in app._notifications)) + note = next(n.message for n in app._notifications if "enter " in n.message) + code = re.search(r"enter ([0-9A-Z]{4}-[0-9A-Z]{4})", note).group(1) + await world["app"].approve(code) + await until(pilot, lambda: "share/" in names(remote)) + remote.table.move_cursor(row=names(remote).index("share/")) + await pilot.press("enter") + await until(pilot, lambda: "notes.txt" in names(remote)) + remote.table.move_cursor(row=names(remote).index("notes.txt")) + await pilot.press("f5") + await until(pilot, lambda: (tmp_path / "notes.txt").exists() + and (tmp_path / "notes.txt").read_text() == "hello\n") |