diff --git a/docs/cli/README.md b/docs/cli/README.md index 978e460..81caad2 100644 --- a/docs/cli/README.md +++ b/docs/cli/README.md @@ -1281,6 +1281,8 @@ the room). | `add INDEX NAME [--secret HEX]` | Add a channel. A `#` at the start of the name makes the channel public (the firmware derives the key from the name). Otherwise the channel is private, with a random key unless you give a key. | | `join INDEX NAME SECRET` | Join a channel from its name and its key of 32 hex characters. | | `import INDEX URL` | Import a `meshcore://channel/add` link. | +| `import --file FILE [--dry-run]` | Add the channels of a file from `export` to the device. | +| `export FILE` | Write each channel to a file, with its key, its send scope, and its mute. Another device can import the file. | | `share INDEX` | Print the share link of a slot. | | `clear INDEX --yes` | Clear a slot. This removes the channel from the device. | | `scope INDEX [REGION] [--clear]` | Show or set the region into which the channel's messages are flooded. | @@ -1340,6 +1342,55 @@ $ meshterm channels share 1 --json `share` on an empty slot, and `clear` on a slot that is already empty, give exit `5` with the nulls and `"cleared":false`. There is nothing to report, and it is not a failure. +**Export and import.** `export` and `import --file` move all the channels to another +device. `export` writes one TOML file, with one `[[channels]]` table for each channel, in +slot order: + +```toml +[[channels]] +name = "#news" + +[[channels]] +name = "brigade" +secret = "2c70ca416304d520b60a28d9db35d6b3" +scope = "harbour" +muted = true +``` + +A `#` channel has no `secret`, because the device derives its key from the name. A private +channel has its key in the file. Thus the file is as secret as the keys in it. MeshTerm +makes it readable only by you, where the system lets it, and `export` says on stderr how +many keys of private channels the file holds. A file from `config backup` has a +`[[channels]]` table of the same shape, so `import --file` can also read a config backup. + +`import --file` never removes a channel. It puts the channels of the file first, in the +order of the file, and the other channels of the device after them. A channel that has no +free slot is not added. The send scope and the mute of each channel go into the stores of +this computer. A channel that has no scope or no mute in the file keeps the values that it +has here. `--dry-run` prints the plan and changes nothing. + +```console +$ meshterm channels import --file brigade.toml +SLOT NAME ACTION + 0 #news add + 1 brigade add +``` + +`import --file` prints the layout of the device after the import, with one record for each +channel. `ACTION` is `add`, `move`, or `keep`. A channel that has no free slot comes last, +with the action `skip` and no slot. `export` prints only the path of the file. It writes no +file and exits with `5` when the device has no channel. A file that MeshTerm cannot use is +a usage error (exit `2`), and the message names the bad entry. If MeshTerm cannot read +each slot of the device, the command changes nothing and exits with `4`. + +```console +$ meshterm channels import --file brigade.toml --json +{"dry_run":false,"added":2,"moved":0,"skipped":0,"channels":[{"slot":0,"name":"#news","action":"add"},{"slot":1,"name":"brigade","action":"add"}]} + +$ meshterm channels export channels.toml --json +{"path":null,"channels":0,"private":0} +``` + #### `meshterm courier` A store-and-forward outbox. Queue a message for a contact that you cannot reach now. The diff --git a/docs/guide/features.md b/docs/guide/features.md index cf305aa..c9e08a4 100644 --- a/docs/guide/features.md +++ b/docs/guide/features.md @@ -16,7 +16,7 @@ This page lists each interactive screen with its scripted equivalent, where one | Feature | What it does | Scripted | | --- | --- | --- | | **💬 Chat** | Live full-screen messages on channels and to contacts. The transcript scrolls, the input line stays at the bottom, and sent and received messages appear together. The rooms that you joined are also here. Each post shows under the name of its author. MeshTerm stores each message. Unread counts show in the menu header. | `meshterm chat send / history / list / listen` | -| **📻 Channels** | Create, join, reorder, mute, and share mesh channels. You can share a channel with a QR code or a `meshcore://` link. If you give a channel a **send scope**, its messages are flooded into one region. Only the repeaters that carry that region relay them. | `meshterm channels list / add / join / import / share / clear / scope` | +| **📻 Channels** | Create, join, reorder, mute, and share mesh channels. You can share a channel with a QR code or a `meshcore://` link. If you give a channel a **send scope**, its messages are flooded into one region. Only the repeaters that carry that region relay them. When you change to a new device, export all the channels to a file, then import the file on the new device. | `meshterm channels list / add / join / import / share / clear / scope / export` | | **📌 Rooms** | Each room server that your radio knows. A room server is a message board on a radio. It keeps the latest posts and sends you the posts that you missed. Join a room with its password. If a room does not answer, MeshTerm tells you what it knows and offers the next step to try. For example, it can send again by flood when the route that your radio learned is no longer good. | `meshterm rooms list / join / forget` | | **📨 Courier** | A store-and-forward outbox for contacts that you cannot reach yet. Queue a message. MeshTerm sends it when the contact is next heard, or at a time that you set. It tracks the acks, and it waits longer between each try (exponential backoff). | `meshterm courier queue / list / send / cancel / clear` | | **👥 Contacts** | This node and its known contacts. The screen shows a heat map of how recently each contact was heard, the counts of overheard packets, and the full public keys. The hash of each key is highlighted. The scripted listing shows only the contacts. For this node, run `meshterm info`, which gives much more detail. | `meshterm contacts` | diff --git a/meshterm/core/channel_file.py b/meshterm/core/channel_file.py new file mode 100644 index 0000000..d864ed2 --- /dev/null +++ b/meshterm/core/channel_file.py @@ -0,0 +1,402 @@ +# SPDX-License-Identifier: Apache-2.0 +"""The channel file: export the channels of a device, and import them onto another device. + +When the user changes to a new device, the channels must come too. The new device ships +with only ``Public``, and each other channel must be added again by hand: each private key +pasted, each ``#`` name typed, each send scope and mute set again. A channel file holds all +of these things, so that one import sets up the new device. + +The file is TOML, with one ``[[channels]]`` table for each channel, in slot order: + +.. code-block:: toml + + [[channels]] + name = "Public" + secret = "8b3387e9c5cdea6ac9e5edbaa115cd72" + + [[channels]] + name = "#bots" # the key comes from the name, so the file has no secret + + [[channels]] + name = "Family" + secret = "…32 hex digits…" + scope = "yul" + muted = true + +The table has the same shape as the ``[[channels]]`` table of ``config backup`` +(:mod:`meshterm.core.config_io`). Thus a config backup is also a channel file. Its +``index`` field is not read, because the order of the tables gives the order of the slots. + +An import adds channels and puts them in order. It never removes a channel from the device. +A channel that is on the device and not in the file stays, after the channels of the file. +A private channel that MeshTerm removed would lose its key, if no other place has a copy of +that key. Thus a removal stays a decision that the user makes on each channel, with its own +confirm. + +This module is pure logic: it reads and writes files, and it plans. The channel manager +(:mod:`meshterm.ui.channels`) and the ``channels`` CLI do the writes to the device. +""" + +from __future__ import annotations + +import sys +from collections.abc import Iterable, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Protocol + +import tomli_w + +from .atomicwrite import write_atomically +from .channels import channel_identity, derive_secret, is_name_derived, normalize_secret +from .regions import RegionNameError +from .regions import validate as validate_region + +if sys.version_info >= (3, 11): + import tomllib +else: # pragma: no cover - exercised only on 3.10 + import tomli as tomllib + + +#: The comment at the top of an exported file. A user who opens the file finds what it is, +#: how to use it, and that it holds keys. +_HEADER = ( + "# MeshTerm channels. Import this file on the Channels page, or with\n" + "# meshterm channels import --file \n" + "# It holds the keys of private channels. Keep it private.\n\n" +) + + +class ChannelFileError(ValueError): + """A channel file that MeshTerm cannot use: not readable, not TOML, or a bad entry.""" + + +class SlotLike(Protocol): + """A channel on the device, as the plan reads it. + + The device gives a :class:`~meshterm.core.channel_probe.ChannelSlot`. + """ + + idx: int + name: str + secret: bytes + + @property + def identity(self) -> str: + """The identity of the channel, which does not depend on the slot.""" + ... + + +@dataclass(frozen=True, slots=True) +class ChannelEntry: + """One channel in a channel file. + + Attributes: + name: The channel name. + secret: The 16-byte key, or ``None`` when the key comes from the name (a ``#`` + channel). Then the device calculates the key itself. + scope: The send scope of the channel, or ``None`` for the default scope. + muted: Whether the notifications of the channel are muted. + """ + + name: str + secret: bytes | None = None + scope: str | None = None + muted: bool = False + + @property + def key(self) -> bytes: + """The key of the channel: the stored key, or the key that comes from the name.""" + return self.secret if self.secret is not None else derive_secret(self.name) + + @property + def identity(self) -> str: + """The identity of the channel. It is the same value that the slot of the device gives.""" + return channel_identity(self.name, self.key) + + @property + def write_secret(self) -> bytes | None: + """The secret to send with a write: ``None`` when the device can derive the key. + + This is the same rule that a reorder uses (``_write_slot`` in + :mod:`meshterm.ui.channels`). + """ + return None if is_name_derived(self.name, self.key) else self.key + + +def entry_from_slot( + slot: SlotLike, *, scope: str | None = None, muted: bool = False +) -> ChannelEntry: + """The file entry for one channel of the device. + + Args: + slot: The channel, as the device reports it. + scope: The send scope of the channel (from the region store). + muted: Whether the channel is muted (from the mute store). + + Returns: + The entry. Its secret is ``None`` when the key comes from the name. + """ + secret = None if is_name_derived(slot.name, slot.secret) else bytes(slot.secret) + return ChannelEntry(name=slot.name, secret=secret, scope=scope, muted=muted) + + +def write_channel_file(path: Path, entries: Iterable[ChannelEntry]) -> Path: + """Write the channels to a TOML file, in the order of ``entries``. + + The file holds the keys of private channels. Thus the write limits the file to its + owner, where the platform lets it (refer to + :func:`~meshterm.core.atomicwrite.write_atomically`). + + Args: + path: The destination file. The function makes the parent directories. + entries: The channels to write. + + Returns: + The path that the function wrote. + """ + tables: list[dict[str, Any]] = [] + for entry in entries: + table: dict[str, Any] = {"name": entry.name} + if entry.secret is not None: + table["secret"] = entry.secret.hex() + if entry.scope: + table["scope"] = entry.scope + if entry.muted: + table["muted"] = True + tables.append(table) + write_atomically(path, _HEADER + tomli_w.dumps({"channels": tables}), owner_only=True) + return path + + +def read_channel_file(path: Path) -> list[ChannelEntry]: + """Read a channel file, and check each entry. + + The function removes a second copy of a channel (the same identity), and keeps the + first copy. A field that the function does not know (the ``index`` of a config backup) + is not read. + + Args: + path: The file to read. + + Returns: + The channels of the file, in file order. + + Raises: + ChannelFileError: If the file cannot be read, is not TOML, has no channels, or has + an entry that is not valid. The message names the entry. + """ + try: + raw = path.read_bytes() + except FileNotFoundError: + raise ChannelFileError(f"no such file: {path}") from None + except OSError as exc: + raise ChannelFileError(f"cannot read {path}: {exc.strerror or exc}") from None + try: + doc = tomllib.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, tomllib.TOMLDecodeError) as exc: + raise ChannelFileError(f"{path.name} is not a TOML file: {exc}") from None + tables = doc.get("channels") + if not isinstance(tables, list) or not tables: + raise ChannelFileError(f"{path.name} has no [[channels]] table") + entries: list[ChannelEntry] = [] + seen: set[str] = set() + for number, table in enumerate(tables, start=1): + entry = _entry(number, table) + if entry.identity not in seen: + seen.add(entry.identity) + entries.append(entry) + return entries + + +def _entry(number: int, table: object) -> ChannelEntry: + """Check one ``[[channels]]`` table, and make its entry. + + Raises: + ChannelFileError: If the table is not valid. The message names the entry by its + number (from 1) and by its name, when it has one. + """ + if not isinstance(table, dict): + raise ChannelFileError(f"channel {number} is not a table") + name = table.get("name") + if not isinstance(name, str) or not name.strip(): + raise ChannelFileError(f"channel {number} has no name") + name = name.strip() + where = f"channel {number} ({name})" + raw_secret = table.get("secret") + secret: bytes | None = None + if raw_secret not in (None, ""): + if not isinstance(raw_secret, str): + raise ChannelFileError(f"{where}: the secret must be a string of hex digits") + try: + secret = normalize_secret(raw_secret) + except ValueError as exc: + raise ChannelFileError(f"{where}: {exc}") from None + if is_name_derived(name, secret): + secret = None # the device derives it. A config backup writes it for each channel + elif not name.startswith("#"): + # Only a # name gives its key. Any other name with no secret is a private channel + # whose key is not in the file, and no device can join it. + raise ChannelFileError(f"{where}: a private channel needs its secret") + scope = table.get("scope") + if scope not in (None, ""): + if not isinstance(scope, str): + raise ChannelFileError(f"{where}: the scope must be a region name") + try: + scope = validate_region(scope) + except RegionNameError as exc: + raise ChannelFileError(f"{where}: {exc}") from None + else: + scope = None + muted = table.get("muted", False) + if not isinstance(muted, bool): + raise ChannelFileError(f"{where}: muted must be true or false") + return ChannelEntry(name=name, secret=secret, scope=scope, muted=muted) + + +@dataclass(frozen=True, slots=True) +class ImportPlan: + """What an import does to one device. + + Attributes: + writes: The slot writes, in order. Each is ``(slot, name, secret)``, with the + arguments of ``Device.set_channel``. An empty name clears a slot whose channel + moved to a lower slot. + added: The channels of the file that the device does not have, and that fit, each + with the slot that the plan writes it to. + skipped: The channels of the file that the device does not have, and that do not + fit, because each slot is full. + present: The channels of the file that the device already has. + layout: Each channel on the device after the import, in slot order, as + ``(slot, name, action)``. The action is ``add``, ``move``, or ``keep``. + """ + + writes: tuple[tuple[int, str, bytes | None], ...] + added: tuple[tuple[int, ChannelEntry], ...] + skipped: tuple[ChannelEntry, ...] + present: tuple[ChannelEntry, ...] + layout: tuple[tuple[int, str, str], ...] + + @property + def moved(self) -> int: + """The number of channels that are already on the device and change slot.""" + return sum(1 for _, _, action in self.layout if action == "move") + + @property + def changes_device(self) -> bool: + """Whether the import writes anything to the device.""" + return bool(self.writes) + + +def plan_import( + entries: Sequence[ChannelEntry], slots: Sequence[SlotLike], capacity: int +) -> ImportPlan: + """Plan the import of a channel file onto a device. + + The device gets this order: first the channels of the file, in file order, then each + channel that the device has and the file does not have, in its current order. The order + fills the slots from slot 0. The plan writes only a slot whose channel changes. When the + device has gaps (for example, channels in slots 0, 1, and 5), the channels move down to + close them, and the plan clears each higher slot whose channel moved down. Otherwise the + device holds that channel two times. + + The plan never removes a channel, so the device keeps each channel that it had. A file + channel that does not fit (each slot is full) is in ``skipped``. The plan keeps the + first channels of the file and skips the last ones. + + Args: + entries: The channels of the file, in file order, with no second copies (as + :func:`read_channel_file` returns them). + slots: The channels of the device, as a complete probe reads them. + capacity: The number of channel slots of the device. + + Returns: + The plan. + """ + current = sorted(slots, key=lambda s: s.idx) + on_device = {s.identity for s in current} + in_file = {e.identity for e in entries} + present = [e for e in entries if e.identity in on_device] + new = [e for e in entries if e.identity not in on_device] + room = max(0, capacity - len(current)) + added, skipped = new[:room], new[room:] + placed = {e.identity for e in present} | {e.identity for e in added} + + # The order to write: (identity, name, secret to send) for each slot from 0. + order: list[tuple[str, str, bytes | None]] = [ + (e.identity, e.name, e.write_secret) for e in entries if e.identity in placed + ] + order += [ + (s.identity, s.name, None if is_name_derived(s.name, s.secret) else bytes(s.secret)) + for s in current + if s.identity not in in_file + ] + + occupant = {s.idx: s.identity for s in current} + writes: list[tuple[int, str, bytes | None]] = [ + (idx, name, secret) + for idx, (identity, name, secret) in enumerate(order) + if occupant.get(idx) != identity + ] + # Each channel of the device is in ``order``. Thus a channel in a slot past the end of + # ``order`` is now also in a lower slot, and this copy must go. + writes += [(s.idx, "", None) for s in current if s.idx >= len(order)] + new_slot = {identity: idx for idx, (identity, _, _) in enumerate(order)} + old_slot = {s.identity: s.idx for s in current} + layout = tuple( + ( + idx, + name, + "add" if identity not in old_slot else "keep" if old_slot[identity] == idx else "move", + ) + for idx, (identity, name, _) in enumerate(order) + ) + return ImportPlan( + writes=tuple(writes), + added=tuple((new_slot[e.identity], e) for e in added), + skipped=tuple(skipped), + present=tuple(present), + layout=layout, + ) + + +class _Scopes(Protocol): + def channel_scope(self, channel_id: str | None) -> str | None: ... + def set_channel_scope(self, channel_id: str, name: str | None) -> None: ... + + +class _Mutes(Protocol): + def is_muted(self, channel_id: str | None) -> bool: ... + def set_muted(self, channel_id: str, muted: bool) -> None: ... + + +def apply_preferences( + entries: Iterable[ChannelEntry], scopes: _Scopes | None, mutes: _Mutes | None +) -> int: + """Set the send scope and the mute that the file gives each channel. + + MeshTerm keeps these two values, not the device (refer to + :class:`~meshterm.core.region_store.RegionStore` and + :class:`~meshterm.core.mute_store.MuteStore`). Their key is the identity of the + channel, so they apply to the channel in any slot, on any device. The function only + sets what the file gives. A channel with no scope or no mute in the file keeps the + values that this computer has for it, because an import adds and never removes. + + Args: + entries: The channels whose values to set. + scopes: The region store, or ``None``. + mutes: The mute store, or ``None``. + + Returns: + The number of values that changed. + """ + changed = 0 + for entry in entries: + if entry.scope and scopes is not None: + if scopes.channel_scope(entry.identity) != entry.scope: + scopes.set_channel_scope(entry.identity, entry.scope) + changed += 1 + if entry.muted and mutes is not None and not mutes.is_muted(entry.identity): + mutes.set_muted(entry.identity, True) + changed += 1 + return changed diff --git a/meshterm/tools/channels.py b/meshterm/tools/channels.py index 8fd79f0..84908e6 100644 --- a/meshterm/tools/channels.py +++ b/meshterm/tools/channels.py @@ -6,7 +6,9 @@ In the menu, the tool opens a full-screen channel manager (refer to With it, the user can create a private channel, add a public ``#`` channel, join a channel with a key, import a scanned ``meshcore://`` link, and share any channel as a QR code. On the CLI, the tool gives the ``list``, ``add``, ``join``, ``import``, ``share``, ``clear``, -and ``scope`` subcommands for use in scripts. +``scope``, and ``export`` subcommands for use in scripts. ``export`` writes all the +channels to a file, and ``import --file`` adds the channels of that file to another device +(refer to :mod:`meshterm.core.channel_file`). The ``chat`` tool also lists the channels, so that the user can select a conversation. But this tool owns all the configuration of the slots. @@ -14,6 +16,7 @@ this tool owns all the configuration of the slots. from __future__ import annotations +from pathlib import Path from typing import TYPE_CHECKING, Any import typer @@ -21,6 +24,7 @@ import typer from ..context import AppContext from ..core import exitcodes from ..core.channels import ( + MAX_CHANNELS, channel_hash, derive_secret, is_public_channel, @@ -84,6 +88,10 @@ class ChannelsTool(Tool): return await self._cli_clear(ctx, params) if action == "scope": return await self._cli_scope(ctx, params) + if action == "export": + return await self._cli_export(ctx, params) + if action == "import_file": + return await self._cli_import_file(ctx, params) return await self._cli_list(ctx) async def _cli_list(self, ctx: AppContext) -> ToolResult: @@ -246,6 +254,90 @@ class ChannelsTool(Tool): report=(_scoped(idx, slot.name, slot.secret, scope, changed=changed),), ) + async def _cli_export(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Write all the channels of the device to a file, for ``channels import --file``. + + The plain face prints only the path, as ``config backup`` does. The warning about + the keys in the file goes to stderr, next to the path on the screen and not in the + output. A device with no channels writes no file and exits 5, because there is + nothing to export. + """ + from ..core.channel_file import entry_from_slot, write_channel_file + + _, slots = await _complete_slots(ctx) + if not slots: + return ToolResult( + summary={"channels": 0}, + report=(_exported(None, 0, 0),), + exit_code=exitcodes.NO_RESULT, + ) + regions, mutes = ctx.region_store, ctx.mute_store + entries = [ + entry_from_slot( + slot, + scope=regions.channel_scope(slot.identity) if regions is not None else None, + muted=mutes.is_muted(slot.identity) if mutes is not None else False, + ) + for slot in slots + ] + path = write_channel_file(Path(params["path"]).expanduser().resolve(), entries) + private = sum(1 for slot in slots if not slot.is_public) + if private: + noun = "channel" if private == 1 else "channels" + ctx.ui.ack( + f"[warn]the file holds the keys of {private} private {noun}: keep it private[/warn]" + ) + return ToolResult( + summary={"channels": len(entries)}, + report=(_exported(path, len(entries), private),), + ) + + async def _cli_import_file(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Add the channels of a file from ``channels export`` to the device. + + This is the script form of "Import channels…" on the Channels page, with the same + plan (:func:`~meshterm.core.channel_file.plan_import`): the channels of the file + first, in file order, then the other channels of the device. Nothing is removed. + The answer is the layout of the device after the import, one row for each channel, + with what the import does to it. A channel that has no free slot is a row with no + slot. ``--dry-run`` gives the same answer and writes nothing. + """ + from ..core.channel_file import ( + ChannelFileError, + apply_preferences, + plan_import, + read_channel_file, + ) + from ..ui.channels import write_channel + + try: + entries = read_channel_file(Path(params["file"]).expanduser()) + except ChannelFileError as exc: + raise typer.BadParameter(str(exc)) from exc + device, slots = await _complete_slots(ctx) + capacity = await ctx.devstate.channel_capacity() or MAX_CHANNELS + plan = plan_import(entries, slots, capacity) + dry_run = bool(params.get("dry_run")) + if dry_run: + ctx.ui.ack("[muted]dry run — nothing was changed.[/muted]") + else: + for idx, name, secret in plan.writes: + await write_channel(ctx, device, idx, name, secret) + kept = list(plan.present) + [entry for _, entry in plan.added] + apply_preferences(kept, ctx.region_store, ctx.mute_store) + if plan.skipped: + names = ", ".join(entry.name for entry in plan.skipped) + ctx.ui.ack(f"[warn]no free slot for {names}: clear a channel first[/warn]") + return ToolResult( + summary={ + "added": len(plan.added), + "moved": plan.moved, + "skipped": len(plan.skipped), + "dry_run": dry_run, + }, + report=_imported(plan, dry_run), + ) + @staticmethod async def _show_qr(ctx: AppContext, name: str, url: str) -> None: """Draw the share link of a channel as a QR code, only in the menu. @@ -297,13 +389,39 @@ class ChannelsTool(Tool): self, {"cli_action": "join", "index": index, "name": name, "secret": secret} ) - @channels_app.command("import", help="Import a meshcore:// channel link") + @channels_app.command( + "import", help="Import a meshcore:// link, or a file from 'channels export'" + ) def _import_cmd( - index: int = typer.Argument(..., help="Channel slot index"), - url: str = typer.Argument(..., help="A meshcore://channel/add link"), + index: int | None = typer.Argument(None, help="Channel slot index (with a link)"), + url: str | None = typer.Argument(None, help="A meshcore://channel/add link"), + file: Path | None = typer.Option( + None, "--file", help="A file from 'channels export': add its channels" + ), + dry_run: bool = typer.Option( + False, "--dry-run", help="With --file: show the plan, change nothing" + ), ) -> None: + # One verb, two sources: a link brings one channel into a slot that the caller + # names, and a file brings many channels in the order of the file. + if file is not None: + if index is not None or url is not None: + raise typer.BadParameter("Pass a slot and a link, or --file, not both.") + params = {"cli_action": "import_file", "file": file, "dry_run": dry_run} + run_tool_command(self, params) + return + if index is None or url is None: + raise typer.BadParameter("Pass a slot index and a meshcore:// link, or --file.") + if dry_run: + raise typer.BadParameter("--dry-run works only with --file.") run_tool_command(self, {"cli_action": "import", "index": index, "url": url}) + @channels_app.command("export", help="Write every channel to a file, for another device") + def _export_cmd( + path: Path = typer.Argument(..., help="Destination file"), + ) -> None: + run_tool_command(self, {"cli_action": "export", "path": path}) + @channels_app.command("share", help="Print a channel's share link and QR code") def _share_cmd( index: int = typer.Argument(..., help="Channel slot index"), @@ -353,6 +471,84 @@ class ChannelsTool(Tool): app.add_typer(channels_app, name=self.name) +async def _complete_slots(ctx: AppContext) -> tuple[Any, list]: + """The device and all its channels, from a probe that read each slot. + + An export from a probe that stopped early leaves channels out of the file, and an + import plans over a slot that looks free and is not. Thus a short probe is a device + failure here (exit 4), and the caller can try again. + + Raises: + DeviceCommandError: If the probe did not read each slot. + """ + from ..core.channel_probe import probe_channel_slots + from ..core.connection import DeviceCommandError + + device = await ctx.device() + slots, complete = await probe_channel_slots(device) + if not complete: + raise DeviceCommandError("could not read every channel slot; nothing was changed") + return device, slots + + +def _exported(path: Path | None, channels: int, private: int) -> Facts: + """What ``channels export`` wrote. The plain face prints only the path.""" + from ..ui import fields + from ..ui.report import BARE, Facts + + return Facts( + key="export", + fields=( + fields.word("path", "path"), + fields.integer("channels", "channels"), + fields.integer("private", "private"), + ), + values={"path": str(path) if path else None, "channels": channels, "private": private}, + shape=BARE, + bare="path", + ) + + +def _imported(plan: Any, dry_run: bool) -> tuple[Facts, Listing]: + """What ``channels import --file`` did or plans to do: the counts, and the layout. + + The listing is the device after the import, in slot order. Each row names the action: + ``add`` (the channel is new), ``move`` (it changes slot), or ``keep``. A channel of + the file that has no free slot comes last, with no slot and the action ``skip``. + """ + from ..ui import fields + from ..ui.report import SILENT, Facts, Listing + + rows = [{"slot": slot, "name": name, "action": action} for slot, name, action in plan.layout] + rows += [{"slot": None, "name": entry.name, "action": "skip"} for entry in plan.skipped] + facts = Facts( + key="import", + fields=( + fields.flag("dry_run", "dry_run"), + fields.integer("added", "added"), + fields.integer("moved", "moved"), + fields.integer("skipped", "skipped"), + ), + values={ + "dry_run": dry_run, + "added": len(plan.added), + "moved": plan.moved, + "skipped": len(plan.skipped), + }, + shape=SILENT, + ) + listing = Listing( + key="channels", + columns=( + fields.integer("slot", "SLOT"), + fields.name("name", "NAME"), + fields.word("action", "ACTION"), + ), + rows=rows, + ) + return facts, listing + + def _channel_listing(slots: list, store: object | None = None) -> Listing: """The configured slots, one record for each slot. diff --git a/meshterm/ui/channels.py b/meshterm/ui/channels.py index 8f395c8..e5bf2e0 100644 --- a/meshterm/ui/channels.py +++ b/meshterm/ui/channels.py @@ -10,6 +10,11 @@ comes from the name), joining with a key that the user pastes, or importing a sc ``meshcore://`` link. Each change is written to the device at once, as in a phone app. Thus the list always shows the state of the radio. +The section "Export and import" moves all the channels to another device. The export writes +each channel, with its key, its send scope, and its mute, to one TOML file. The import adds +the channels of such a file to the device, in the order of the file, and it never removes a +channel (refer to :mod:`meshterm.core.channel_file`). + Muting is the one setting of a channel that is a local *preference* and not a configuration of the device. The new messages of a muted channel do not make the unread badge higher. Its inbound messages do not add to the unread count, and muting sets the count to zero. @@ -51,13 +56,23 @@ from __future__ import annotations import time from collections.abc import Awaitable, Callable from functools import lru_cache +from pathlib import Path from typing import TYPE_CHECKING from rich.cells import cell_len from rich.console import Group from rich.text import Text -from ..core.channel_probe import ChannelSlot +from ..core.channel_file import ( + ChannelFileError, + ImportPlan, + apply_preferences, + entry_from_slot, + plan_import, + read_channel_file, + write_channel_file, +) +from ..core.channel_probe import ChannelSlot, probe_channel_slots from ..core.channels import ( DEFAULT_PUBLIC_SECRET, MAX_CHANNELS, @@ -104,6 +119,8 @@ _DEFAULT_PUBLIC = "__default_public__" _JOIN = "__join__" _IMPORT = "__import__" _REORDER = "__reorder__" +_EXPORT_FILE = "__export_file__" +_IMPORT_FILE = "__import_file__" # The sentinels of the actions of the channel detail screen. _QR = "qr" @@ -222,6 +239,10 @@ async def manage_channels(ctx: AppContext) -> int: changes += await _import_link(ctx, device, slots, capacity) elif choice == _REORDER: changes += await _reorder_channels(ctx, device, slots) + elif choice == _EXPORT_FILE: + await _export_channels(ctx, slots) # it writes a file, and the device stays the same + elif choice == _IMPORT_FILE: + changes += await _import_channels(ctx, device, capacity) else: # an existing slot index slot = next((s for s in slots if s.idx == choice), None) if slot is not None: @@ -746,54 +767,70 @@ def _menu_items( # cells (＋ ＃ 🌐 🔑 🔗), and not one column early. The Organize row once corrected this # by hand, with two spaces and a comment of four lines. The column is empty where the # platform draws no icons, and the labels use the cells. - lane = icon_lane(("↕", "🌐", "＋", "＃", "🔑", "🔗")) + lane = icon_lane(("↕", "💾", "📂", "🌐", "＋", "＃", "🔑", "🔗")) if len(slots) > 1: items.append(section_heading("Organize")) reorder = marked_label("↕", "Reorder channels", "", lane=lane) items.append(Choice(title=reorder, value=_REORDER)) + # Before "Add a channel", because that section ends early when each slot is full, and an + # import still has work then: it puts the channels in the order of its file. + transfer = [] + if slots: + export = marked_label("💾", "Export channels…", "", lane=lane) + transfer.append((export, "Save them all to a file", _EXPORT_FILE)) + import_file = marked_label("📂", "Import channels…", "", lane=lane) + transfer.append((import_file, "Add them from a file", _IMPORT_FILE)) + + adds = [] + if _next_free_slot(slots, capacity) is not None: + # The Public channel of the firmware has a fixed key. It has one well-known secret, so + # it is the same channel on each slot. Offer to restore it only if no slot has it + # already. + if not any(s.secret == DEFAULT_PUBLIC_SECRET for s in slots): + adds.append( + ( + marked_label("🌐", "Standard Public channel", "", lane=lane), + "MeshCore's built-in meshwide channel", + _DEFAULT_PUBLIC, + ) + ) + adds.extend( + [ + ( + marked_label("＋", "New private channel…", "", lane=lane), + "A fresh random key", + _CREATE, + ), + ( + marked_label("＃", "Public channel…", "", lane=lane), + "Key derived from its name", + _PUBLIC, + ), + ( + marked_label("🔑", "Join with a key…", "", lane=lane), + "Paste a channel's 32-hex key", + _JOIN, + ), + ( + marked_label("🔗", "Import a link…", "", lane=lane), + "Paste a meshcore:// share link", + _IMPORT, + ), + ] + ) + # One call for the rows of the two sections, so that their descriptions start in the same + # column. Two calls aligned each section to its own widest label, four cells apart. + aligned = menu_rows(transfer + adds) + items.append(section_heading("Export and import")) + items.extend(aligned[: len(transfer)]) items.append(section_heading("Add a channel")) - if _next_free_slot(slots, capacity) is None: + if adds: + items.extend(aligned[len(transfer) :]) + else: # The list already knows that there is no free slot. It says so here, and it does not # offer four rows that all end in the same refusal. items.append(Separator(" every slot is full — clear one first")) - return f"Channels · {len(slots)}/{capacity} slots", items - rows = [] - # The Public channel of the firmware has a fixed key. It has one well-known secret, so it - # is the same channel on each slot. Offer to restore it only if no slot has it already. - if not any(s.secret == DEFAULT_PUBLIC_SECRET for s in slots): - rows.append( - ( - marked_label("🌐", "Standard Public channel", "", lane=lane), - "MeshCore's built-in meshwide channel", - _DEFAULT_PUBLIC, - ) - ) - rows.extend( - [ - ( - marked_label("＋", "New private channel…", "", lane=lane), - "A fresh random key", - _CREATE, - ), - ( - marked_label("＃", "Public channel…", "", lane=lane), - "Key derived from its name", - _PUBLIC, - ), - ( - marked_label("🔑", "Join with a key…", "", lane=lane), - "Paste a channel's 32-hex key", - _JOIN, - ), - ( - marked_label("🔗", "Import a link…", "", lane=lane), - "Paste a meshcore:// share link", - _IMPORT, - ), - ] - ) - items.extend(menu_rows(rows)) # ``·`` chains a status atom. ``—`` introduces a subject, and the slot count is not the # subject of this screen (refer to Contacts, which reads "Contacts · 12 known"). @@ -1335,6 +1372,141 @@ async def _reorder_channels(ctx: AppContext, device: Device, slots: list[Channel return 0 +# --- export and import --------------------------------------------------------- + +#: The file that the export and import prompts offer. The user can type another path. +_CHANNEL_FILE = "meshterm-channels.toml" + + +async def _export_channels(ctx: AppContext, slots: list[ChannelSlot]) -> None: + """Write the channels of the device to a file, so that another device can import them. + + Each channel goes into the file with its key, its send scope, and its mute, in slot + order (refer to :mod:`meshterm.core.channel_file`). The export writes nothing to the + device, so the list does not change. Thus a dialog says what the export wrote and where, + because nothing else on the screen shows that the file exists. + """ + raw = await ctx.ui.path( + "Export channels", prompt="Write the channels to this file:", default=_CHANNEL_FILE + ) + if not raw: + return + path = Path(raw).expanduser().resolve() + entries = [ + entry_from_slot(slot, scope=_channel_scope(ctx, slot), muted=_is_muted(ctx, slot)) + for slot in slots + ] + try: + write_channel_file(path, entries) + except OSError as exc: + await _say(ctx, f"could not write {path} — {exc.strerror or exc}", "err") + return + private = sum(1 for slot in slots if not slot.is_public) + text = f"{_count(len(entries), 'channel')} written to {path}" + if private: + text += f". It holds the keys of {_count(private, 'private channel')}: keep it private." + await _say(ctx, text, "ok") + + +async def _import_channels(ctx: AppContext, device: Device, capacity: int) -> int: + """Add the channels of an exported file to the device, in the order of the file. + + The import reads all the slots again before it plans. The plan writes many slots, and a + probe that stopped early makes a slot that holds a channel look free (refer to + :func:`_pick_free_slot` for the same reason on a single add). Then a dialog shows the + plan, and nothing is written until the user selects Import. A file that has nothing new + for the device still gives its send scopes and mutes. + + Returns: + The number of slot writes that the import made. + """ + raw = await ctx.ui.path( + "Import channels", prompt="Read the channels from this file:", default=_CHANNEL_FILE + ) + if not raw: + return 0 + path = Path(raw).expanduser() + try: + entries = read_channel_file(path) + except ChannelFileError as exc: + await _say(ctx, str(exc), "err") + return 0 + async with ctx.ui.busy_dialog("reading channels…", title="Channels"): + slots, complete = await probe_channel_slots(device) + if not complete: + await _say(ctx, "could not read every slot — nothing was written", "err") + return 0 + plan = plan_import(entries, slots, capacity) + kept = list(plan.present) + [entry for _, entry in plan.added] + if not plan.changes_device: + apply_preferences(kept, ctx.region_store, ctx.mute_store) + if plan.skipped: + await _say(ctx, "every slot is full — clear a channel, then import again", "warn") + else: + text = f"this device already has the {_count(len(entries), 'channel')} of {path.name}" + await _say(ctx, text, "ok") + return 0 + choice = await ctx.ui.dialog( + _plan_text(path, entries, plan), + [("Cancel", None), ("Import", "import")], + title="Import channels", + default=1, + ) + if choice != "import": + return 0 + done = 0 + try: + # One card for the whole import, as for a reorder. The card of each write is nested + # in it, and the count shows that the import advances. + async with ctx.ui.busy_dialog("importing channels…", title="Channels") as busy: + for position, (idx, name, secret) in enumerate(plan.writes, start=1): + what = f"saving {name}" if name else f"clearing slot {idx}" + busy.message = f"{what} · {position}/{len(plan.writes)}" + await write_channel(ctx, device, idx, name, secret) + done += 1 + except Exception as exc: # noqa: BLE001 - reported here, where the half-done state is known + # The same case as a reorder that stops: the writes have no transaction, so the + # slots are between the two layouts. Say so at once. The list reads the slots again + # when this round ends. + ctx.log.debug("channels: import stopped partway: %s", exc) + await _say(ctx, f"import stopped partway — {exc}", "err") + return done + apply_preferences(kept, ctx.region_store, ctx.mute_store) + if plan.skipped: + names = ", ".join(entry.name for entry in plan.skipped) + await _say(ctx, f"no free slot for {names} — clear a channel, then import again", "warn") + return done + + +def _plan_text(path: Path, entries: list, plan: ImportPlan) -> Text: + """The body of the import dialog: what the import adds, moves, and cannot add. + + The last line says that nothing is removed. A user who imports onto a device that has + channels of its own must know that before Import, and not find it after. + """ + width = max((cell_len(entry.name) for _, entry in plan.added), default=0) + text = Text(f"{path.name} has {_count(len(entries), 'channel')}.\n") + for slot, entry in plan.added: + text.append(f" + {channel_glyph(entry.name, entry.key)} ") + text.append(entry.name.ljust(width), style="brand") + text.append(f" into slot {slot}\n", style="muted") + if plan.moved: + moved = _count(plan.moved, "channel") + text.append(f" ↕ {moved} on this device change slot, to match the file\n") + if plan.skipped: + names = ", ".join(entry.name for entry in plan.skipped) + text.append(f" ⚠ no free slot for {names}\n", style="warn") + if plan.present: + text.append(f" {_count(len(plan.present), 'channel')} already on this device\n") + text.append("Nothing is removed from the device.", style="muted") + return text + + +def _count(n: int, noun: str) -> str: + """``n`` and the noun, plural when ``n`` is not 1: ``1 channel``, ``3 channels``.""" + return f"{n} {noun}" if n == 1 else f"{n} {noun}s" + + # --- shared views ------------------------------------------------------------ @@ -1434,7 +1606,7 @@ async def _say(ctx: AppContext, text: str, style: str) -> None: the user. Thus it goes to a dialog and not into the buffer that the menu empties when the whole tool has finished. A failed import once stayed in that buffer. """ - mark = {"err": "✗", "warn": "⚠"}.get(style, "") + mark = {"err": "✗", "warn": "⚠", "ok": "✓"}.get(style, "") body = Text(f"{mark} {text}" if mark else text, style=style) session = getattr(ctx.ui, "session", None) if session is None: # the scripted CLI has no dialogs, so print the text diff --git a/tests/test_channel_file.py b/tests/test_channel_file.py new file mode 100644 index 0000000..97c42c6 --- /dev/null +++ b/tests/test_channel_file.py @@ -0,0 +1,367 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Tests for the channel file: the export, the import, and the plan that an import makes. + +The file logic and the plan need no hardware. The export and the import run against the +:class:`MockDevice` simulator, through the CLI tool and through the channel manager. Two +contexts with two configuration directories stand for two computers, or for the old +device and the new device. +""" + +from __future__ import annotations + +import io +from contextlib import asynccontextmanager +from pathlib import Path +from types import SimpleNamespace + +import pytest +from rich.console import Console + +from meshterm.context import AppContext +from meshterm.core.admin_store import AdminStore +from meshterm.core.channel_file import ( + ChannelEntry, + ChannelFileError, + apply_preferences, + plan_import, + read_channel_file, + write_channel_file, +) +from meshterm.core.channel_probe import ChannelSlot, read_channel_slots +from meshterm.core.channels import DEFAULT_PUBLIC_SECRET, channel_identity, derive_secret +from meshterm.core.config import Settings +from meshterm.core.device_store import DeviceStore +from meshterm.core.mute_store import MuteStore +from meshterm.core.region_store import RegionStore +from meshterm.persistence.repository import Repository +from meshterm.tools.channels import ChannelsTool +from meshterm.ui.channels import ( + _EXPORT_FILE, + _IMPORT_FILE, + _LiveStats, + _menu_items, + manage_channels, +) + +_FAMILY = bytes(range(16)) +_PUBLIC = ChannelEntry("Public", DEFAULT_PUBLIC_SECRET) +_BOTS = ChannelEntry("#bots") +_FAMILY_ENTRY = ChannelEntry("Family", _FAMILY, scope="yul", muted=True) + + +def _slot(idx: int, name: str, secret: bytes | None = None) -> ChannelSlot: + """A channel as the device reports it. A ``#`` channel comes back with its derived key.""" + return ChannelSlot(idx=idx, name=name, secret=secret or derive_secret(name)) + + +# -- the file ------------------------------------------------------------------------- + + +def test_a_channel_file_round_trips_names_keys_scopes_and_mutes(tmp_path: Path) -> None: + """The file gives back each channel, in order, with its key, its scope, and its mute.""" + entries = [_PUBLIC, _BOTS, _FAMILY_ENTRY] + path = write_channel_file(tmp_path / "channels.toml", entries) + + assert read_channel_file(path) == entries + text = path.read_text(encoding="utf-8") + assert text.startswith("# MeshTerm channels.") # a user who opens it finds what it is + assert text.count("secret =") == 2 # the # channel has none: the device derives its key + + +def test_a_config_backup_is_also_a_channel_file(tmp_path: Path) -> None: + """The ``[[channels]]`` table of ``config backup`` reads as a channel file. + + The backup writes an index and a secret for each channel. The reader does not use the + index, and it removes the secret of a ``#`` channel, which the device derives. + """ + from meshterm.core.config_io import backup_config + + channels = [ + {"channel_idx": 0, "channel_name": "#bots", "channel_secret": derive_secret("#bots")}, + {"channel_idx": 1, "channel_name": "Family", "channel_secret": _FAMILY}, + ] + path = backup_config(tmp_path / "backup.toml", {}, {}, channels) + + entries = read_channel_file(path) + assert [(e.name, e.secret) for e in entries] == [("#bots", None), ("Family", _FAMILY)] + + +@pytest.mark.parametrize( + ("body", "message"), + [ + ("", "has no [[channels]] table"), + ("[[channels]]\nsecret = 'aa'\n", "channel 1 has no name"), + ("[[channels]]\nname = 'Family'\n", "channel 1 (Family): a private channel needs"), + ("[[channels]]\nname = 'Family'\nsecret = 'abc'\n", "channel 1 (Family):"), + ("[[channels]]\nname = '#bots'\nscope = '*'\n", "channel 1 (#bots):"), + ("[[channels]]\nname = '#bots'\nmuted = 'yes'\n", "muted must be true or false"), + ("name = = 1\n", "is not a TOML file"), + ], + ids=["empty", "no-name", "no-secret", "bad-secret", "bad-scope", "bad-mute", "not-toml"], +) +def test_a_bad_channel_file_names_what_is_wrong(tmp_path: Path, body: str, message: str) -> None: + """A file that MeshTerm cannot use gives one error, which names the entry.""" + path = tmp_path / "channels.toml" + path.write_text(body, encoding="utf-8") + with pytest.raises(ChannelFileError, match=None) as caught: + read_channel_file(path) + assert message in str(caught.value) + + +def test_a_missing_file_is_a_channel_file_error(tmp_path: Path) -> None: + """A path with no file is an error that says so, not an OSError from deep inside.""" + with pytest.raises(ChannelFileError, match="no such file"): + read_channel_file(tmp_path / "absent.toml") + + +def test_a_second_copy_of_a_channel_is_read_one_time(tmp_path: Path) -> None: + """The same channel two times in a file is one channel, at its first position.""" + path = write_channel_file(tmp_path / "c.toml", [_BOTS, _PUBLIC, _BOTS]) + assert read_channel_file(path) == [_BOTS, _PUBLIC] + + +# -- the plan ------------------------------------------------------------------------- + + +def test_a_new_device_gets_the_channels_of_the_file_in_order() -> None: + """A new device ships with only Public. The import adds the other channels after it.""" + plan = plan_import( + [_PUBLIC, _BOTS, _FAMILY_ENTRY], [_slot(0, "Public", DEFAULT_PUBLIC_SECRET)], 8 + ) + + assert plan.writes == ((1, "#bots", None), (2, "Family", _FAMILY)) + assert plan.layout == ((0, "Public", "keep"), (1, "#bots", "add"), (2, "Family", "add")) + assert [slot for slot, _ in plan.added] == [1, 2] + assert plan.present == (_PUBLIC,) and plan.moved == 0 + + +def test_an_import_puts_the_file_first_and_removes_nothing() -> None: + """The channels of the file come first, the other channels of the device after them. + + The device has gaps (slots 0, 3, and 5). The channels move down to close them, and the + plan clears slot 5, whose channel moved down. Each channel that the device had is still + on it, once. + """ + device = [_slot(0, "#test"), _slot(3, "Public", DEFAULT_PUBLIC_SECRET), _slot(5, "#bots")] + plan = plan_import([_PUBLIC, _BOTS, _FAMILY_ENTRY], device, 8) + + assert plan.layout == ( + (0, "Public", "move"), + (1, "#bots", "move"), + (2, "Family", "add"), + (3, "#test", "move"), + ) + assert (5, "", None) in plan.writes # the second copy of #bots goes + after = {idx: name for idx, name, _ in plan.writes} + assert sorted(n for n in after.values() if n) == ["#bots", "#test", "Family", "Public"] + + +def test_a_full_device_skips_the_last_channels_of_the_file() -> None: + """When the free slots run out, the plan keeps the first new channels and skips the rest.""" + device = [_slot(0, "#a"), _slot(1, "#b")] + plan = plan_import([ChannelEntry("#c"), ChannelEntry("#d"), ChannelEntry("#e")], device, 3) + + assert [entry.name for _, entry in plan.added] == ["#c"] + assert [entry.name for entry in plan.skipped] == ["#d", "#e"] + + +def test_a_device_that_matches_the_file_needs_no_write() -> None: + """Nothing to add and nothing to move is a plan with no write.""" + device = [_slot(0, "Public", DEFAULT_PUBLIC_SECRET), _slot(1, "#bots")] + plan = plan_import([_PUBLIC, _BOTS], device, 8) + assert not plan.changes_device + assert plan.layout == ((0, "Public", "keep"), (1, "#bots", "keep")) + + +def test_the_file_sets_scopes_and_mutes_and_never_clears_them(tmp_path: Path) -> None: + """An import adds a scope and a mute. A channel with none in the file keeps its own.""" + scopes = RegionStore(tmp_path / "regions.json") + mutes = MuteStore(tmp_path / "mutes.json") + bots_id = channel_identity("#bots", derive_secret("#bots")) + scopes.set_channel_scope(bots_id, "yow") + + changed = apply_preferences([_BOTS, _FAMILY_ENTRY], scopes, mutes) + + family_id = channel_identity("Family", _FAMILY) + assert changed == 2 + assert scopes.channel_scope(family_id) == "yul" and mutes.is_muted(family_id) + assert scopes.channel_scope(bots_id) == "yow" # the file gives #bots no scope: it stays + + +# -- the CLI and the channel manager -------------------------------------------------- + + +def _ctx(config_dir: Path) -> AppContext: + """A context on the mock device, with its own configuration directory and stores.""" + settings = Settings(config_dir=config_dir, db_path=config_dir / "chan.db") + config_dir.mkdir(parents=True, exist_ok=True) + return AppContext( + console=Console(file=io.StringIO()), + settings=settings, + repo=Repository(settings.db_path), + device_store=DeviceStore(config_dir / "devices.json"), + admin_store=AdminStore(config_dir / "admin.json"), + mock=True, + ) + + +async def _old_device(ctx: AppContext) -> None: + """Give the device three channels, a scope on Family, and a mute on #bots.""" + device = await ctx.device() + await device.set_channel(0, "Public", DEFAULT_PUBLIC_SECRET) + await device.set_channel(1, "#bots", None) + await device.set_channel(2, "Family", _FAMILY) + ctx.region_store.set_channel_scope(channel_identity("Family", _FAMILY), "yul") + ctx.mute_store.set_muted(channel_identity("#bots", derive_secret("#bots")), True) + + +async def test_cli_export_then_import_sets_up_a_new_device(tmp_path: Path) -> None: + """``export`` on the old device and ``import --file`` on the new one give the same channels. + + The new context has its own stores, as another computer does. Thus the scope and the + mute that it ends with came from the file. + """ + tool = ChannelsTool() + old, new = _ctx(tmp_path / "old"), _ctx(tmp_path / "new") + try: + await _old_device(old) + path = tmp_path / "channels.toml" + result = await tool.run(old, {"cli_action": "export", "path": path}) + assert result.summary == {"channels": 3} and path.exists() + + await (await new.device()).set_channel(0, "Public", DEFAULT_PUBLIC_SECRET) # as shipped + params = {"cli_action": "import_file", "file": path, "dry_run": False} + result = await tool.run(new, params) + + assert result.summary == {"added": 2, "moved": 0, "skipped": 0, "dry_run": False} + slots = await read_channel_slots(await new.device()) + assert [s.name for s in slots] == ["Public", "#bots", "Family"] + assert slots[2].secret == _FAMILY + assert new.region_store.channel_scope(slots[2].identity) == "yul" + assert new.mute_store.is_muted(slots[1].identity) + finally: + old.repo.close() + new.repo.close() + + +async def test_cli_import_dry_run_writes_nothing(tmp_path: Path) -> None: + """``--dry-run`` gives the plan and leaves the device and the stores as they are.""" + tool = ChannelsTool() + ctx = _ctx(tmp_path / "home") + try: + path = write_channel_file(tmp_path / "c.toml", [_BOTS, _FAMILY_ENTRY]) + params = {"cli_action": "import_file", "file": path, "dry_run": True} + result = await tool.run(ctx, params) + + assert result.summary == {"added": 2, "moved": 0, "skipped": 0, "dry_run": True} + assert await read_channel_slots(await ctx.device()) == [] + assert not ctx.mute_store.is_muted(_FAMILY_ENTRY.identity) + finally: + ctx.repo.close() + + +async def test_cli_export_of_a_device_with_no_channels_writes_no_file(tmp_path: Path) -> None: + """Nothing to export is exit 5, and no file that holds nothing.""" + from meshterm.core import exitcodes + + ctx = _ctx(tmp_path / "home") + try: + path = tmp_path / "c.toml" + result = await ChannelsTool().run(ctx, {"cli_action": "export", "path": path}) + assert result.exit_code == exitcodes.NO_RESULT and not path.exists() + finally: + ctx.repo.close() + + +async def test_cli_import_of_a_bad_file_is_a_usage_error(tmp_path: Path) -> None: + """A file that is not valid is a bad argument (exit 2). Nothing is sent to the device.""" + import typer + + ctx = _ctx(tmp_path / "home") + try: + path = tmp_path / "c.toml" + path.write_text("[[channels]]\nname = 'Family'\n", encoding="utf-8") + params = {"cli_action": "import_file", "file": path, "dry_run": False} + with pytest.raises(typer.BadParameter, match="needs its secret"): + await ChannelsTool().run(ctx, params) + finally: + ctx.repo.close() + + +class _ScriptedUi: + """A UI that replays scripted answers, to drive the channel manager with no screen. + + It records the prompt of each dialog and each line that the manager says, so that a + test can read what the user saw. + """ + + def __init__(self, selects: list, paths: list, dialogs: list) -> None: + self._selects, self._paths, self._dialogs = list(selects), list(paths), list(dialogs) + self.prompts: list[str] = [] + self.notes: list[str] = [] + + def show(self, *renderables) -> None: # noqa: ANN002 + pass + + @asynccontextmanager + async def busy_dialog(self, message: str = "", *, title: str = ""): # noqa: ANN201 + yield SimpleNamespace(message=message) + + def note(self, markup: str) -> None: + self.notes.append(markup) + + def ack(self, markup: str) -> None: + pass + + async def select(self, title: str, items: list, *, default=None): # noqa: ANN001, ANN201 + return self._selects.pop(0) + + async def path(self, title: str, *, prompt: str = "", default: str = "") -> str | None: + return self._paths.pop(0) + + async def dialog(self, prompt, buttons, **_kw): # noqa: ANN001, ANN003, ANN201 + self.prompts.append(getattr(prompt, "plain", str(prompt))) + return self._dialogs.pop(0) + + +async def test_the_manager_exports_and_imports_through_its_rows(tmp_path: Path) -> None: + """The two rows of "Export and import" move the channels to another device. + + The import shows its plan first, and the plan says that nothing is removed. + """ + old, new = _ctx(tmp_path / "old"), _ctx(tmp_path / "new") + path = tmp_path / "channels.toml" + try: + await _old_device(old) + old.ui = _ScriptedUi(selects=[_EXPORT_FILE, None], paths=[str(path)], dialogs=[]) + await manage_channels(old) + assert [e.name for e in read_channel_file(path)] == ["Public", "#bots", "Family"] + assert "keys of 1 private channel" in old.ui.notes[-1] + + new.ui = ui = _ScriptedUi( + selects=[_IMPORT_FILE, None], paths=[str(path)], dialogs=["import"] + ) + await manage_channels(new) + + slots = await read_channel_slots(await new.device()) + assert [s.name for s in slots] == ["Public", "#bots", "Family"] + assert "into slot 2" in ui.prompts[0] and "Nothing is removed" in ui.prompts[0] + assert new.region_store.channel_scope(slots[2].identity) == "yul" + finally: + old.repo.close() + new.repo.close() + + +async def test_export_is_offered_only_when_there_is_a_channel(tmp_path: Path) -> None: + """With no channel, there is nothing to export. Import is always offered.""" + ctx = _ctx(tmp_path / "home") + try: + + def values(slots: list) -> list: + _, items = _menu_items(ctx, slots, 8, _LiveStats(ctx)) + return [getattr(item, "value", None) for item in items] + + assert _EXPORT_FILE not in values([]) and _IMPORT_FILE in values([]) + assert _EXPORT_FILE in values([_slot(0, "#bots")]) + finally: + ctx.repo.close()