path: root/packages/meshbay-cli/tests/test_cli_actions.py

"""
Every action, run through `main()` with the hub and the node stubbed.

A branch nobody has run is covered by nothing — the node's `reload` once shipped
with a missing import. `test_every_action_is_run_here` refuses an action added
to the parser without a run below.
"""

import io
import json
import os
import sys

import pytest
from fake_hub import PASSPHRASE, USERNAME, FakeHub
from meshbay_cli import __main__ as cli
from meshbay_cli.client import pool
from meshbay_cli.commands import context

RUN: set[str] = set()


@pytest.fixture
def hub(monkeypatch):
    fake = FakeHub()
    monkeypatch.setattr(context, "Hub", lambda url: fake)
    return fake


def main(*argv: str) -> int:
    RUN.add(next(a for a in argv if a in cli.ACTIONS))
    return cli.main(list(argv))


class FakeSession:
    async def index(self):
        return {"roots": [{"name": "share", "available": True, "writable": True}],
                "dirs": ["share"],
                "entries": [{"id": "a" * 64, "name": "notes.txt", "path": "share",
                             "size": 6, "type": "document", "added_at": 0}]}

    async def close(self):
        pass


def test_login_ls_logout(hub, monkeypatch, capsys):
    monkeypatch.setattr(pool, "open_group", lambda account, group: _session())

    assert main("--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE, "login") == 0
    assert "Signed in: alice-test" in capsys.readouterr().out

    assert main("ls") == 0
    assert capsys.readouterr().out.strip() == f"Family@{USERNAME}"

    assert main("ls", "--json", "Family/share") == 0
    [item] = json.loads(capsys.readouterr().out)
    assert item["path"] == f"Family@{USERNAME}/share/notes.txt" and item["kind"] == "file"

    assert main("logout") == 0
    assert "Signed out." in capsys.readouterr().out
    assert not hub.devices

    assert main("ls") == 2
    assert "Not signed in" in capsys.readouterr().err


def test_login_without_remembering_stores_nothing(hub, isolated_home, capsys):
    assert main("--hub", hub.url, "-u", USERNAME, "login", "--no-remember",
                "--password", PASSPHRASE) == 0
    assert "Credentials accepted" in capsys.readouterr().out
    assert not hub.devices
    assert not any(isolated_home.rglob("secrets.enc"))


def test_a_wrong_passphrase_exits_2(hub, capsys):
    assert main("--hub", hub.url, "-u", USERNAME, "--password", "nope", "login") == 2
    assert "wrong username or passphrase" in capsys.readouterr().err


def test_no_passphrase_and_no_terminal_exits_2(hub, monkeypatch, capsys):
    monkeypatch.setattr("sys.stdin.isatty", lambda: False)
    assert main("--hub", hub.url, "-u", USERNAME, "login") == 2
    assert "MESHBAY_PASSWORD" in capsys.readouterr().err


def test_the_null_device_is_no_terminal(hub, monkeypatch, capsys):
    """`< NUL` on Windows: a character device, which isatty() calls a terminal.
    A script run so was asked for a passphrase it could not type."""
    from meshbay_cli.output import terminal
    with open(os.devnull) as null:
        assert not terminal(null)
        monkeypatch.setattr("sys.stdin", null)
        assert main("--hub", hub.url, "-u", USERNAME, "login") == 2
    assert "MESHBAY_PASSWORD" in capsys.readouterr().err


async def _session():
    return FakeSession()


@pytest.fixture
def node(hub, monkeypatch, capsys):
    from fake_node import FakeNode
    fake = FakeNode()

    async def open_group(account, group):
        return fake.session()
    monkeypatch.setattr(pool, "open_group", open_group)
    assert main("--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE, "login") == 0
    capsys.readouterr()
    return fake


def test_get_files_folders_and_patterns(node, tmp_path, monkeypatch, capsys):
    node.add("share", "notes.txt", b"hello")
    node.add("share/2024", "a.jpg", b"A" * 3000)
    node.add("share/2024", "b.jpg", b"B" * 10)
    node.add("share/2024/deep", "c.txt", b"C")
    node.dirs.add("share/2024/empty")
    monkeypatch.chdir(tmp_path)

    assert main("get", "Family/share/notes.txt") == 0
    assert (tmp_path / "notes.txt").read_bytes() == b"hello"
    assert "Downloaded 1 file" in capsys.readouterr().err

    # Already there: refused by default, the run fails at the end; -f overwrites.
    (tmp_path / "notes.txt").write_bytes(b"mine")
    assert main("get", "Family/share/notes.txt") == 4
    assert "already exists" in capsys.readouterr().err
    assert (tmp_path / "notes.txt").read_bytes() == b"mine"
    assert main("get", "-f", "Family/share/notes.txt") == 0
    assert (tmp_path / "notes.txt").read_bytes() == b"hello"

    assert main("get", "Family/share/2024/*.jpg", "-o", "pics/") == 0
    assert sorted(p.name for p in (tmp_path / "pics").iterdir()) == ["a.jpg", "b.jpg"]

    assert main("get", "Family/share/2024") == 1          # a folder needs -R
    assert "use -R" in capsys.readouterr().err
    assert main("get", "-R", "Family/share/2024", "-o", "tree/") == 0
    assert (tmp_path / "tree/2024/deep/c.txt").read_bytes() == b"C"
    assert (tmp_path / "tree/2024/empty").is_dir()

    assert main("get", "Family/share/nope*") == 3

    assert main("get", "Family/share/notes.txt", "-o", "-") == 0
    assert capsys.readouterr().out == "hello"


def test_put_files_folders_and_stdin(node, tmp_path, monkeypatch, capsys):
    node.add("share", "taken.txt", b"theirs")
    (tmp_path / "a.txt").write_bytes(b"one")
    (tmp_path / "Taken.TXT").write_bytes(b"mine")
    scans = tmp_path / "Scans"
    (scans / "2024").mkdir(parents=True)
    (scans / "2024" / "p1.pdf").write_bytes(b"%PDF" + b"x" * 100_000)
    (scans / "empty").mkdir()

    assert main("put", str(tmp_path / "a.txt"), "Family/share") == 0
    assert node.stored("share/a.txt") == b"one"

    # A name already there, whatever its case: nothing is sent at all.
    assert main("put", str(tmp_path / "a.txt"), str(tmp_path / "Taken.TXT"),
                "Family/share") == 8
    assert "already there" in capsys.readouterr().err
    assert node.stored("share/Taken.TXT") is None
    # -n skips it and sends the rest.
    assert main("put", "-n", str(tmp_path / "Taken.TXT"), "Family/share") == 0

    assert main("put", str(scans), "Family/share") == 1      # a folder needs -R
    assert main("put", "-R", str(scans), "Family/share") == 0
    assert node.stored("share/Scans/2024/p1.pdf").startswith(b"%PDF")
    assert "share/Scans/empty" in node.dirs

    bad = tmp_path / ".hidden"
    bad.write_bytes(b"x")
    assert main("put", str(bad), "Family/share") == 8
    assert node.stored("share/.hidden") is None

    monkeypatch.setattr("sys.stdin", io.TextIOWrapper(io.BytesIO(b"piped bytes")))
    assert main("put", "-", "Family/share/from-stdin.txt") == 0
    assert node.stored("share/from-stdin.txt") == b"piped bytes"


def test_put_into_a_read_only_folder_sends_nothing(node, tmp_path, capsys):
    node.roots[0]["writable"] = False
    (tmp_path / "a.txt").write_bytes(b"one")
    assert main("put", str(tmp_path / "a.txt"), "Family/share") == 8
    assert "read-only" in capsys.readouterr().err


def _max_path_in_force() -> bool:
    if sys.platform != "win32":
        return False
    import winreg
    try:
        with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE,
                            r"SYSTEM\CurrentControlSet\Control\FileSystem") as key:
            return winreg.QueryValueEx(key, "LongPathsEnabled")[0] == 0
    except OSError:
        return True


@pytest.mark.skipif(not _max_path_in_force(), reason="Windows with MAX_PATH in force")
def test_get_and_put_past_max_path(node, tmp_path, capsys):
    """Windows refuses a plain path of 260 characters or more. The node uses the
    long form at each disk call, and the client has to as well: `get -R` into a
    deep folder died with a traceback creating its first folder."""
    import shutil

    from meshbay_common.paths import long_path
    segment = "A folder name as long as the ones a real library has"
    deep = tmp_path / "deep"
    while len(str(deep)) < 240:
        deep = deep / segment
    try:
        node.add("share/2024", "a.jpg", b"A" * 3000)
        assert main("get", "-R", "Family/share/2024", "-o", f"{deep}{os.sep}") == 0, \
            capsys.readouterr().err
        got = deep / "2024" / "a.jpg"
        assert len(str(got)) >= 260
        with open(long_path(got), "rb") as f:
            assert f.read() == b"A" * 3000

        # A part left from an interrupted run, resumed past the limit too.
        node.add("share/2024", "b.bin", b"B" * 10)
        with open(long_path(deep / "2024" / "b.bin.meshbay-part"), "wb") as f:
            f.write(b"B")
        assert main("get", "-R", "Family/share/2024", "-o", f"{deep}{os.sep}", "-n") == 0
        with open(long_path(deep / "2024" / "b.bin"), "rb") as f:
            assert f.read() == b"B" * 10

        # And a tree past the limit, sent.
        src = deep / "Scans"
        os.makedirs(long_path(src / segment))
        with open(long_path(src / segment / "page.pdf"), "wb") as f:
            f.write(b"%PDF")
        assert main("put", "-R", str(src), "Family/share") == 0, capsys.readouterr().err
        assert node.stored(f"share/Scans/{segment}/page.pdf") == b"%PDF"
    finally:
        shutil.rmtree("\\\\?\\" + str(tmp_path / "deep"), ignore_errors=True)


def test_trust_lists_and_forgets_pins(hub, capsys):
    assert main("trust", "list") == 0                    # nothing stored: nothing pinned
    assert main("--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE,
                "--json", "login") == 0
    assert json.loads(capsys.readouterr().out)["groups"] == 1
    from meshbay_cli.client.store import Store
    store = Store()
    store.load()
    store.state["pins"] = {"node-1": "pk-1"}
    store.save()
    assert main("--json", "trust", "list") == 0
    assert json.loads(capsys.readouterr().out) == {"node-1": "pk-1"}
    assert main("trust", "reset") == 1
    assert main("trust", "reset", "node-2") == 3
    assert main("trust", "reset", "node-1") == 0
    assert main("--json", "trust", "list") == 0
    capsys.readouterr()
    assert main("--json", "logout") == 0
    assert json.loads(capsys.readouterr().out) == {"signed_out": True}


def test_a_usage_error_is_exit_1_not_the_sign_in_code(capsys):
    """argparse's own 2 would read as a refused sign-in to a script."""
    assert cli.main(["ls", "--no-such-option"]) == 1
    assert cli.main(["no-such-action"]) == 1



def test_account_shows_the_account(hub, capsys):
    assert main("--hub", "https://hub.test", "-u", USERNAME, "--password", PASSPHRASE,
                "login") == 0
    capsys.readouterr()
    assert main("account") == 0
    out = capsys.readouterr().out
    assert f"Account:  {USERNAME}" in out and "https://hub.test" in out
    assert main("--json", "account", "show") == 0
    assert json.loads(capsys.readouterr().out)["username"] == USERNAME


def test_completion_prints_each_shells_script(capsys):
    assert main("completion", "bash") == 0
    assert "complete -F _meshbay_cli meshbay-cli" in capsys.readouterr().out
    assert main("completion", "powershell") == 0
    assert "Register-ArgumentCompleter -Native" in capsys.readouterr().out
    assert main("completion", "fish") == 1


def test_help_prints_what_help_prints(capsys):
    assert cli.main(["--help"]) == 0
    whole = capsys.readouterr().out
    assert main("help") == 0
    assert capsys.readouterr().out == whole
    assert cli.main(["chat", "--help"]) == 0
    chat = capsys.readouterr().out
    assert main("help", "chat") == 0
    assert capsys.readouterr().out == chat
    assert main("help", "no-such-action") == 1
    assert "No action 'no-such-action'" in capsys.readouterr().err


def test_groups_lists_each_group_with_its_state(hub, capsys):
    hub.groups.append({"id": "g2", "name": "Club/Photos", "owner_username": "bob-test",
                       "node_online": False, "visibility": "public", "is_admin": False,
                       "hosted": True, "last_activity_at": "2026-10-01T12:00:00+00:00"})
    hub.groups.append({"id": "g3", "name": "Draft\x1b[31m", "owner_username": USERNAME,
                       "visibility": "private", "is_admin": True, "hosted": False})
    assert main("groups") == 2                           # signed in nowhere
    capsys.readouterr()
    assert main("--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE, "login") == 0
    capsys.readouterr()

    assert main("groups") == 0
    header, *lines = capsys.readouterr().out.splitlines()
    assert header.split() == ["NODE", "ROLE", "VISIBILITY", "LAST", "ACTIVITY", "GROUP"]
    # In name order; a "/" in a name is encoded so the path can be typed back.
    assert lines[0].split()[:3] == ["down", "member", "public"]
    assert lines[0].endswith("Club%2FPhotos@bob-test")
    assert "2026-10-0" in lines[0]
    assert lines[1].split()[:3] == ["none", "owner", "private"]
    assert "\x1b" not in lines[1] and lines[1].endswith(f"Draft\\x1b[31m@{USERNAME}")
    assert lines[2].split()[:3] == ["up", "owner", "private"]

    assert main("--json", "groups") == 0
    rows = json.loads(capsys.readouterr().out)
    assert [(r["group"], r["node"], r["role"]) for r in rows] == [
        ("Club/Photos", "down", "member"), ("Draft\x1b[31m", "none", "owner"),
        ("Family", "up", "owner")]
    assert rows[0]["path"] == "Club%2FPhotos@bob-test"


class _Terminal(io.StringIO):
    """stdin at a terminal, with the answers a person types."""

    def isatty(self):
        return True


def _signed_in(hub, capsys):
    assert main("--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE, "login") == 0
    capsys.readouterr()


CLUB = {"id": "g2", "name": "Club/Photos", "owner_username": "bob-test",
        "node_online": True, "visibility": "private", "is_admin": False, "hosted": True}


@pytest.fixture
def joined(monkeypatch):
    """`join`'s node half, recorded: the code and the node each try was given."""
    from meshbay_cli.client.session import NodeRefusal
    from meshbay_cli.commands import join
    tries = []
    state = {"needs_code": False, "away": False}

    async def open_group(account, group, *, join_code=None, node_pk=None):
        tries.append({"group": group["id"], "code": join_code, "node_pk": node_pk})
        if state["away"]:
            from meshbay_cli.errors import NODE, CliError
            raise CliError(NODE, "No node serving it is online.")
        if state["needs_code"] and not join_code:
            raise NodeRefusal("this node has not admitted you yet", "code_required")
        return FakeSession()
    monkeypatch.setattr(join, "open_group", open_group)
    return tries, state


def test_an_invitation_is_listed_joined_and_named_in_groups(hub, joined, capsys):
    tries, _ = joined
    hub.pending.append({**CLUB, "invited_by": "bob-test"})
    _signed_in(hub, capsys)

    assert main("groups") == 0
    assert "1 invitation waiting for your answer" in capsys.readouterr().err
    assert main("invitations") == 0
    header, line = capsys.readouterr().out.splitlines()
    assert header.split() == ["INVITED", "BY", "DATE", "GROUP"]
    assert line.startswith("bob-test") and line.endswith("Club%2FPhotos@bob-test")

    assert main("join", "Club%2FPhotos") == 0
    out = capsys.readouterr()
    assert "Joined Club%2FPhotos@bob-test" in out.out
    assert tries == [{"group": "g2", "code": None, "node_pk": None}]
    # The hub keeps no word on browser access: said once, nothing done.
    assert "meshbay-cli browser-access on" in out.err
    assert [g["id"] for g in hub.groups] == ["g1", "g2"] and not hub.pending

    assert main("invitations") == 0
    assert "No invitation is waiting" in capsys.readouterr().err
    assert main("join", "Nothing") == 3


def test_a_node_that_asks_for_a_code_gets_it_from_the_option_or_the_terminal(
        hub, joined, monkeypatch, capsys):
    tries, state = joined
    state["needs_code"] = True
    hub.pending.append({**CLUB, "invited_by": "bob-test"})
    _signed_in(hub, capsys)

    monkeypatch.setattr("sys.stdin.isatty", lambda: False)
    assert main("join", "Club%2FPhotos") == 5          # a member now, not admitted
    err = capsys.readouterr().err
    assert "join Club%2FPhotos@bob-test --code CODE" in err

    assert main("join", "Club%2FPhotos", "--code", "k7p2-9wqx") == 0
    assert tries[-1]["code"] == "K7P2-9WQX"

    monkeypatch.setattr("sys.stdin", _Terminal("ab12-cd34\n"))
    assert main("join", "Club%2FPhotos@bob-test") == 0
    assert tries[-1]["code"] == "AB12-CD34"


def test_an_invitation_link_goes_to_its_hub_and_its_code_to_its_node(hub, joined, capsys):
    tries, state = joined
    ticket, node = "AbCdEfGhIjKlMnOpQr-_12", "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8"
    link = (f"{hub.url}/#/invite?v=1&g=0f8fad5b-d9cb-469f-a165-70867728950e&t={ticket}"
            f"&n={node}&c=k7p2-9wqx")
    hub.links[ticket] = {**CLUB, "id": "0f8fad5b-d9cb-469f-a165-70867728950e"}
    _signed_in(hub, capsys)

    assert main("join", link.replace(hub.url, "https://elsewhere.test")) == 1
    assert "--hub https://elsewhere.test" in capsys.readouterr().err

    state["away"] = True                     # redeemed, but the node is not there
    assert main("join", link) == 5
    assert "The code is kept" in capsys.readouterr().err
    assert tries[-1] == {"group": "0f8fad5b-d9cb-469f-a165-70867728950e",
                         "code": "K7P2-9WQX",
                         "node_pk": "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="}
    from meshbay_cli.client.store import Store
    store = Store()
    store.load()
    assert store.secrets["join_codes"]["0f8fad5b-d9cb-469f-a165-70867728950e"] == {
        "code": "K7P2-9WQX", "node_pk": "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="}

    assert main("join", link) == 3                  # spent: the hub says so
    assert "no longer works" in capsys.readouterr().err


def test_declining_and_leaving_ask_first(hub, monkeypatch, capsys):
    hub.pending.append({**CLUB, "invited_by": "bob-test"})
    hub.groups.append({**CLUB, "id": "g3", "name": "Choir"})
    _signed_in(hub, capsys)

    monkeypatch.setattr("sys.stdin.isatty", lambda: False)
    assert main("decline", "Club%2FPhotos") == 1        # no terminal, no --yes
    assert "pass --yes" in capsys.readouterr().err and hub.pending
    assert main("leave", "Choir") == 1
    assert any(g["id"] == "g3" for g in hub.groups)
    assert main("leave", "Family") == 8                 # the owner, before any question
    assert "You own" in capsys.readouterr().err

    monkeypatch.setattr("sys.stdin", _Terminal("n\n"))
    assert main("leave", "Choir") == 1
    assert any(g["id"] == "g3" for g in hub.groups)
    monkeypatch.setattr("sys.stdin", _Terminal("y\n"))
    assert main("leave", "Choir") == 0
    assert "Left Choir@bob-test" in capsys.readouterr().out
    assert not any(g["id"] == "g3" for g in hub.groups)

    assert main("decline", "--yes", "Club%2FPhotos") == 0
    assert not hub.pending and not any(g["id"] == "g2" for g in hub.groups)


def test_devices_are_listed_approved_by_code_and_revoked(node, monkeypatch, capsys):
    from meshbay_cli.client.groups import open_group  # noqa: F401  (patched below)
    from meshbay_cli.commands import devices

    async def open_group_(account, group):
        return node.session()
    monkeypatch.setattr(devices, "open_group", open_group_)

    other = node.request_device("ZX4P-7QRT")
    assert main("devices", "list", "Family") == 0
    out = capsys.readouterr()
    assert out.out.splitlines()[1].startswith(f"* {node.identity.pk_ed[:16]}")
    assert "1 device waiting for approval" in out.err

    assert main("devices", "approve", "Family", "zx4p-7qrz") == 3     # no request has it
    assert main("devices", "approve", "Family", "ZX4P 7QRT") == 0
    assert other.pk_ed in [d["pk_ed25519"] for d in node.devices]

    # The same code filed twice: neither is approved.
    node.request_device("AAAA-BBBB")
    node.request_device("AAAA-BBBB")
    assert main("devices", "approve", "Family", "AAAA-BBBB") == 3
    assert "none was approved" in capsys.readouterr().err
    assert len(node.devices) == 2

    monkeypatch.setattr("sys.stdin.isatty", lambda: False)
    assert main("devices", "revoke", "Family", node.identity.pk_ed[:12], "--yes") == 8
    assert main("devices", "revoke", "Family", other.pk_ed[:12]) == 1     # not confirmed
    assert main("devices", "revoke", "Family", other.pk_ed[:12], "--yes") == 0
    assert [d["pk_ed25519"] for d in node.devices] == [node.identity.pk_ed]
    assert main("devices", "approve", "Family") == 1


def test_browser_access_is_set_here_and_mirrored_never_taken_from_the_hub(
        hub, joined, capsys):
    hub.prefs["browser_access"] = "on"         # anyone with a session can write it
    _signed_in(hub, capsys)
    assert main("--json", "browser-access") == 0
    assert json.loads(capsys.readouterr().out) == {"this_machine": None, "hub": "on"}
    from meshbay_cli.client.store import Store
    store = Store()
    store.load()
    assert "browser_access" not in store.state

    assert main("browser-access", "off") == 0
    assert hub.prefs["browser_access"] == "off"
    store.load()
    assert store.state["browser_access"] == "off"
    # Set here: `join` has nothing to point out.
    hub.pending.append({**CLUB, "invited_by": "bob-test"})
    assert main("join", "Club%2FPhotos") == 0
    assert "browser-access" not in capsys.readouterr().err

    # Kept through a new sign-in of the same account.
    _signed_in(hub, capsys)
    store.load()
    assert store.state["browser_access"] == "off"


def test_rm_deletes_only_what_this_account_may_and_asks_first(node, monkeypatch, capsys):
    from fake_hub import USER_ID
    mine = node.add("share", "mine.txt", b"1", uploader_id=USER_ID)
    node.add("share", "also-mine.txt", b"2", uploader_id=USER_ID)
    theirs = node.add("share", "theirs.txt", b"3", uploader_id="someone-else")
    node.dirs.add("share/sub")

    # Somebody else's file, a folder: refused whole, before any question.
    assert main("rm", "--yes", "Family/share/mine.txt", "Family/share/theirs.txt") == 8
    assert "Family@alice-test/share/theirs.txt" in capsys.readouterr().err
    assert main("rm", "--yes", "Family/share/sub") == 1
    assert main("rm", "--yes", "Family/share/nothing*") == 3
    assert mine["id"] in node.files

    monkeypatch.setattr("sys.stdin.isatty", lambda: False)
    assert main("rm", "Family/share/mine.txt") == 1                  # not confirmed
    monkeypatch.setattr("sys.stdin", _Terminal("n\n"))
    assert main("rm", "Family/share/mine.txt") == 1
    assert mine["id"] in node.files
    monkeypatch.setattr("sys.stdin", _Terminal("y\n"))
    assert main("rm", "Family/share/*mine.txt") == 0
    assert "Deleted 2 files" in capsys.readouterr().err
    assert set(node.files) == {theirs["id"]}

    # From the node's operator key, anyone's file.
    node.role = "operator"
    assert main("--json", "rm", "--yes", "Family/share/theirs.txt") == 0
    assert json.loads(capsys.readouterr().out)["deleted"] == [
        "Family@alice-test/share/theirs.txt"]
    assert not node.files


def test_rm_r_removes_a_folder_and_all_it_holds_from_the_operator_key(node, capsys):
    from fake_hub import USER_ID
    node.add("share/trip/day 1", "a.jpg", b"A" * 2000, uploader_id=USER_ID)
    node.add("share/trip/day 2", "b.jpg", b"B" * 1000, uploader_id="someone-else")
    node.add("share/trip", "notes.txt", b"n", uploader_id=USER_ID)
    node.dirs |= {"share/trip/empty", "share/kept"}
    kept = node.add("share/kept", "c.txt", b"c", uploader_id=USER_ID)

    # Not the operator: refused whole, nothing deleted, not even its own files.
    assert main("rm", "-r", "--yes", "Family/share/trip") == 8
    assert "operator" in capsys.readouterr().err
    assert len(node.files) == 4
    # A root is never removed.
    node.role = "operator"
    assert main("rm", "-r", "--yes", "Family/share") == 8
    assert "root" in capsys.readouterr().err

    assert main("--json", "rm", "-r", "--yes", "Family/share/trip") == 0
    out = json.loads(capsys.readouterr().out)
    assert len(out["deleted"]) == 3 and out["failed"] == []
    assert out["folders_removed"][-1] == "Family@alice-test/share/trip"   # deepest first
    assert {"Family@alice-test/share/trip/empty", "Family@alice-test/share/trip/day 1",
            "Family@alice-test/share/trip/day 2"} <= set(out["folders_removed"])
    assert set(node.files) == {kept["id"]}
    assert not any(d == "share/trip" or d.startswith("share/trip/") for d in node.dirs)


def test_rm_r_says_what_it_will_delete_before_it_asks(node, monkeypatch, capsys):
    node.role = "operator"
    for i in range(12):
        node.add("share/big", f"f{i}.bin", b"x" * 1024)
    monkeypatch.setattr("sys.stdin", _Terminal("n\n"))
    assert main("rm", "-r", "Family/share/big") == 1
    asked = capsys.readouterr().err
    assert "Delete 12 files (12.0 KB) and 1 folder" in asked
    assert "Family@alice-test/share/big/" in asked
    assert len(node.files) == 12


def test_mkdir_makes_folders_where_the_root_accepts_them(node, capsys):
    assert main("mkdir", "Family/share/2026") == 0
    assert "share/2026" in node.dirs
    assert main("mkdir", "Family/share/2026") == 8                 # already there
    assert main("mkdir", "Family/share/a/b/c") == 3                # no parent
    assert "--parents makes it too" in capsys.readouterr().err
    assert main("--json", "mkdir", "--parents", "Family/share/a/b/c", "Family/share/2026") == 0
    assert json.loads(capsys.readouterr().out)["made"] == [
        "Family@alice-test/share/a", "Family@alice-test/share/a/b",
        "Family@alice-test/share/a/b/c"]
    node.add("share", "Notes.txt", b"n")
    assert main("mkdir", "Family/share/notes.TXT") == 8           # a file, without case
    assert "(a file)" in capsys.readouterr().err
    assert main("mkdir", "Family/share/bad\x01name") == 8
    assert main("mkdir", "Family/share") == 1                      # a root is not made here
    node.roots[0]["writable"] = False
    assert main("mkdir", "Family/share/x") == 8
    assert "read-only" in capsys.readouterr().err


def test_search_finds_across_groups_once_per_file(hub, monkeypatch, capsys):
    from fake_node import FakeNode
    from meshbay_cli.client.session import DeviceLinkPending
    family, club, hidden = FakeNode(), FakeNode(), FakeNode()
    hub.groups += [{**CLUB, "id": "g2"}, {**CLUB, "id": "g3", "name": "Album"},
                   {**CLUB, "id": "g4", "name": "Away"}, {**CLUB, "id": "g5", "name": "New"}]
    nodes = {"g1": family, "g2": club, "g3": hidden}
    hidden.ack["search_listed"] = False
    club.ack["is_node_admin"] = True

    async def open_group(account, group):
        if group["id"] == "g4":
            from meshbay_cli.errors import NODE, CliError
            raise CliError(NODE, "No node serving it is online.")
        if group["id"] == "g5":
            raise DeviceLinkPending("ABCD-EFGH", 0)
        return nodes[group["id"]].session()
    monkeypatch.setattr(pool, "open_group", open_group)
    _signed_in(hub, capsys)

    family.add("share/Holidays 2024", "Beach.JPG", b"same bytes")
    club.add("share/Holidays 2024", "Beach.JPG", b"same bytes")     # the same file
    track = family.add("share/music", "track.flac", b"other")
    track.update(type="audio", artist="The Holiday Band")
    family.add("share", "notes 100%.txt", b"n")
    hidden.add("share", "holiday-secret.jpg", b"s")

    assert main("search", "HOLIDAY") == 4               # two groups not searched
    out = capsys.readouterr()
    # The beach photo once, from the reader's own node first; the track by its tag.
    assert out.out.splitlines() == [
        "Club%2FPhotos@bob-test/share/Holidays 2024/Beach.JPG",
        f"Family@{USERNAME}/share/music/track.flac"]
    assert "holiday-secret" not in out.out
    assert "not searched: Away@bob-test" in out.err
    assert "ABCD-EFGH" in out.err

    assert main("--json", "search", "beach", "2024") == 4
    [hit] = json.loads(capsys.readouterr().out)
    assert hit["also"] == [f"Family@{USERNAME}/share/Holidays 2024/Beach.JPG"]

    assert main("search", "-l", "-t", "audio", "holiday") == 4
    line = capsys.readouterr().out.strip()
    assert line.startswith("audio") and line.endswith("track.flac")
    assert main("search", "100%") == 4
    assert capsys.readouterr().out.strip() == f"Family@{USERNAME}/share/notes 100%25.txt"
    assert main("search", "nothing-like-this") == 4

    hub.groups = [g for g in hub.groups if g["id"] in ("g1", "g2")]
    assert main("search", "nothing-like-this") == 3
    assert main("search", "beach") == 0


def test_chat_reads_the_last_messages_and_posts(node, monkeypatch, capsys):
    from meshbay_cli.commands import chat

    async def open_group(account, group):
        return node.session()
    monkeypatch.setattr(chat, "open_group", open_group)

    for i in range(1, 231):                       # more than one page of the node's
        node.say(f"message {i}")
    node.say("forged", forge=True)
    node.say("two lines\nand an \x1b[31m escape")
    node.chat_epochs[2] = b"\x09" * 32
    node.say("from an epoch this member was not given", epoch=2)
    del node.chat_epochs[2]

    assert main("chat", "Family") == 0
    lines = capsys.readouterr().out.splitlines()
    assert len(lines) == 11 and lines[0].endswith("bob-test: message 224")
    assert lines[7].endswith("bob-test: [this message cannot be read: signature]")
    assert lines[8].endswith("bob-test: two lines")
    assert lines[9].strip() == "and an \\x1b[31m escape"            # shown, never run
    assert "cannot be read: no key for its epoch" in lines[10]

    assert main("chat", "-5", "Family") == 0
    assert capsys.readouterr().out.splitlines()[0].endswith("message 229")
    assert main("--json", "chat", "Family", "-n", "300") == 0     # paged back
    got = json.loads(capsys.readouterr().out)
    assert len(got) == 233 and got[0]["text"] == "message 1"
    assert got[230]["unreadable"] == "signature"

    assert main("chat", "Family", "-m", "hello from the terminal") == 0
    assert "Posted to Family@alice-test" in capsys.readouterr().err
    monkeypatch.setattr("sys.stdin", io.StringIO("piped\n"))
    assert main("chat", "Family", "-m", "-") == 0
    capsys.readouterr()
    assert main("chat", "-2", "Family") == 0
    out = capsys.readouterr().out.splitlines()
    assert out[0].endswith(f"{USERNAME}: hello from the terminal")
    assert out[1].endswith(f"{USERNAME}: piped")

    assert main("chat", "Family", "-m", "   ") == 1
    assert main("chat", "Family", "-m", "x" * 70_000) == 1          # over the node's bound
    assert main("chat", "-0", "Family") == 1


def test_stream_serves_a_film_to_the_player_it_starts(node, tmp_path, monkeypatch, capsys):
    from meshbay_cli.commands import stream

    async def open_group(account, group):
        return node.session()
    monkeypatch.setattr(stream, "open_group", open_group)
    film = os.urandom(1024 * 1024 + 7)
    node.add("share/films", "Night Drive.mkv", film)["type"] = "video"
    node.add("share", "notes.txt", b"hello")

    assert main("stream", "Family/share/notes.txt") == 1
    assert "not a video or a sound file" in capsys.readouterr().err
    if sys.platform == "win32":
        return                                  # a script is not a program there
    played = tmp_path / "played"
    player = tmp_path / "player"
    player.write_text(f"#!{sys.executable}\nimport sys, urllib.request\n"
                      "opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))\n"
                      f"open({str(played)!r}, 'wb').write(opener.open(sys.argv[1]).read())\n")
    player.chmod(0o755)
    assert main("stream", "--player", str(player), "Family/share/films/Night Drive.mkv") == 0
    assert played.read_bytes() == film
    out = capsys.readouterr()
    assert out.out.startswith("http://127.0.0.1:") and "player was closed" in out.err


def test_every_action_is_run_here():
    """Runs last in this module: every action the parser knows was run above."""
    assert RUN == set(cli.ACTIONS), f"never run: {set(cli.ACTIONS) - RUN}"