"""
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")