""" 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 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 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 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_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_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}"